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 184450a65483 CAMEL-25753: camel-kafka - stop the consumer without
waiting for its current poll (#28060)
184450a65483 is described below
commit 184450a65483f98764a84c9afbc81e4d69ad98ae
Author: Claus Ibsen <[email protected]>
AuthorDate: Sun Oct 18 18:03:03 2026 +0200
CAMEL-25753: camel-kafka - stop the consumer without waiting for its
current poll (#28060)
Co-Authored-By: Claude Opus 5.5 (1M context) <[email protected]>
---
.../camel/component/kafka/KafkaFetchRecords.java | 87 ++++++---
.../component/kafka/KafkaConsumerStopTest.java | 197 +++++++++++++++++++++
2 files changed, 255 insertions(+), 29 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 8994c2d80641..23f09a65f9da 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
@@ -90,6 +90,8 @@ public class KafkaFetchRecords implements Runnable {
private PollExceptionStrategy pollExceptionStrategy;
private final BridgeExceptionHandlerToErrorHandler bridge;
private final ReentrantLock lock = new ReentrantLock();
+ // whether the fetcher thread is in poll, which stop() may then wake up
(CAMEL-25753)
+ private volatile boolean polling;
private CommitManager commitManager;
private volatile Exception lastError;
private final KafkaConsumerListener consumerListener;
@@ -317,12 +319,6 @@ public class KafkaFetchRecords implements Runnable {
protected void startPolling() {
try {
- /*
- * We lock the processing of the record to avoid raising a
WakeUpException as a result to a call
- * to stop() or shutdown().
- */
- lock.lock();
-
long pollTimeoutMs =
kafkaConsumer.getEndpoint().getConfiguration().getPollTimeoutMs();
Duration pollDuration = Duration.ofMillis(pollTimeoutMs);
@@ -338,27 +334,51 @@ public class KafkaFetchRecords implements Runnable {
// which must happen using this polling thread, so we use the
commitRecordsRequested to trigger this
metricsCollector.collectCommitMetrics(consumer);
- ConsumerRecords<Object, Object> allRecords =
consumer.poll(pollDuration);
- if (consumerListener != null) {
- if (!consumerListener.afterConsume(consumer)) {
- continue;
+ /*
+ * The poll is not under the lock: stop() wakes up a consumer
that is waiting in poll, which then
+ * throws a WakeupException, so the consumer stops at once
instead of after the poll timeout
+ * (CAMEL-25753). The stop is checked again once polling is
set, so a stop that began just before is
+ * seen here, or stop() sees polling and wakes up the poll.
+ */
+ ConsumerRecords<Object, Object> allRecords;
+ polling = true;
+ try {
+ if (!isKafkaConsumerRunnableAndNotStopped()) {
+ break;
}
+ allRecords = consumer.poll(pollDuration);
+ } finally {
+ polling = false;
}
- ProcessingResult result =
recordProcessorFacade.processPolledRecords(allRecords);
- if (result != null && result.getTopic() != null) {
- metricsCollector.storeLastRecord(result);
- }
- updateTaskState();
-
- // when breakOnFirstError we want to unsubscribe from Kafka
- if (result != null && result.isBreakOnErrorHit() &&
!this.state.get().equals(State.PAUSED)) {
- LOG.debug("We hit an error ... setting flags to force
reconnect");
- // force re-connect
- setReconnect(true);
- setConnected(false);
- }
+ /*
+ * The processing of the polled records is under the lock:
stop() waits for it to finish, including
+ * the commit, before it wakes up the consumer, so a
WakeupException does not interrupt it.
+ */
+ lock.lock();
+ try {
+ if (consumerListener != null) {
+ if (!consumerListener.afterConsume(consumer)) {
+ continue;
+ }
+ }
+ ProcessingResult result =
recordProcessorFacade.processPolledRecords(allRecords);
+ if (result != null && result.getTopic() != null) {
+ metricsCollector.storeLastRecord(result);
+ }
+ updateTaskState();
+
+ // when breakOnFirstError we want to unsubscribe from Kafka
+ if (result != null && result.isBreakOnErrorHit() &&
!this.state.get().equals(State.PAUSED)) {
+ LOG.debug("We hit an error ... setting flags to force
reconnect");
+ // force re-connect
+ setReconnect(true);
+ setConnected(false);
+ }
+ } finally {
+ lock.unlock();
+ }
}
if (!isConnected()) {
@@ -400,7 +420,6 @@ public class KafkaFetchRecords implements Runnable {
safeUnsubscribe();
safeConsumerClose();
}
- lock.unlock();
}
}
@@ -470,6 +489,9 @@ public class KafkaFetchRecords implements Runnable {
} catch (IllegalStateException e) {
LOG.warn("The consumer is likely already closed. Skipping
unsubscribing thread {} from kafka {}", threadId,
printableTopic);
+ } catch (WakeupException e) {
+ // stop() woke up the consumer after the last poll: the consumer
is closed next, which leaves the group
+ LOG.debug("The kafka consumer was woken up while unsubscribing
thread {} from {}", threadId, printableTopic);
} catch (Exception e) {
LOG.debug("Something went wrong while unsubscribing from Kafka:
{}", e.getMessage());
kafkaConsumer.getExceptionHandler().handleException(
@@ -500,7 +522,8 @@ public class KafkaFetchRecords implements Runnable {
/*
* This wraps a safe stop procedure that should help ensure a clean
termination procedure for consumer code.
* This means that it should wait for the last process call to finish
cleanly, including the commit of the
- * record being processed at the current moment.
+ * record being processed at the current moment. The lock is held only
while records are processed: when the
+ * consumer is waiting in poll, the lock is free and the wakeup ends the
poll at once (CAMEL-25753).
*
* Note: keep in mind that the KafkaConsumer is not thread-safe, so no
other call to the consumer instance
* should be made here besides the wakeUp.
@@ -517,13 +540,19 @@ public class KafkaFetchRecords implements Runnable {
of whether the processing have finished or not.
*/
LOG.info("Waiting up to {} milliseconds for the processing to
finish", timeout);
- if (!lock.tryLock(timeout, TimeUnit.MILLISECONDS)) {
+ boolean locked = lock.tryLock(timeout, TimeUnit.MILLISECONDS);
+ if (!locked) {
LOG.warn("The processing of the current record did not finish
within {} seconds", timeout);
}
- // As advised in the KAFKA-1894 ticket, calling this wakeup method
breaks the infinite loop
- LOG.trace("Waking up Kafka consumer");
- consumer.wakeup();
+ // As advised in the KAFKA-1894 ticket, calling this wakeup method
breaks the infinite loop. Only a
+ // consumer in poll is woken up: a consumer between two polls sees
the stop before it polls again, and a
+ // wakeup would stay pending and interrupt the processing of the
records it already has. A processing
+ // that did not finish in time is woken up regardless, as before.
+ if (polling || !locked) {
+ LOG.trace("Waking up Kafka consumer");
+ consumer.wakeup();
+ }
} catch (InterruptedException e) {
LOG.trace("Interrupted while waiting for processing to finish:
waking up Kafka consumer");
consumer.wakeup();
diff --git
a/components/camel-kafka/src/test/java/org/apache/camel/component/kafka/KafkaConsumerStopTest.java
b/components/camel-kafka/src/test/java/org/apache/camel/component/kafka/KafkaConsumerStopTest.java
new file mode 100644
index 000000000000..8638b90ff417
--- /dev/null
+++
b/components/camel-kafka/src/test/java/org/apache/camel/component/kafka/KafkaConsumerStopTest.java
@@ -0,0 +1,197 @@
+/*
+ * 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.List;
+import java.util.Map;
+import java.util.Properties;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+
+import org.apache.camel.BindToRegistry;
+import org.apache.camel.builder.RouteBuilder;
+import org.apache.camel.component.mock.MockEndpoint;
+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.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.assertTrue;
+
+/**
+ * Stopping a route with a Kafka consumer (CAMEL-25753): a consumer waiting in
poll is woken up at once instead of after
+ * the poll timeout, and a consumer processing records finishes them first. A
{@link MockConsumer} instead of a broker,
+ * whose poll waits like a real one.
+ */
+class KafkaConsumerStopTest extends CamelTestSupport {
+
+ private static final String TOPIC = "stop";
+ private static final TopicPartition PARTITION = new TopicPartition(TOPIC,
0);
+
+ private final BlockingMockConsumer mockConsumer = new
BlockingMockConsumer();
+ private final CountDownLatch processing = new CountDownLatch(1);
+
+ @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
+ public boolean isUseRouteBuilder() {
+ return false;
+ }
+
+ @Test
+ void testStopWhileWaitingInPoll() throws Exception {
+ startRoute(0);
+ // the fetcher thread is in a poll that waits up to 10 seconds
+ await().atMost(20, TimeUnit.SECONDS).until(mockConsumer::isWaiting);
+
+ long start = System.nanoTime();
+ context.getRouteController().stopRoute("kafka");
+ long millis = TimeUnit.NANOSECONDS.toMillis(System.nanoTime() - start);
+
+ assertTrue(millis < 3000, "The route stopped after " + millis + " ms,
it should not wait for the 10 s poll timeout");
+ }
+
+ @Test
+ void testStopWhileProcessing() throws Exception {
+ startRoute(1000);
+ MockEndpoint mock = getMockEndpoint("mock:result");
+ mock.expectedBodiesReceived("a");
+ await().atMost(20, TimeUnit.SECONDS).until(() ->
mockConsumer.assignment().contains(PARTITION));
+ mockConsumer.addRecord(new ConsumerRecord<>(TOPIC, 0, 0, null, "a"));
+ assertTrue(processing.await(20, TimeUnit.SECONDS));
+
+ // the record takes a second: the stop waits for it, and the wakeup
does not interrupt it
+ long start = System.nanoTime();
+ context.getRouteController().stopRoute("kafka");
+ long millis = TimeUnit.NANOSECONDS.toMillis(System.nanoTime() - start);
+
+ mock.assertIsSatisfied();
+ assertTrue(millis >= 500, "The route stopped after " + millis + " ms,
before the record was processed");
+ }
+
+ private void startRoute(long processingMillis) throws Exception {
+ mockConsumer.updateBeginningOffsets(Map.of(PARTITION, 0L));
+ mockConsumer.schedulePollTask(() ->
mockConsumer.rebalance(List.of(PARTITION)));
+ context.addRoutes(new RouteBuilder() {
+ @Override
+ public void configure() {
+ from("kafka:" + TOPIC
+ +
"?brokers=localhost:9092&kafkaClientFactory=#mockFactory&groupId=test&pollTimeoutMs=10000")
+ .routeId("kafka")
+ .process(e -> {
+ processing.countDown();
+ Thread.sleep(processingMillis);
+ })
+ .to("mock:result");
+ }
+ });
+ context.start();
+ }
+
+ /**
+ * A poll that waits up to its timeout for records, as a real consumer on
an idle topic, unless it is woken up. The
+ * wait is outside the monitor of {@link MockConsumer}, whose poll and
wakeup are synchronized.
+ */
+ private static final class BlockingMockConsumer extends
MockConsumer<Object, Object> {
+ private final Object signal = new Object();
+ private boolean wokenUp;
+ private boolean waiting;
+
+ BlockingMockConsumer() {
+ super("earliest");
+ }
+
+ @Override
+ public ConsumerRecords<Object, Object> poll(Duration timeout) {
+ long deadline = System.nanoTime() + timeout.toNanos();
+ while (true) {
+ // throws a WakeupException when woken up
+ ConsumerRecords<Object, Object> records =
super.poll(Duration.ZERO);
+ if (!records.isEmpty()) {
+ return records;
+ }
+ synchronized (signal) {
+ long left = deadline - System.nanoTime();
+ if (left <= 0) {
+ return records;
+ }
+ if (!wokenUp) {
+ waiting = true;
+ try {
+ signal.wait(Math.max(1,
TimeUnit.NANOSECONDS.toMillis(left)));
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ return records;
+ } finally {
+ waiting = false;
+ }
+ }
+ }
+ }
+ }
+
+ @Override
+ public void addRecord(ConsumerRecord<Object, Object> record) {
+ super.addRecord(record);
+ synchronized (signal) {
+ signal.notifyAll();
+ }
+ }
+
+ boolean isWaiting() {
+ synchronized (signal) {
+ return waiting;
+ }
+ }
+
+ @Override
+ public void wakeup() {
+ super.wakeup();
+ synchronized (signal) {
+ wokenUp = true;
+ signal.notifyAll();
+ }
+ }
+
+ @Override
+ public synchronized void close() {
+ // keep the mock usable: a stopped route closes the consumer
+ }
+ }
+}
