You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@camel.apache.org by da...@apache.org on 2026/10/19 14:31:42 UTC
(camel) branch main updated: CAMEL-25756: camel-kafka - a failed poll must not drop the records of an incomplete batch (#28071)
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 55339dd9c435 CAMEL-25756: camel-kafka - a failed poll must not drop the records of an incomplete batch (#28071)
55339dd9c435 is described below
commit 55339dd9c435599c58be0c704439b1ab0baff46c
Author: allthingssecurity <49...@users.noreply.github.com>
AuthorDate: Mon Oct 19 20:01:35 2026 +0530
CAMEL-25756: camel-kafka - a failed poll must not drop the records of an incomplete batch (#28071)
Co-authored-by: Claus Ibsen <cl...@gmail.com>
Co-authored-by: Claude Opus 5.5 <no...@anthropic.com>
---
.../camel/component/kafka/KafkaFetchRecords.java | 9 +-
.../batching/KafkaRecordBatchingProcessor.java | 19 ++-
.../kafka/KafkaBatchingConsumerPollErrorTest.java | 188 +++++++++++++++++++++
.../KafkaBatchingConsumerProcessErrorTest.java | 146 ++++++++++++++++
4 files changed, 355 insertions(+), 7 deletions(-)
diff --git a/components/camel-kafka/src/main/java/org/apache/camel/component/kafka/KafkaFetchRecords.java b/components/camel-kafka/src/main/java/org/apache/camel/component/kafka/KafkaFetchRecords.java
index 23f09a65f9da..63084b2f79cc 100644
--- a/components/camel-kafka/src/main/java/org/apache/camel/component/kafka/KafkaFetchRecords.java
+++ b/components/camel-kafka/src/main/java/org/apache/camel/component/kafka/KafkaFetchRecords.java
@@ -93,6 +93,9 @@ public class KafkaFetchRecords implements Runnable {
// whether the fetcher thread is in poll, which stop() may then wake up (CAMEL-25753)
private volatile boolean polling;
private CommitManager commitManager;
+ // kept while the Kafka consumer is connected: the batching processor holds the records of earlier polls until its
+ // batch is complete, and the consumer does not return them again
+ private KafkaRecordProcessorFacade recordProcessorFacade;
private volatile Exception lastError;
private final KafkaConsumerListener consumerListener;
@@ -163,6 +166,9 @@ public class KafkaFetchRecords implements Runnable {
}
setConnected(true);
+ // a new Kafka consumer starts from the committed offsets: the records held for an incomplete batch
+ // of the previous consumer are polled again
+ recordProcessorFacade = createRecordProcessor();
// a new Kafka consumer is not paused: pause it again if the consumer is suspended
state.compareAndSet(State.PAUSED, State.PAUSE_REQUESTED);
}
@@ -326,7 +332,8 @@ public class KafkaFetchRecords implements Runnable {
LOG.trace("Polling {} from {} with timeout: {}", threadId, getPrintableTopic(), pollTimeoutMs);
}
- final KafkaRecordProcessorFacade recordProcessorFacade = createRecordProcessor();
+ // the processor of the connection is used again after a failed poll (pollOnError ERROR_HANDLER, DISCARD
+ // or RETRY): the records it holds were returned by earlier polls and must still be processed
while (isKafkaConsumerRunnableAndNotStopped() && isConnected() && pollExceptionStrategy.canContinue()) {
diff --git a/components/camel-kafka/src/main/java/org/apache/camel/component/kafka/consumer/support/batching/KafkaRecordBatchingProcessor.java b/components/camel-kafka/src/main/java/org/apache/camel/component/kafka/consumer/support/batching/KafkaRecordBatchingProcessor.java
index 8efa0eeeea7b..bbf2875adbef 100644
--- a/components/camel-kafka/src/main/java/org/apache/camel/component/kafka/consumer/support/batching/KafkaRecordBatchingProcessor.java
+++ b/components/camel-kafka/src/main/java/org/apache/camel/component/kafka/consumer/support/batching/KafkaRecordBatchingProcessor.java
@@ -153,9 +153,7 @@ final class KafkaRecordBatchingProcessor extends KafkaRecordProcessor {
exchangeList.size());
// poll timeout has elapsed, so check for expired records
- ProcessingResult result = processBatch(camelKafkaConsumer);
- exchangeList.clear();
- timeoutWatch.takenAndRestart(); // restart timer after processing expired batch
+ ProcessingResult result = processAndClearBatch(camelKafkaConsumer);
// if breakOnFirstError was hit, return immediately
if (result.isBreakOnErrorHit()) {
@@ -171,9 +169,7 @@ final class KafkaRecordBatchingProcessor extends KafkaRecordProcessor {
exchangeList.add(childExchange);
if (exchangeList.size() >= configuration.getMaxPollRecords()) {
- ProcessingResult result = processBatch(camelKafkaConsumer);
- exchangeList.clear();
- timeoutWatch.takenAndRestart(); // restart timer after batch processed
+ ProcessingResult result = processAndClearBatch(camelKafkaConsumer);
// if breakOnFirstError was hit, return immediately
if (result.isBreakOnErrorHit()) {
@@ -251,6 +247,17 @@ final class KafkaRecordBatchingProcessor extends KafkaRecordProcessor {
return offsets;
}
+ private ProcessingResult processAndClearBatch(KafkaConsumer camelKafkaConsumer) {
+ try {
+ return processBatch(camelKafkaConsumer);
+ } finally {
+ // clear the batch even when processing it throws: the processor is kept while the Kafka consumer is
+ // connected, and a queue that fills up refuses every record of the next polls
+ exchangeList.clear();
+ timeoutWatch.takenAndRestart(); // restart timer after batch processed
+ }
+ }
+
private ProcessingResult processBatch(KafkaConsumer camelKafkaConsumer) {
intervalWatch.restart();
diff --git a/components/camel-kafka/src/test/java/org/apache/camel/component/kafka/KafkaBatchingConsumerPollErrorTest.java b/components/camel-kafka/src/test/java/org/apache/camel/component/kafka/KafkaBatchingConsumerPollErrorTest.java
new file mode 100644
index 000000000000..79941f8f92e1
--- /dev/null
+++ b/components/camel-kafka/src/test/java/org/apache/camel/component/kafka/KafkaBatchingConsumerPollErrorTest.java
@@ -0,0 +1,188 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.camel.component.kafka;
+
+import java.time.Duration;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+import java.util.Properties;
+import java.util.Set;
+import java.util.concurrent.CopyOnWriteArrayList;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.concurrent.locks.LockSupport;
+
+import org.apache.camel.BindToRegistry;
+import org.apache.camel.Exchange;
+import org.apache.camel.builder.RouteBuilder;
+import org.apache.camel.test.junit6.CamelTestSupport;
+import org.apache.kafka.clients.consumer.Consumer;
+import org.apache.kafka.clients.consumer.ConsumerRecord;
+import org.apache.kafka.clients.consumer.ConsumerRecords;
+import org.apache.kafka.clients.consumer.MockConsumer;
+import org.apache.kafka.clients.consumer.OffsetAndMetadata;
+import org.apache.kafka.clients.producer.Producer;
+import org.apache.kafka.common.KafkaException;
+import org.apache.kafka.common.TopicPartition;
+import org.apache.kafka.common.errors.RecordDeserializationException;
+import org.apache.kafka.common.errors.RecordDeserializationException.DeserializationExceptionOrigin;
+import org.apache.kafka.common.header.internals.RecordHeaders;
+import org.apache.kafka.common.record.TimestampType;
+import org.junit.jupiter.api.Test;
+
+import static org.awaitility.Awaitility.await;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+
+/**
+ * A batching consumer holds the records of earlier polls until its batch is complete. A failed poll (handled by the
+ * default pollOnError=ERROR_HANDLER) must not drop them: the consumer does not return them again, and the next batch
+ * commits past them. Uses a {@link MockConsumer} instead of a broker.
+ */
+class KafkaBatchingConsumerPollErrorTest extends CamelTestSupport {
+
+ private static final String TOPIC = "batching-poll-error";
+ private static final TopicPartition PARTITION = new TopicPartition(TOPIC, 0);
+
+ private final PollCountingMockConsumer mockConsumer = new PollCountingMockConsumer();
+ private final List<String> routed = new CopyOnWriteArrayList<>();
+ private long offset;
+
+ @BindToRegistry("mockFactory")
+ private final KafkaClientFactory factory = new KafkaClientFactory() {
+ @Override
+ public Producer getProducer(Properties kafkaProps) {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public Consumer getConsumer(Properties kafkaProps) {
+ return mockConsumer;
+ }
+
+ @Override
+ public String getBrokers(KafkaConfiguration configuration) {
+ return "localhost:9092";
+ }
+ };
+
+ @Override
+ protected RouteBuilder createRouteBuilder() {
+ return new RouteBuilder() {
+ @Override
+ public void configure() {
+ // a batch of 10 records; the poll timeout is long enough that an incomplete batch is never dispatched
+ // during the test
+ from("kafka:" + TOPIC + "?brokers=localhost:9092&kafkaClientFactory=#mockFactory&groupId=test"
+ + "&batching=true&maxPollRecords=10&pollTimeoutMs=60000")
+ .routeId("kafka")
+ .process(exchange -> {
+ for (Object record : exchange.getMessage().getBody(List.class)) {
+ routed.add(((Exchange) record).getMessage().getBody(String.class));
+ }
+ });
+ }
+ };
+ }
+
+ @Override
+ protected void setupResources() {
+ mockConsumer.updateBeginningOffsets(Map.of(PARTITION, 0L));
+ mockConsumer.schedulePollTask(() -> mockConsumer.rebalance(List.of(PARTITION)));
+ }
+
+ @Test
+ void testFailedPollKeepsTheRecordsOfTheIncompleteBatch() {
+ addRecords("a", 3);
+ // the three records are polled and held for the batch of 10
+ await().atMost(20, TimeUnit.SECONDS).until(() -> mockConsumer.position(PARTITION) == 3);
+
+ mockConsumer.setPollException(new KafkaException("Simulated poll failure"));
+ await().atMost(20, TimeUnit.SECONDS).until(() -> mockConsumer.pollErrors.get() == 1);
+
+ addRecords("b", 10);
+ await().atMost(20, TimeUnit.SECONDS).until(() -> routed.contains("b6"));
+
+ assertEquals(List.of("a0", "a1", "a2", "b0", "b1", "b2", "b3", "b4", "b5", "b6"), routed);
+ assertEquals(10L, committedOffset());
+ }
+
+ @Test
+ void testPoisonRecordKeepsTheRecordsOfTheIncompleteBatch() {
+ addRecords("a", 3);
+ await().atMost(20, TimeUnit.SECONDS).until(() -> mockConsumer.position(PARTITION) == 3);
+
+ // the record at offset 3 cannot be deserialized (the poll fails on it): the error handler seeks past it
+ long poison = offset++;
+ mockConsumer.setPollException(
+ new RecordDeserializationException(
+ DeserializationExceptionOrigin.VALUE, PARTITION, poison, 0L, TimestampType.CREATE_TIME, null, null,
+ new RecordHeaders(), "Simulated deserialization failure", null));
+ await().atMost(20, TimeUnit.SECONDS).until(() -> mockConsumer.pollErrors.get() == 1);
+
+ addRecords("b", 10);
+ await().atMost(20, TimeUnit.SECONDS).until(() -> routed.contains("b6"));
+
+ assertEquals(List.of("a0", "a1", "a2", "b0", "b1", "b2", "b3", "b4", "b5", "b6"), routed);
+ assertEquals(11L, committedOffset());
+ }
+
+ private void addRecords(String prefix, int count) {
+ await().atMost(20, TimeUnit.SECONDS).until(() -> mockConsumer.assignment().contains(PARTITION));
+ List<ConsumerRecord<Object, Object>> records = new ArrayList<>();
+ for (int i = 0; i < count; i++) {
+ records.add(new ConsumerRecord<>(TOPIC, 0, offset++, null, prefix + i));
+ }
+ records.forEach(mockConsumer::addRecord);
+ }
+
+ private long committedOffset() {
+ await().atMost(20, TimeUnit.SECONDS)
+ .until(() -> mockConsumer.committed(Set.of(PARTITION)).get(PARTITION) != null);
+ OffsetAndMetadata committed = mockConsumer.committed(Set.of(PARTITION)).get(PARTITION);
+ return committed.offset();
+ }
+
+ private static final class PollCountingMockConsumer extends MockConsumer<Object, Object> {
+ final AtomicInteger pollErrors = new AtomicInteger();
+
+ PollCountingMockConsumer() {
+ super("earliest");
+ }
+
+ @Override
+ public synchronized ConsumerRecords<Object, Object> poll(Duration timeout) {
+ ConsumerRecords<Object, Object> records;
+ try {
+ records = super.poll(timeout);
+ } catch (KafkaException e) {
+ pollErrors.incrementAndGet();
+ throw e;
+ }
+ if (records.isEmpty()) {
+ // a real consumer waits up to the poll timeout when there is nothing to fetch
+ LockSupport.parkNanos(TimeUnit.MILLISECONDS.toNanos(10));
+ }
+ return records;
+ }
+
+ @Override
+ public synchronized void close() {
+ // keep the mock usable: a stopped route closes the consumer
+ }
+ }
+}
diff --git a/components/camel-kafka/src/test/java/org/apache/camel/component/kafka/KafkaBatchingConsumerProcessErrorTest.java b/components/camel-kafka/src/test/java/org/apache/camel/component/kafka/KafkaBatchingConsumerProcessErrorTest.java
new file mode 100644
index 000000000000..53a963929b55
--- /dev/null
+++ b/components/camel-kafka/src/test/java/org/apache/camel/component/kafka/KafkaBatchingConsumerProcessErrorTest.java
@@ -0,0 +1,146 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.camel.component.kafka;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+import java.util.Properties;
+import java.util.concurrent.CopyOnWriteArrayList;
+import java.util.concurrent.TimeUnit;
+
+import org.apache.camel.BindToRegistry;
+import org.apache.camel.Exchange;
+import org.apache.camel.builder.RouteBuilder;
+import org.apache.camel.spi.ExceptionHandler;
+import org.apache.camel.test.junit6.CamelTestSupport;
+import org.apache.kafka.clients.consumer.Consumer;
+import org.apache.kafka.clients.consumer.ConsumerRecord;
+import org.apache.kafka.clients.consumer.MockConsumer;
+import org.apache.kafka.clients.producer.Producer;
+import org.apache.kafka.common.TopicPartition;
+import org.junit.jupiter.api.Test;
+
+import static org.awaitility.Awaitility.await;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+
+/**
+ * A batch whose processing throws (here the exception handler of the consumer throws) must still be cleared: the batch
+ * is kept while the Kafka consumer is connected, and a full batch would refuse every record of the next polls. Uses a
+ * {@link MockConsumer} instead of a broker.
+ */
+class KafkaBatchingConsumerProcessErrorTest extends CamelTestSupport {
+
+ private static final String TOPIC = "batching-process-error";
+ private static final TopicPartition PARTITION = new TopicPartition(TOPIC, 0);
+
+ private final MockConsumer<Object, Object> mockConsumer = new MockConsumer<>("earliest") {
+ @Override
+ public synchronized void close() {
+ // keep the mock usable: a stopped route closes the consumer
+ }
+ };
+ private final List<String> routed = new CopyOnWriteArrayList<>();
+ private long offset;
+
+ @BindToRegistry("mockFactory")
+ private final KafkaClientFactory factory = new KafkaClientFactory() {
+ @Override
+ public Producer getProducer(Properties kafkaProps) {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public Consumer getConsumer(Properties kafkaProps) {
+ return mockConsumer;
+ }
+
+ @Override
+ public String getBrokers(KafkaConfiguration configuration) {
+ return "localhost:9092";
+ }
+ };
+
+ @BindToRegistry("throwingHandler")
+ private final ExceptionHandler throwingHandler = new ExceptionHandler() {
+ @Override
+ public void handleException(Throwable exception) {
+ throw new IllegalStateException("Simulated exception handler failure", exception);
+ }
+
+ @Override
+ public void handleException(String message, Throwable exception) {
+ handleException(exception);
+ }
+
+ @Override
+ public void handleException(String message, Exchange exchange, Throwable exception) {
+ handleException(exception);
+ }
+ };
+
+ @Override
+ protected RouteBuilder createRouteBuilder() {
+ return new RouteBuilder() {
+ @Override
+ public void configure() {
+ // a batch of 2 records; with manual commit a failed batch is passed to the exception handler, which throws
+ from("kafka:" + TOPIC + "?brokers=localhost:9092&kafkaClientFactory=#mockFactory&groupId=test"
+ + "&batching=true&maxPollRecords=2&pollTimeoutMs=60000&allowManualCommit=true"
+ + "&exceptionHandler=#throwingHandler")
+ .routeId("kafka")
+ .process(exchange -> {
+ boolean fail = false;
+ for (Object record : exchange.getMessage().getBody(List.class)) {
+ String body = ((Exchange) record).getMessage().getBody(String.class);
+ routed.add(body);
+ fail |= body.startsWith("a");
+ }
+ if (fail) {
+ throw new IllegalArgumentException("Simulated processing failure");
+ }
+ });
+ }
+ };
+ }
+
+ @Override
+ protected void setupResources() {
+ mockConsumer.updateBeginningOffsets(Map.of(PARTITION, 0L));
+ mockConsumer.schedulePollTask(() -> mockConsumer.rebalance(List.of(PARTITION)));
+ }
+
+ @Test
+ void testFailedBatchIsCleared() {
+ addRecords("a", 2);
+ await().atMost(20, TimeUnit.SECONDS).until(() -> routed.contains("a1"));
+
+ addRecords("b", 2);
+ await().atMost(20, TimeUnit.SECONDS).until(() -> routed.contains("b1"));
+
+ assertEquals(List.of("a0", "a1", "b0", "b1"), routed);
+ }
+
+ private void addRecords(String prefix, int count) {
+ await().atMost(20, TimeUnit.SECONDS).until(() -> mockConsumer.assignment().contains(PARTITION));
+ List<ConsumerRecord<Object, Object>> records = new ArrayList<>();
+ for (int i = 0; i < count; i++) {
+ records.add(new ConsumerRecord<>(TOPIC, 0, offset++, null, prefix + i));
+ }
+ records.forEach(mockConsumer::addRecord);
+ }
+}