Retry if there are any not-leader errors in test

Signed-off-by: Robert Young <robertyoungnz@gmail.com>

Author: robobario

kroxylicious-integration-tests/src/test/java/io/kroxylicious/proxy/filter/authorization/OffsetForLeaderEpochAuthzIT.java

@@ -24,6 +24,7 @@
 import org.apache.kafka.common.message.OffsetForLeaderEpochRequestData.OffsetForLeaderTopic;
 import org.apache.kafka.common.message.OffsetForLeaderEpochResponseData;
 import org.apache.kafka.common.protocol.ApiKeys;
+import org.apache.kafka.common.protocol.Errors;
 import org.apache.kafka.common.resource.PatternType;
 import org.apache.kafka.common.resource.ResourcePattern;
 import org.apache.kafka.common.resource.ResourceType;
@@ -152,6 +153,12 @@ public OffsetForLeaderEpochRequestData requestData(String user, BaseClusterFixtu
             offsetForLeaderEpochRequestData.topics().addAll(List.of(topicA, topicB, topicC));
             return offsetForLeaderEpochRequestData;
         }
+
+        @Override
+        public boolean needsRetry(OffsetForLeaderEpochResponseData response) {
+            return response.topics() != null && response.topics().stream().anyMatch(topicResult -> topicResult.partitions() != null
+                    && topicResult.partitions().stream().anyMatch(epochEndOffset -> Errors.forCode(epochEndOffset.errorCode()) == Errors.NOT_LEADER_OR_FOLLOWER));
+        }
     }
 
     private static OffsetForLeaderTopic createOffsetForLeaderTopic(String topicName, int... partitions) {