This is an automated email from the ASF dual-hosted git repository. davsclaus pushed a commit to branch fix/CAMEL-23994-pollonerror-seek in repository https://gitbox.apache.org/repos/asf/camel.git
commit 1c8e0ddb54d4581c86d079bc53b86c2160127414 Author: Claus Ibsen <[email protected]> AuthorDate: Tue Jul 14 16:44:06 2026 +0200 CAMEL-23994: Fix pollOnError seek skipping messages on all partitions When an exception was caught in the poll loop, SeekUtil.seekToNextOffset was called with a hardcoded offset of -1, causing it to seek position+1 on ALL assigned partitions. This silently dropped one unread message per partition, even for partitions unrelated to the error. Now SeekUtil uses the exception itself: for RecordDeserializationException (poison message), it seeks only the affected partition to the failing offset+1. For other exceptions (transient errors, commit failures), no positions are advanced since the error may not be record-related. Co-Authored-By: Claude Opus 4.6 <[email protected]> Signed-off-by: Claus Ibsen <[email protected]> --- .../consumer/errorhandler/BridgeErrorStrategy.java | 2 +- .../errorhandler/DiscardErrorStrategy.java | 2 +- .../kafka/consumer/errorhandler/SeekUtil.java | 40 ++++++---------------- 3 files changed, 13 insertions(+), 31 deletions(-) diff --git a/components/camel-kafka/src/main/java/org/apache/camel/component/kafka/consumer/errorhandler/BridgeErrorStrategy.java b/components/camel-kafka/src/main/java/org/apache/camel/component/kafka/consumer/errorhandler/BridgeErrorStrategy.java index 098d8d10a362..52d75be21343 100644 --- a/components/camel-kafka/src/main/java/org/apache/camel/component/kafka/consumer/errorhandler/BridgeErrorStrategy.java +++ b/components/camel-kafka/src/main/java/org/apache/camel/component/kafka/consumer/errorhandler/BridgeErrorStrategy.java @@ -48,7 +48,7 @@ public class BridgeErrorStrategy implements PollExceptionStrategy { // use bridge error handler to route with exception recordFetcher.getBridge().handleException(exception); // skip this poison message and seek to the next message - SeekUtil.seekToNextOffset(consumer, partitionLastOffset); + SeekUtil.seekToNextOffset(consumer, exception); if (exception instanceof AuthenticationException || exception instanceof AuthorizationException) { continueFlag = false; diff --git a/components/camel-kafka/src/main/java/org/apache/camel/component/kafka/consumer/errorhandler/DiscardErrorStrategy.java b/components/camel-kafka/src/main/java/org/apache/camel/component/kafka/consumer/errorhandler/DiscardErrorStrategy.java index 46c61b92d9ee..875ba83b394f 100644 --- a/components/camel-kafka/src/main/java/org/apache/camel/component/kafka/consumer/errorhandler/DiscardErrorStrategy.java +++ b/components/camel-kafka/src/main/java/org/apache/camel/component/kafka/consumer/errorhandler/DiscardErrorStrategy.java @@ -40,6 +40,6 @@ public class DiscardErrorStrategy implements PollExceptionStrategy { LOG.warn("Requesting the consumer to discard the message and continue to the next based on polling exception strategy"); // skip this poison message and seek to the next message - SeekUtil.seekToNextOffset(consumer, partitionLastOffset); + SeekUtil.seekToNextOffset(consumer, exception); } } diff --git a/components/camel-kafka/src/main/java/org/apache/camel/component/kafka/consumer/errorhandler/SeekUtil.java b/components/camel-kafka/src/main/java/org/apache/camel/component/kafka/consumer/errorhandler/SeekUtil.java index b3c6e77adc12..5ed7021cae0c 100644 --- a/components/camel-kafka/src/main/java/org/apache/camel/component/kafka/consumer/errorhandler/SeekUtil.java +++ b/components/camel-kafka/src/main/java/org/apache/camel/component/kafka/consumer/errorhandler/SeekUtil.java @@ -17,10 +17,9 @@ package org.apache.camel.component.kafka.consumer.errorhandler; -import java.util.Set; - import org.apache.kafka.clients.consumer.Consumer; import org.apache.kafka.common.TopicPartition; +import org.apache.kafka.common.errors.RecordDeserializationException; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -31,33 +30,16 @@ final class SeekUtil { } - public static void seekToNextOffset(Consumer<?, ?> consumer, long partitionLastOffset) { - boolean logged = false; - Set<TopicPartition> tps = consumer.assignment(); - if (tps != null && partitionLastOffset != -1) { - long next = partitionLastOffset + 1; - - for (TopicPartition tp : tps) { - if (LOG.isInfoEnabled()) { - LOG.info( - "Consumer seeking to next offset {} to continue polling next message from topic {} on partition {}", - next, tp.topic(), tp.partition()); - } - - consumer.seek(tp, next); - } - } else if (tps != null) { - for (TopicPartition tp : tps) { - long next = consumer.position(tp) + 1; - if (!logged) { - LOG.info( - "Consumer seeking to next offset {} to continue polling next message from topic {} on partition {}", - next, - tp.topic(), tp.partition()); - logged = true; - } - consumer.seek(tp, next); - } + public static void seekToNextOffset(Consumer<?, ?> consumer, Exception exception) { + if (exception instanceof RecordDeserializationException rde) { + TopicPartition tp = rde.topicPartition(); + long next = rde.offset() + 1; + LOG.info("Consumer seeking to next offset {} to skip poison message on topic {} partition {}", + next, tp.topic(), tp.partition()); + consumer.seek(tp, next); + } else { + LOG.warn("Non-record exception caught during poll, not advancing any partition offsets: {}", + exception.getMessage()); } } }
