This is an automated email from the ASF dual-hosted git repository.
davsclaus pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/camel.git
The following commit(s) were added to refs/heads/main by this push:
new d160a2edcd14 CAMEL-23994: Fix pollOnError seek skipping messages on
all partitions
d160a2edcd14 is described below
commit d160a2edcd14faa5045d8ba687bc5a9f7403e98d
Author: Claus Ibsen <[email protected]>
AuthorDate: Tue Jul 14 18:15:03 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.
Closes #24687
Co-Authored-By: Claude Opus 4.6 <[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());
}
}
}
