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 8723d9168286 CAMEL-25313: camel-kafka - a suspended consumer must keep
its fetcher thread, and a pause or resume request must not be lost (#27389)
8723d9168286 is described below
commit 8723d916828675d694449451c5babf3a637d4adc
Author: allthingssecurity <[email protected]>
AuthorDate: Tue Oct 6 14:52:15 2026 +0530
CAMEL-25313: camel-kafka - a suspended consumer must keep its fetcher
thread, and a pause or resume request must not be lost (#27389)
* CAMEL-25313: camel-kafka - a suspended consumer must keep its fetcher
thread, and a pause or resume request must not be lost
Suspending and resuming the route of a Kafka consumer only sets the
requested
state of each KafkaFetchRecords task (pause()/resume() from the caller
thread);
the fetcher thread applies it after the next poll. Two defects:
- updateTaskState read the state, called consumer.pause()/resume() and then
set PAUSED/RUNNING unconditionally. A resume (or suspend) requested in the
meantime, e.g. by the half-open timer of ThrottlingExceptionRoutePolicy or
by the route controller, was overwritten: the Kafka consumer stayed paused
while the route was started (it never consumed again), or kept consuming
while the route was suspended.
- run() returns at once, and the do-while loop ends, when the consumer is
suspended. A fetcher thread that starts while the consumer is suspended
(keepOpen of ThrottlingExceptionRoutePolicy suspends the consumer when the
route starts, before the pending consumers are started) or that leaves the
polling loop while it is suspended (a failed poll with the default
pollOnError=ERROR_HANDLER, or a reconnect) terminated, and resuming the
route
only set the state of a task that no thread runs: the route was started
but
never consumed again.
updateTaskState now moves to PAUSED/RUNNING with compareAndSet, so a newer
request is handled on the next iteration; the thread only ends when the
consumer is stopping or stopped (the polling loop already keeps polling,
with
the consumer paused, while it is suspended); and after a (re)connect a
PAUSED
state becomes PAUSE_REQUESTED again, as the new Kafka consumer is not
paused.
A fetcher thread that now starts (keepOpen=true: the route policy suspends
the consumer before its tasks are
submitted) or reconnects while the consumer is suspended processed the
records of the poll that assigns the
partitions before updateTaskState paused them: with keepOpen=true the first
batch was consumed with the
circuit open. The rebalance listener now pauses the assigned partitions
when a pause is requested or
applied (it runs on the fetcher thread, inside poll, so that poll does not
return their records), and
keeps re-pausing the whole assignment after the poll as before.
Co-Authored-By: Claude Opus 5.5 <[email protected]>
* CAMEL-25313: camel-kafka - a suspended consumer is recoverable, and
upgrade guide note for keepOpen
The fetcher thread of a suspended consumer now keeps polling and
reconnecting, but isRecoverable() (health check
message) still treated a suspended consumer as not runnable. The upgrade
guide notes that a route started with its
consumer suspended (ThrottlingExceptionRoutePolicy with keepOpen=true) now
joins the consumer group and holds its
partitions paused.
Co-Authored-By: Claude Opus 5.5 <[email protected]>
* CAMEL-25313: camel-kafka - Give the suspend note its own upgrade-guide
heading
Co-Authored-By: Claude Opus 5.5 <[email protected]>
---------
Co-authored-by: Claude Opus 5.5 <[email protected]>
---
.../camel/component/kafka/KafkaFetchRecords.java | 28 +-
.../kafka/KafkaConsumerSuspendResumeTest.java | 304 +++++++++++++++++++++
.../ROOT/pages/camel-4x-upgrade-guide-4_23.adoc | 9 +
3 files changed, 331 insertions(+), 10 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 40455caddfc0..4a2c2eacd645 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
@@ -126,7 +126,8 @@ public class KafkaFetchRecords implements Runnable {
@Override
public void run() {
- if (!isKafkaConsumerRunnable()) {
+ // a suspended consumer keeps its thread, which polls with the
consumer paused, so it can be resumed
+ if (!isKafkaConsumerRunnableAndNotStopped()) {
return;
}
@@ -171,6 +172,8 @@ public class KafkaFetchRecords implements Runnable {
}
setConnected(true);
+ // a new Kafka consumer is not paused: pause it again if
the consumer is suspended
+ state.compareAndSet(State.PAUSED, State.PAUSE_REQUESTED);
}
if (isConnected()) {
@@ -179,7 +182,7 @@ public class KafkaFetchRecords implements Runnable {
setLastError(null);
startPolling();
- } while ((pollExceptionStrategy.canContinue() || isReconnect()) &&
isKafkaConsumerRunnable());
+ } while ((pollExceptionStrategy.canContinue() || isReconnect()) &&
isKafkaConsumerRunnableAndNotStopped());
if (LOG.isInfoEnabled()) {
LOG.info("Terminating KafkaConsumer thread {} receiving from
{}", threadId, getPrintableTopic());
@@ -424,11 +427,13 @@ public class KafkaFetchRecords implements Runnable {
}
private void updateTaskState() {
+ // pause() and resume() are called from other threads: only move to
PAUSED or RUNNING when no other request
+ // was made in the meantime, otherwise that request is handled on the
next iteration
switch (state.get()) {
case PAUSE_REQUESTED:
LOG.info("Pausing the consumer as a response to a pause
request");
consumer.pause(consumer.assignment());
- state.set(State.PAUSED);
+ state.compareAndSet(State.PAUSE_REQUESTED, State.PAUSED);
break;
case RESUME_REQUESTED:
LOG.info("Resuming the consumer as a response to a resume
request");
@@ -443,7 +448,7 @@ public class KafkaFetchRecords implements Runnable {
});
}
consumer.resume(consumer.assignment());
- state.set(State.RUNNING);
+ state.compareAndSet(State.RESUME_REQUESTED, State.RUNNING);
break;
default:
break;
@@ -491,11 +496,6 @@ public class KafkaFetchRecords implements Runnable {
return TopicHelper.getPrintableTopic(topicPattern, topicName);
}
- private boolean isKafkaConsumerRunnable() {
- return kafkaConsumer.isRunAllowed() &&
!kafkaConsumer.isStoppingOrStopped()
- && !kafkaConsumer.isSuspendingOrSuspended();
- }
-
private boolean isKafkaConsumerRunnableAndNotStopped() {
return kafkaConsumer.isRunAllowed() &&
!kafkaConsumer.isStoppingOrStopped();
}
@@ -580,8 +580,9 @@ public class KafkaFetchRecords implements Runnable {
}
private boolean isRecoverable() {
+ // a suspended consumer keeps polling and reconnecting
return (pollExceptionStrategy != null &&
pollExceptionStrategy.canContinue() || isReconnect())
- && isKafkaConsumerRunnable();
+ && isKafkaConsumerRunnableAndNotStopped();
}
// concurrent access happens here
@@ -643,6 +644,13 @@ public class KafkaFetchRecords implements Runnable {
@Override
public void onPartitionsAssigned(Collection<TopicPartition>
partitions) {
+ State current = state.get();
+ if (current == State.PAUSED || current == State.PAUSE_REQUESTED) {
+ // the consumer is suspended: pause the assigned partitions
now (this runs on the fetcher thread, inside
+ // poll), so that the poll does not return their records, e.g.
the first poll of a consumer that starts
+ // or reconnects while suspended
+ consumer.pause(partitions);
+ }
if (state.compareAndSet(State.PAUSED, State.PAUSE_REQUESTED)) {
LOG.debug("Partitions were assigned while paused, the consumer
will be re-paused");
}
diff --git
a/components/camel-kafka/src/test/java/org/apache/camel/component/kafka/KafkaConsumerSuspendResumeTest.java
b/components/camel-kafka/src/test/java/org/apache/camel/component/kafka/KafkaConsumerSuspendResumeTest.java
new file mode 100644
index 000000000000..6b1400f06e97
--- /dev/null
+++
b/components/camel-kafka/src/test/java/org/apache/camel/component/kafka/KafkaConsumerSuspendResumeTest.java
@@ -0,0 +1,304 @@
+/*
+ * 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.Collection;
+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 java.util.concurrent.locks.LockSupport;
+
+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.camel.throttling.ThrottlingExceptionRoutePolicy;
+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.KafkaException;
+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.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+/**
+ * Suspending and resuming a route with a Kafka consumer (route controller,
JMX, route policies) against its record
+ * fetcher thread, with a {@link MockConsumer} instead of a broker.
+ */
+class KafkaConsumerSuspendResumeTest extends CamelTestSupport {
+
+ private static final String TOPIC = "suspend-resume";
+ private static final TopicPartition PARTITION = new TopicPartition(TOPIC,
0);
+
+ private final GatedMockConsumer mockConsumer = new GatedMockConsumer();
+ private final ThrottlingExceptionRoutePolicy keepOpenPolicy
+ = new ThrottlingExceptionRoutePolicy(1, 60000, 100, null, true);
+ 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
+ public boolean isUseRouteBuilder() {
+ return false;
+ }
+
+ @Test
+ void testResumeWhileThePauseIsApplied() throws Exception {
+ startRoute(false);
+ KafkaConsumer consumer = kafkaConsumer();
+
+ // the fetcher thread is inside consumer.pause() when the route is
resumed
+ Gate gate = new Gate();
+ mockConsumer.pauseGate = gate;
+ context.getRouteController().suspendRoute("kafka");
+ assertTrue(gate.entered.await(20, TimeUnit.SECONDS));
+ context.getRouteController().resumeRoute("kafka");
+ gate.proceed.countDown();
+
+ MockEndpoint mock = getMockEndpoint("mock:result");
+ mock.expectedBodiesReceived("b");
+ addRecord("b");
+ mock.assertIsSatisfied();
+ assertFalse(consumer.isKafkaPaused());
+ }
+
+ @Test
+ void testSuspendWhileTheResumeIsApplied() throws Exception {
+ startRoute(false);
+ KafkaConsumer consumer = kafkaConsumer();
+ context.getRouteController().suspendRoute("kafka");
+ await().atMost(20, TimeUnit.SECONDS).until(consumer::isKafkaPaused);
+
+ // the fetcher thread is inside consumer.resume() when the route is
suspended again
+ Gate gate = new Gate();
+ mockConsumer.resumeGate = gate;
+ context.getRouteController().resumeRoute("kafka");
+ assertTrue(gate.entered.await(20, TimeUnit.SECONDS));
+ context.getRouteController().suspendRoute("kafka");
+ gate.proceed.countDown();
+
+ await().atMost(20, TimeUnit.SECONDS).untilAsserted(
+ () -> assertTrue(consumer.isKafkaPaused(), "The Kafka consumer
must be paused while the route is suspended"));
+ MockEndpoint mock = getMockEndpoint("mock:result");
+ mock.expectedMessageCount(0);
+ mock.setAssertPeriod(500);
+ addRecord("b");
+ mock.assertIsSatisfied();
+
+ mock.reset();
+ mock.expectedBodiesReceived("b");
+ context.getRouteController().resumeRoute("kafka");
+ mock.assertIsSatisfied();
+ }
+
+ @Test
+ void testPollErrorWhileSuspended() throws Exception {
+ startRoute(false);
+ KafkaConsumer consumer = kafkaConsumer();
+ context.getRouteController().suspendRoute("kafka");
+ await().atMost(20, TimeUnit.SECONDS).until(consumer::isKafkaPaused);
+
+ // one poll fails while the route is suspended (handled by the default
pollOnError=ERROR_HANDLER)
+ mockConsumer.setPollException(new KafkaException("Simulated poll
failure"));
+ await().atMost(20, TimeUnit.SECONDS).until(() ->
mockConsumer.pollErrors.get() == 1);
+
+ MockEndpoint mock = getMockEndpoint("mock:result");
+ mock.expectedBodiesReceived("b");
+ context.getRouteController().resumeRoute("kafka");
+ addRecord("b");
+ mock.assertIsSatisfied();
+ }
+
+ @Test
+ void testSuspendedConsumerIsRecoverable() throws Exception {
+ startRoute(false);
+ KafkaConsumer consumer = kafkaConsumer();
+ context.getRouteController().suspendRoute("kafka");
+ await().atMost(20, TimeUnit.SECONDS).until(consumer::isKafkaPaused);
+
+ // the fetcher thread of a suspended consumer keeps polling (and
reconnecting), so the health check must not
+ // report that it gave up recovering
+
assertTrue(consumer.healthStates().stream().allMatch(TaskHealthState::isRecoverable),
+ "A suspended consumer is recoverable");
+ }
+
+ @Test
+ void testStartedWithOpenCircuit() throws Exception {
+ // keepOpen suspends the consumer when the route starts, before the
fetcher thread runs
+ startRoute(true);
+
+ // the record is added by the poll after the one that assigns the
partition
+ long next = offset++;
+ mockConsumer.schedulePollTask(() -> mockConsumer.addRecord(new
ConsumerRecord<>(TOPIC, 0, next, null, "b")));
+
+ MockEndpoint mock = getMockEndpoint("mock:result");
+ mock.expectedBodiesReceived("b");
+ keepOpenPolicy.setKeepOpen(false);
+ mock.assertIsSatisfied();
+ }
+
+ @Test
+ void testStartedWithOpenCircuitConsumesNothing() throws Exception {
+ // a record is on the topic when the route starts with the circuit
open: the poll that assigns the partition
+ // must not return it
+ long first = offset++;
+ startRoute(true, () -> mockConsumer.addRecord(new
ConsumerRecord<>(TOPIC, 0, first, null, "a")));
+ KafkaConsumer consumer = kafkaConsumer();
+ await().atMost(20, TimeUnit.SECONDS).until(consumer::isKafkaPaused);
+
+ MockEndpoint mock = getMockEndpoint("mock:result");
+ mock.expectedMessageCount(0);
+ mock.setAssertPeriod(500);
+ mock.assertIsSatisfied();
+
+ mock.reset();
+ mock.expectedBodiesReceived("a");
+ keepOpenPolicy.setKeepOpen(false);
+ mock.assertIsSatisfied();
+ }
+
+ private void startRoute(boolean keepOpen) throws Exception {
+ startRoute(keepOpen, () -> {
+ });
+ }
+
+ private void startRoute(boolean keepOpen, Runnable afterAssignment) throws
Exception {
+ mockConsumer.updateBeginningOffsets(Map.of(PARTITION, 0L));
+ mockConsumer.schedulePollTask(() -> {
+ mockConsumer.rebalance(List.of(PARTITION));
+ afterAssignment.run();
+ });
+ context.addRoutes(new RouteBuilder() {
+ @Override
+ public void configure() {
+ from("kafka:" + TOPIC +
"?brokers=localhost:9092&kafkaClientFactory=#mockFactory&groupId=test&pollTimeoutMs=10")
+ .routeId("kafka").routePolicy(keepOpenPolicy)
+ .to("mock:result");
+ }
+ });
+ if (!keepOpen) {
+ keepOpenPolicy.setKeepOpen(false);
+ }
+ context.start();
+ if (!keepOpen) {
+ MockEndpoint mock = getMockEndpoint("mock:result");
+ mock.expectedBodiesReceived("a");
+ addRecord("a");
+ mock.assertIsSatisfied();
+ mock.reset();
+ }
+ }
+
+ private KafkaConsumer kafkaConsumer() {
+ return (KafkaConsumer) context.getRoute("kafka").getConsumer();
+ }
+
+ private void addRecord(String value) {
+ await().atMost(20, TimeUnit.SECONDS).until(() ->
mockConsumer.assignment().contains(PARTITION));
+ mockConsumer.addRecord(new ConsumerRecord<>(TOPIC, 0, offset++, null,
value));
+ }
+
+ private static final class Gate {
+ final CountDownLatch entered = new CountDownLatch(1);
+ final CountDownLatch proceed = new CountDownLatch(1);
+
+ void pass() {
+ entered.countDown();
+ try {
+ proceed.await(20, TimeUnit.SECONDS);
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ }
+ }
+ }
+
+ private static final class GatedMockConsumer extends MockConsumer<Object,
Object> {
+ volatile Gate pauseGate;
+ volatile Gate resumeGate;
+ final AtomicInteger pollErrors = new AtomicInteger();
+
+ GatedMockConsumer() {
+ super("earliest");
+ }
+
+ @Override
+ public 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 void pause(Collection<TopicPartition> partitions) {
+ Gate gate = pauseGate;
+ pauseGate = null;
+ if (gate != null) {
+ gate.pass();
+ }
+ super.pause(partitions);
+ }
+
+ @Override
+ public void resume(Collection<TopicPartition> partitions) {
+ Gate gate = resumeGate;
+ resumeGate = null;
+ if (gate != null) {
+ gate.pass();
+ }
+ super.resume(partitions);
+ }
+
+ @Override
+ public synchronized void close() {
+ // keep the mock usable: a stopped route closes the consumer
+ }
+ }
+}
diff --git
a/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_23.adoc
b/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_23.adoc
index dce46b5c5d5b..671d952fec02 100644
--- a/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_23.adoc
+++ b/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_23.adoc
@@ -3978,6 +3978,15 @@ the SSL endpoint options were ignored when
`sslContextParameters` was configured
When `sslContextParameters` is configured and `securityProtocol` is the
default (`PLAINTEXT`), the security protocol
`SSL` is now used. Previously `PLAINTEXT` was used unless `securityProtocol`
(or `saslAuthType`) was configured.
+=== camel-kafka - a suspended consumer keeps its partitions
+
+A route started with its consumer suspended, for example by a
`ThrottlingExceptionRoutePolicy` with `keepOpen=true`,
+now starts its Kafka consumer: the consumer joins the consumer group and holds
its assigned partitions paused until the
+route is resumed. Previously its fetcher thread ended at once, so it never
connected, and the other members of the group
+consumed those partitions while the circuit was open; now they stay with this
consumer, as for a route suspended while
+it runs. A suspended consumer also keeps its fetcher thread after a failed
poll or a reconnect, and the health check
+reports it as recoverable.
+
=== camel-pulsar - the producer no longer replaces the body with the message id
After sending, the producer used to overwrite the message body with the
`MessageId` returned by the
