CAMEL-17424: do handle Authorization/Authentication issues when failling to poll

Author: orpiske

components/camel-kafka/src/main/java/org/apache/camel/component/kafka/consumer/errorhandler/BridgeErrorStrategy.java

@@ -20,13 +20,16 @@
 import org.apache.camel.component.kafka.KafkaFetchRecords;
 import org.apache.camel.component.kafka.PollExceptionStrategy;
 import org.apache.kafka.clients.consumer.Consumer;
+import org.apache.kafka.common.errors.AuthenticationException;
+import org.apache.kafka.common.errors.AuthorizationException;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
 public class BridgeErrorStrategy implements PollExceptionStrategy {
     private static final Logger LOG = LoggerFactory.getLogger(BridgeErrorStrategy.class);
     private final KafkaFetchRecords recordFetcher;
     private final Consumer<?, ?> consumer;
+    private boolean continueFlag = true; // whether to continue polling or not
 
     public BridgeErrorStrategy(KafkaFetchRecords recordFetcher, Consumer<?, ?> consumer) {
         this.recordFetcher = recordFetcher;
@@ -35,7 +38,7 @@ public BridgeErrorStrategy(KafkaFetchRecords recordFetcher, Consumer<?, ?> consu
 
     @Override
     public boolean canContinue() {
-        return true;
+        return continueFlag;
     }
 
     @Override
@@ -46,5 +49,9 @@ public void handle(long partitionLastOffset, Exception exception) {
         recordFetcher.getBridge().handleException(exception);
         // skip this poison message and seek to next message
         SeekUtil.seekToNextOffset(consumer, partitionLastOffset);
+
+        if (exception instanceof AuthenticationException || exception instanceof AuthorizationException) {
+            continueFlag = false;
+        }
     }
 }