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 4d7629668f68 CAMEL-26709: camel-kafka - SingleNodeKafkaResumeStrategy
does not work after a route restart (#29408)
4d7629668f68 is described below
commit 4d7629668f684efcc5bc5a14ef54c9f95323aff8
Author: allthingssecurity <[email protected]>
AuthorDate: Sun Nov 8 15:49:05 2026 +0530
CAMEL-26709: camel-kafka - SingleNodeKafkaResumeStrategy does not work
after a route restart (#29408)
* CAMEL-26709: camel-kafka - SingleNodeKafkaResumeStrategy does not work
after a route restart
* CAMEL-26709: camel-kafka - review: each refresh task of
SingleNodeKafkaResumeStrategy closes only its own consumer
Co-Authored-By: Claude Opus 5.5 <[email protected]>
---
.../kafka/SingleNodeKafkaResumeStrategy.java | 75 +++++--
.../kafka/SingleNodeKafkaResumeStrategyTest.java | 225 +++++++++++++++++++++
2 files changed, 283 insertions(+), 17 deletions(-)
diff --git
a/components/camel-kafka/src/main/java/org/apache/camel/processor/resume/kafka/SingleNodeKafkaResumeStrategy.java
b/components/camel-kafka/src/main/java/org/apache/camel/processor/resume/kafka/SingleNodeKafkaResumeStrategy.java
index 7bd0ad00875b..d4b872cc5375 100644
---
a/components/camel-kafka/src/main/java/org/apache/camel/processor/resume/kafka/SingleNodeKafkaResumeStrategy.java
+++
b/components/camel-kafka/src/main/java/org/apache/camel/processor/resume/kafka/SingleNodeKafkaResumeStrategy.java
@@ -65,7 +65,8 @@ import org.slf4j.LoggerFactory;
public class SingleNodeKafkaResumeStrategy implements KafkaResumeStrategy,
CamelContextAware {
private static final Logger LOG =
LoggerFactory.getLogger(SingleNodeKafkaResumeStrategy.class);
- private Consumer<byte[], byte[]> consumer;
+ // the consumer of the running refresh task, for wakeup() in stop(); each
task closes only its own consumer
+ private volatile Consumer<byte[], byte[]> consumer;
private Producer<byte[], byte[]> producer;
private Duration pollDuration = Duration.ofSeconds(1);
@@ -109,10 +110,14 @@ public class SingleNodeKafkaResumeStrategy implements
KafkaResumeStrategy, Camel
*
*/
protected void produce(byte[] key, byte[] message, UpdateCallBack
updateCallBack) {
+ Producer<byte[], byte[]> current = producer;
+ if (current == null) {
+ throw new IllegalStateException("Cannot update the offset as the
Kafka resume strategy is not started");
+ }
ProducerRecord<byte[], byte[]> producerRecord
= new ProducerRecord<>(resumeStrategyConfiguration.getTopic(),
key, message);
- producer.send(producerRecord, (recordMetadata, e) -> {
+ current.send(producerRecord, (recordMetadata, e) -> {
if (e != null) {
LOG.error("Failed to send message {}", e.getMessage(), e);
}
@@ -203,22 +208,28 @@ public class SingleNodeKafkaResumeStrategy implements
KafkaResumeStrategy, Camel
private void refresh(CountDownLatch latch) {
LOG.trace("Creating a offset cache refresher");
+ // a local consumer: a task of an earlier start that outlives stop()
must close its own consumer, not the one of
+ // the task that runs now
+ Consumer<byte[], byte[]> current = null;
try {
- consumer = createConsumer();
+ current = createConsumer();
+ consumer = current;
+ // a new consumer (such as after a restart) must subscribe again
+ subscribed = false;
- subscribe(consumer);
+ subscribe(current);
LOG.debug("Loading records from topic {}",
resumeStrategyConfiguration.getTopic());
- poll(consumer, latch);
+ poll(current, latch);
} catch (WakeupException e) {
LOG.info("Kafka consumer was interrupted during a blocking call");
} catch (Exception e) {
LOG.error("Error while refreshing the local cache: {}",
e.getMessage(), e);
} finally {
- if (consumer != null) {
- consumer.unsubscribe();
+ if (current != null) {
+ current.unsubscribe();
try {
-
consumer.close(CloseOptions.timeout(Duration.ofSeconds(5)));
+ current.close(CloseOptions.timeout(Duration.ofSeconds(5)));
} catch (Exception e) {
LOG.warn("Error closing the consumer: {} (this error will
be ignored)", e.getMessage(), e);
}
@@ -241,9 +252,25 @@ public class SingleNodeKafkaResumeStrategy implements
KafkaResumeStrategy, Camel
consumerRecord.timestampType(), value);
}
- if
(!deserializable.deserialize(ByteBuffer.wrap(consumerRecord.key()),
- ByteBuffer.wrap(consumerRecord.value()))) {
- LOG.warn("Deserializer indicates that this is the last
record to deserialize");
+ if (value == null) {
+ // a tombstone, which on a compacted topic means "delete
this key". Camel never writes tombstones
+ // to the resume topic, so it is skipped, and the key
keeps the offset it has in the cache
+ LOG.warn("Ignoring the tombstone (a record without a
value) at offset {} of {}-{}: Camel does not write"
+ + " tombstones to the resume topic, so the offset
of its key is kept",
+ consumerRecord.offset(), consumerRecord.topic(),
consumerRecord.partition());
+ continue;
+ }
+
+ try {
+ if
(!deserializable.deserialize(ByteBuffer.wrap(consumerRecord.key()),
+ ByteBuffer.wrap(consumerRecord.value()))) {
+ LOG.warn("Deserializer indicates that this is the last
record to deserialize");
+ }
+ } catch (Exception e) {
+ // for example a record without a key: skip it and keep
loading the others
+ LOG.warn("Ignoring the record at offset {} of {}-{}
because it cannot be deserialized: {}",
+ consumerRecord.offset(), consumerRecord.topic(),
consumerRecord.partition(), e.getMessage(),
+ e);
}
}
@@ -384,7 +411,8 @@ public class SingleNodeKafkaResumeStrategy implements
KafkaResumeStrategy, Camel
}
}
- private Consumer<byte[], byte[]> createConsumer() {
+ // package-private for the tests
+ Consumer<byte[], byte[]> createConsumer() {
return new
KafkaConsumer<>(resumeStrategyConfiguration.getConsumerProperties());
}
@@ -398,6 +426,8 @@ public class SingleNodeKafkaResumeStrategy implements
KafkaResumeStrategy, Camel
LOG.info("Closing the Kafka producer");
IOHelper.close(producer, "Kafka producer", LOG);
+ // a new producer is created when this strategy is started again
+ producer = null;
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
} catch (Exception e) {
@@ -418,12 +448,19 @@ public class SingleNodeKafkaResumeStrategy implements
KafkaResumeStrategy, Camel
}
if (executorService != null) {
- executorService.shutdown();
-
- if (!executorService.awaitTermination(2, TimeUnit.SECONDS)) {
- LOG.warn("Kafka consumer did not shutdown within 2
seconds");
- executorService.shutdownNow();
+ if (camelContext != null) {
+ // also removes the executor from the manager (and JMX)
+
camelContext.getExecutorServiceManager().shutdownGraceful(executorService,
2000);
+ } else {
+ executorService.shutdown();
+
+ if (!executorService.awaitTermination(2,
TimeUnit.SECONDS)) {
+ LOG.warn("Kafka consumer did not shutdown within 2
seconds");
+ executorService.shutdownNow();
+ }
}
+ // a new executor is created when this strategy is started
again
+ executorService = null;
} else {
// This may happen if the start up has failed in some other
part
LOG.trace("There's no executor service to shutdown");
@@ -456,6 +493,10 @@ public class SingleNodeKafkaResumeStrategy implements
KafkaResumeStrategy, Camel
return producer;
}
+ protected ExecutorService getExecutorService() {
+ return executorService;
+ }
+
@Override
public void setResumeStrategyConfiguration(ResumeStrategyConfiguration
resumeStrategyConfiguration) {
if (resumeStrategyConfiguration instanceof
KafkaResumeStrategyConfiguration kafkaResumeStrategyConfiguration) {
diff --git
a/components/camel-kafka/src/test/java/org/apache/camel/processor/resume/kafka/SingleNodeKafkaResumeStrategyTest.java
b/components/camel-kafka/src/test/java/org/apache/camel/processor/resume/kafka/SingleNodeKafkaResumeStrategyTest.java
index 46175720ddad..4d21050feca7 100644
---
a/components/camel-kafka/src/test/java/org/apache/camel/processor/resume/kafka/SingleNodeKafkaResumeStrategyTest.java
+++
b/components/camel-kafka/src/test/java/org/apache/camel/processor/resume/kafka/SingleNodeKafkaResumeStrategyTest.java
@@ -17,13 +17,34 @@
package org.apache.camel.processor.resume.kafka;
import java.lang.reflect.Field;
+import java.nio.ByteBuffer;
+import java.nio.charset.StandardCharsets;
+import java.util.HashSet;
+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.CountDownLatch;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.locks.ReentrantLock;
+import org.apache.camel.CamelContext;
+import org.apache.camel.impl.DefaultCamelContext;
+import org.apache.camel.resume.Deserializable;
import org.apache.camel.resume.ResumeAdapter;
+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.apache.kafka.common.errors.WakeupException;
import org.junit.jupiter.api.Test;
+import static org.awaitility.Awaitility.await;
import static org.junit.jupiter.api.Assertions.*;
/**
@@ -78,6 +99,210 @@ public class SingleNodeKafkaResumeStrategyTest {
assertSame(adapter, strategy.getAdapter());
}
+ @Test
+ void testStrategyCanBeStartedAgainAfterStop() throws Exception {
+ try (CamelContext context = new DefaultCamelContext()) {
+ context.start();
+
+ // nothing listens on port 9: no broker is needed to create,
subscribe and close the clients
+ KafkaResumeStrategyConfiguration config
+ = new KafkaResumeStrategyConfigurationBuilder(
+
KafkaResumeStrategyConfigurationBuilder.createProducerProperties(),
+
KafkaResumeStrategyConfigurationBuilder.createConsumerProperties())
+ .withBootstrapServers("localhost:9")
+ .withGroupId("test-group")
+ .withEnableAutoCommit(false)
+ .withTopic("test-topic")
+ .build();
+
+ SubscriptionRecordingStrategy strategy = new
SubscriptionRecordingStrategy(config);
+ strategy.setCamelContext(context);
+ strategy.setAdapter(new TestDeserializableAdapter());
+
+ try {
+ strategy.start();
+ strategy.loadCache();
+ Producer<byte[], byte[]> firstProducer =
strategy.getProducer();
+ ExecutorService firstExecutor = strategy.getExecutorService();
+ assertTrue(executorsOf(context).contains(firstExecutor));
+ await().atMost(10, TimeUnit.SECONDS).until(() ->
strategy.subscriptions.size() == 1);
+ strategy.stop();
+ // the executor is shut down, and removed from the manager of
the context (and so from JMX)
+ assertTrue(firstExecutor.isShutdown());
+ assertFalse(executorsOf(context).contains(firstExecutor), "the
executor stays registered");
+
+ // a route stop/start stops and starts the strategy again
+ strategy.start();
+ strategy.loadCache();
+ ExecutorService executor = strategy.getExecutorService();
+ assertAll(
+ () -> assertNotNull(strategy.getProducer()),
+ () -> assertNotSame(firstProducer,
strategy.getProducer(), "the closed producer is used again"),
+ () -> assertFalse(executor.isShutdown(), "the executor
that was shut down is used again"));
+
+ // the new consumer subscribes to the topic again
+ await().atMost(10, TimeUnit.SECONDS).untilAsserted(
+ () -> assertEquals(List.of(Set.of("test-topic"),
Set.of("test-topic")), strategy.subscriptions));
+ } finally {
+ strategy.stop();
+ }
+ }
+ }
+
+ @Test
+ void testRefreshTaskThatOutlivesStopClosesOnlyItsOwnConsumer() throws
Exception {
+ try (CamelContext context = new DefaultCamelContext()) {
+ context.start();
+ // the producer is created, and never sends: nothing listens on
port 9
+ KafkaResumeStrategyConfiguration config
+ = new KafkaResumeStrategyConfigurationBuilder(
+
KafkaResumeStrategyConfigurationBuilder.createProducerProperties(),
+
KafkaResumeStrategyConfigurationBuilder.createConsumerProperties())
+ .withBootstrapServers("localhost:9")
+ .withTopic("test-topic")
+ .build();
+
+ // each refresh task gets its own MockConsumer, and blocks in its
first poll until released, ignoring the
+ // wakeup and the interrupt of stop(), so the first task outlives
stop()
+ List<MockConsumer<byte[], byte[]>> consumers = new
CopyOnWriteArrayList<>();
+ List<CountDownLatch> polling = List.of(new CountDownLatch(1), new
CountDownLatch(1));
+ List<CountDownLatch> release = List.of(new CountDownLatch(1), new
CountDownLatch(1));
+ SingleNodeKafkaResumeStrategy strategy = new
SingleNodeKafkaResumeStrategy(config) {
+ @Override
+ Consumer<byte[], byte[]> createConsumer() {
+ MockConsumer<byte[], byte[]> mock = new
MockConsumer<>("earliest");
+ consumers.add(mock);
+ return mock;
+ }
+
+ @Override
+ protected ConsumerRecords<byte[], byte[]>
consume(Consumer<byte[], byte[]> consumer) {
+ int task = consumers.indexOf(consumer);
+ polling.get(task).countDown();
+ awaitUninterruptibly(release.get(task));
+ // ends the polling loop, as a wakeup does
+ throw new WakeupException();
+ }
+ };
+ strategy.setCamelContext(context);
+ strategy.setAdapter(new TestDeserializableAdapter());
+
+ try {
+ strategy.start();
+ strategy.loadCache();
+ assertTrue(polling.get(0).await(10, TimeUnit.SECONDS));
+ // the first task does not end within the shutdown wait of
stop()
+ strategy.stop();
+
+ strategy.start();
+ strategy.loadCache();
+ assertTrue(polling.get(1).await(10, TimeUnit.SECONDS));
+
+ // the first task ends now, and closes its consumer
+ release.get(0).countDown();
+ await().atMost(10, TimeUnit.SECONDS).until(() ->
consumers.get(0).closed());
+ assertFalse(consumers.get(1).closed(), "the old refresh task
closed the consumer of the new one");
+ } finally {
+ release.get(0).countDown();
+ release.get(1).countDown();
+ strategy.stop();
+ }
+ await().atMost(10, TimeUnit.SECONDS).until(() ->
consumers.get(1).closed());
+ }
+ }
+
+ private static void awaitUninterruptibly(CountDownLatch latch) {
+ boolean interrupted = false;
+ try {
+ while (true) {
+ try {
+ if (latch.await(10, TimeUnit.SECONDS)) {
+ return;
+ }
+ throw new IllegalStateException("not released");
+ } catch (InterruptedException e) {
+ interrupted = true;
+ }
+ }
+ } finally {
+ if (interrupted) {
+ Thread.currentThread().interrupt();
+ }
+ }
+ }
+
+ @SuppressWarnings("unchecked")
+ private static List<ExecutorService> executorsOf(CamelContext context)
throws Exception {
+ Class<?> type = context.getExecutorServiceManager().getClass();
+ while (type.getSuperclass() != null) {
+ try {
+ Field field = type.getDeclaredField("executorServices");
+ field.setAccessible(true);
+ return (List<ExecutorService>)
field.get(context.getExecutorServiceManager());
+ } catch (NoSuchFieldException e) {
+ type = type.getSuperclass();
+ }
+ }
+ throw new NoSuchFieldException("executorServices");
+ }
+
+ @Test
+ void testRecordThatCannotBeDeserializedIsSkipped() {
+ TestDeserializableAdapter adapter = new TestDeserializableAdapter();
+ SingleNodeKafkaResumeStrategy strategy = new
SingleNodeKafkaResumeStrategy() {
+ private int polls;
+
+ @Override
+ protected ConsumerRecords<byte[], byte[]> consume(Consumer<byte[],
byte[]> consumer) {
+ if (polls++ > 0) {
+ // ends the polling loop, as stop() does
+ throw new WakeupException();
+ }
+ byte[] data = "data".getBytes(StandardCharsets.UTF_8);
+ return new ConsumerRecords<>(
+ Map.of(new TopicPartition("test-topic", 0),
+ List.of(new ConsumerRecord<>("test-topic", 0,
0L, null, data),
+ // a tombstone: Camel does not write
them, so it is skipped
+ new ConsumerRecord<>("test-topic", 0,
1L, data, null),
+ new ConsumerRecord<>("test-topic", 0,
2L, data, data))),
+ Map.of());
+ }
+ };
+ strategy.setAdapter(adapter);
+
+ assertThrows(WakeupException.class, () -> strategy.poll(null, new
CountDownLatch(1)));
+ // the record without a key and the tombstone are skipped, the next
record is still loaded
+ assertEquals(1, adapter.deserialized.get());
+ }
+
+ static class SubscriptionRecordingStrategy extends
SingleNodeKafkaResumeStrategy {
+ final List<Set<String>> subscriptions = new CopyOnWriteArrayList<>();
+
+ SubscriptionRecordingStrategy(KafkaResumeStrategyConfiguration config)
{
+ super(config);
+ }
+
+ @Override
+ protected void checkAndSubscribe(Consumer<byte[], byte[]> consumer,
String topic) {
+ super.checkAndSubscribe(consumer, topic);
+ subscriptions.add(new HashSet<>(consumer.subscription()));
+ }
+ }
+
+ static class TestDeserializableAdapter implements ResumeAdapter,
Deserializable {
+ final AtomicInteger deserialized = new AtomicInteger();
+
+ @Override
+ public void resume() {
+ }
+
+ @Override
+ public boolean deserialize(ByteBuffer keyBuffer, ByteBuffer
valueBuffer) {
+ deserialized.incrementAndGet();
+ return true;
+ }
+ }
+
static class TestAdapter implements ResumeAdapter {
@Override
public void resume() {
