This is an automated email from the ASF dual-hosted git repository. davsclaus pushed a commit to branch fix/CAMEL-25753 in repository https://gitbox.apache.org/repos/asf/camel.git
commit 59b20ec7c19c0d86cac7a171f7b8f34e75ff89b5 Author: Claus Ibsen <[email protected]> AuthorDate: Sun Oct 18 16:46:40 2026 +0200 CAMEL-25753: camel-kafka - stop the consumer without waiting for its current poll KafkaFetchRecords held its lock for the whole poll loop, and stop() waited for that lock before it woke up the consumer, so the wakeup never ended a poll: each consumer needed up to pollTimeoutMs (5 s by default) to stop, one route after another. camel run --dev stops the routes of a reload within 5 s, so a reload with two or more Kafka consumer routes was forced, and the interrupted Kafka clients logged errors. The lock now covers the processing of the polled records only. stop() waits for a processing to finish, including its commit, as before, and wakes up a consumer that is in poll, which then stops at once. A consumer between two polls sees the stop before it polls again, so no wakeup is left pending to interrupt the records it already has. Co-Authored-By: Claude Opus 5.5 (1M context) <[email protected]> Claude-Session: https://claude.ai/code/session_01STT6whBgK1AqsSsUKrnE8m --- .../camel/component/kafka/KafkaFetchRecords.java | 87 ++++++--- .../component/kafka/KafkaConsumerStopTest.java | 200 +++++++++++++++++++++ 2 files changed, 258 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..986481dafdc8 --- /dev/null +++ b/components/camel-kafka/src/test/java/org/apache/camel/component/kafka/KafkaConsumerStopTest.java @@ -0,0 +1,200 @@ +/* + * 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 java.util.concurrent.atomic.AtomicInteger; + +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> { + final AtomicInteger polls = new AtomicInteger(); + private final Object signal = new Object(); + private boolean wokenUp; + private boolean waiting; + + BlockingMockConsumer() { + super("earliest"); + } + + @Override + public ConsumerRecords<Object, Object> poll(Duration timeout) { + polls.incrementAndGet(); + 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 + } + } +}
