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 20c94435789e CAMEL-25872: camel-kafka-share - consume again after the
CamelContext is suspended and resumed (#28386)
20c94435789e is described below
commit 20c94435789eb2efa3c36d31cafb85fc7d520379
Author: allthingssecurity <[email protected]>
AuthorDate: Sat Oct 24 19:12:18 2026 +0530
CAMEL-25872: camel-kafka-share - consume again after the CamelContext is
suspended and resumed (#28386)
Co-Authored-By: Claude Opus 5.5 <[email protected]>
---
.../component/kafka/AbstractKafkaComponent.java | 55 +++++++++-
.../component/kafka/share/KafkaShareConsumer.java | 10 +-
.../share/KafkaShareConsumerSuspendResumeTest.java | 119 +++++++++++++++++++++
.../camel/component/kafka/KafkaConsumer.java | 10 +-
4 files changed, 176 insertions(+), 18 deletions(-)
diff --git
a/components/camel-kafka-common/src/main/java/org/apache/camel/component/kafka/AbstractKafkaComponent.java
b/components/camel-kafka-common/src/main/java/org/apache/camel/component/kafka/AbstractKafkaComponent.java
index 471f86be982f..3e40d4a66f86 100644
---
a/components/camel-kafka-common/src/main/java/org/apache/camel/component/kafka/AbstractKafkaComponent.java
+++
b/components/camel-kafka-common/src/main/java/org/apache/camel/component/kafka/AbstractKafkaComponent.java
@@ -20,6 +20,8 @@ import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.CopyOnWriteArrayList;
+import java.util.concurrent.locks.Lock;
+import java.util.concurrent.locks.ReentrantLock;
import org.apache.camel.CamelContext;
import org.apache.camel.ExtendedStartupListener;
@@ -41,6 +43,10 @@ public abstract class AbstractKafkaComponent extends
HealthCheckComponent
private static final Logger LOG =
LoggerFactory.getLogger(AbstractKafkaComponent.class);
private final List<Runnable> pendingConsumers = new
CopyOnWriteArrayList<>();
+ private final Lock pendingConsumersLock = new ReentrantLock();
+ // whether the CamelContext has been started since this component was
started, which a CamelContext that resumes
+ // does not notify again
+ private boolean pendingConsumersStarted;
@Metadata(label = "security", defaultValue = "false", description =
"Enable usage of global SSL context parameters.")
private boolean useGlobalSslContextParameters;
@@ -110,6 +116,29 @@ public abstract class AbstractKafkaComponent extends
HealthCheckComponent
pendingConsumers.add(task);
}
+ /**
+ * Starts the task now if the CamelContext has been started, or else once
the CamelContext is started, so that a
+ * consumer does not process messages before the CamelContext is fully
started.
+ * <p/>
+ * Unlike checking only whether the CamelContext is started, this also
starts at once a consumer that is started
+ * while the CamelContext resumes (a suspend of the CamelContext stops the
consumers that cannot be suspended, and
+ * its resume starts them again), as the CamelContext does not notify that
it is started again after a resume.
+ */
+ public void startConsumer(Runnable task) {
+ pendingConsumersLock.lock();
+ try {
+ // the CamelContext is also checked, as a component added with
addComponent to a started CamelContext is
+ // not notified that the CamelContext started
+ if (pendingConsumersStarted || getCamelContext().isStarted()) {
+ task.run();
+ } else {
+ pendingConsumers.add(task);
+ }
+ } finally {
+ pendingConsumersLock.unlock();
+ }
+ }
+
@Override
public boolean isUseGlobalSslContextParameters() {
return this.useGlobalSslContextParameters;
@@ -179,10 +208,28 @@ public abstract class AbstractKafkaComponent extends
HealthCheckComponent
}
private void startPendingConsumers() {
- if (!pendingConsumers.isEmpty()) {
- LOG.info("Starting {} pending Kafka consumers as CamelContext is
fully started", pendingConsumers.size());
- pendingConsumers.forEach(Runnable::run);
- pendingConsumers.clear();
+ pendingConsumersLock.lock();
+ try {
+ pendingConsumersStarted = true;
+ if (!pendingConsumers.isEmpty()) {
+ LOG.info("Starting {} pending Kafka consumers as CamelContext
is fully started", pendingConsumers.size());
+ pendingConsumers.forEach(Runnable::run);
+ pendingConsumers.clear();
+ }
+ } finally {
+ pendingConsumersLock.unlock();
+ }
+ }
+
+ @Override
+ protected void doStop() throws Exception {
+ super.doStop();
+ pendingConsumersLock.lock();
+ try {
+ // the consumers started after the CamelContext is started again
wait for it
+ pendingConsumersStarted = false;
+ } finally {
+ pendingConsumersLock.unlock();
}
}
diff --git
a/components/camel-kafka-share/src/main/java/org/apache/camel/component/kafka/share/KafkaShareConsumer.java
b/components/camel-kafka-share/src/main/java/org/apache/camel/component/kafka/share/KafkaShareConsumer.java
index abba13876df2..ae6749392178 100644
---
a/components/camel-kafka-share/src/main/java/org/apache/camel/component/kafka/share/KafkaShareConsumer.java
+++
b/components/camel-kafka-share/src/main/java/org/apache/camel/component/kafka/share/KafkaShareConsumer.java
@@ -106,13 +106,9 @@ public class KafkaShareConsumer extends DefaultConsumer {
BridgeExceptionHandlerToErrorHandler bridge = new
BridgeExceptionHandlerToErrorHandler(this);
for (int i = 0; i < configuration.getConsumersCount(); i++) {
KafkaShareFetchRecords task = new KafkaShareFetchRecords(this,
bridge, Integer.toString(i), getProps());
- if (!endpoint.getCamelContext().isStarted()) {
- // if camel has not been fully started yet then delay starting
this consumer to avoid
- // process incoming message before camel is fully started
- endpoint.getComponent().pendingConsumer(() ->
executor.submit(task));
- } else {
- executor.submit(task);
- }
+ // if camel has not been fully started yet then delay starting
this consumer to avoid
+ // process incoming message before camel is fully started
+ endpoint.getComponent().startConsumer(() -> executor.submit(task));
tasks.add(task);
}
}
diff --git
a/components/camel-kafka-share/src/test/java/org/apache/camel/component/kafka/share/KafkaShareConsumerSuspendResumeTest.java
b/components/camel-kafka-share/src/test/java/org/apache/camel/component/kafka/share/KafkaShareConsumerSuspendResumeTest.java
new file mode 100644
index 000000000000..b4dbbbd189f1
--- /dev/null
+++
b/components/camel-kafka-share/src/test/java/org/apache/camel/component/kafka/share/KafkaShareConsumerSuspendResumeTest.java
@@ -0,0 +1,119 @@
+/*
+ * 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.share;
+
+import java.util.List;
+import java.util.Properties;
+import java.util.concurrent.CopyOnWriteArrayList;
+import java.util.concurrent.TimeUnit;
+
+import org.apache.camel.BindToRegistry;
+import org.apache.camel.ServiceStatus;
+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.ShareConsumer;
+import org.junit.jupiter.api.Test;
+
+import static
org.apache.camel.component.kafka.share.RecordingShareConsumer.record;
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.awaitility.Awaitility.await;
+
+/**
+ * The share consumer, which does not support suspension, is stopped when the
CamelContext is suspended and must consume
+ * again when the CamelContext is resumed. A consumer of a started
CamelContext starts at once.
+ */
+class KafkaShareConsumerSuspendResumeTest extends CamelTestSupport {
+
+ private static final String TOPIC = "orders";
+
+ private final List<RecordingShareConsumer> consumers = new
CopyOnWriteArrayList<>();
+
+ @BindToRegistry("recordingFactory")
+ private final KafkaShareClientFactory factory = new
KafkaShareClientFactory() {
+ @Override
+ public ShareConsumer<Object, Object> getShareConsumer(Properties
kafkaProps) {
+ RecordingShareConsumer consumer = new RecordingShareConsumer();
+ consumers.add(consumer);
+ return consumer;
+ }
+
+ @Override
+ public String getBrokers(KafkaShareConfiguration configuration) {
+ return "localhost:9092";
+ }
+ };
+
+ @Override
+ protected RouteBuilder createRouteBuilder() {
+ return new RouteBuilder() {
+ @Override
+ public void configure() {
+ from("kafka-share:" + TOPIC +
"?brokers=localhost:9092&groupId=resume&pollTimeoutMs=100").routeId("share")
+ .to("mock:result");
+ }
+ };
+ }
+
+ @Test
+ void consumesAgainAfterCamelContextIsResumed() throws Exception {
+ RecordingShareConsumer first = subscribedConsumer(0);
+
+ context.suspend();
+ await().atMost(10, TimeUnit.SECONDS).until(first::isClosed);
+
+ context.resume();
+
assertThat(context.getRouteController().getRouteStatus("share")).isEqualTo(ServiceStatus.Started);
+
+ MockEndpoint result = getMockEndpoint("mock:result");
+ result.expectedBodiesReceived("after resume");
+
+ subscribedConsumer(1).addRecord(record(TOPIC, 0, null, "after resume",
(short) 1));
+
+ result.assertIsSatisfied();
+ }
+
+ @Test
+ void consumesWithAComponentAddedAfterCamelContextIsStarted() throws
Exception {
+ subscribedConsumer(0);
+
+ // a component added to a started CamelContext is not notified that
the CamelContext started
+ KafkaShareComponent component = new KafkaShareComponent();
+ component.setKafkaShareClientFactory(factory);
+ context.addComponent("kafka-share-late", component);
+ context.addRoutes(new RouteBuilder() {
+ @Override
+ public void configure() {
+ from("kafka-share-late:" + TOPIC +
"?brokers=localhost:9092&groupId=late&pollTimeoutMs=100")
+ .routeId("late").to("mock:late");
+ }
+ });
+
+ MockEndpoint late = getMockEndpoint("mock:late");
+ late.expectedBodiesReceived("late record");
+
+ subscribedConsumer(1).addRecord(record(TOPIC, 0, null, "late record",
(short) 1));
+
+ late.assertIsSatisfied();
+ }
+
+ private RecordingShareConsumer subscribedConsumer(int index) {
+ await().atMost(10, TimeUnit.SECONDS)
+ .until(() -> consumers.size() > index &&
consumers.get(index).subscription().contains(TOPIC));
+ return consumers.get(index);
+ }
+}
diff --git
a/components/camel-kafka/src/main/java/org/apache/camel/component/kafka/KafkaConsumer.java
b/components/camel-kafka/src/main/java/org/apache/camel/component/kafka/KafkaConsumer.java
index c1bdfe08c240..000252c6d74b 100644
---
a/components/camel-kafka/src/main/java/org/apache/camel/component/kafka/KafkaConsumer.java
+++
b/components/camel-kafka/src/main/java/org/apache/camel/component/kafka/KafkaConsumer.java
@@ -167,13 +167,9 @@ public class KafkaConsumer extends DefaultConsumer
KafkaFetchRecords task = new KafkaFetchRecords(
this, bridge, topic, pattern, Integer.toString(i),
getProps(), consumerListener);
- if (!endpoint.getCamelContext().isStarted()) {
- // if camel has not been fully started yet then delay starting
this consumer to avoid
- // process incoming message before camel is fully started
- endpoint.getComponent().pendingConsumer(() ->
executor.submit(task));
- } else {
- executor.submit(task);
- }
+ // if camel has not been fully started yet then delay starting
this consumer to avoid
+ // process incoming message before camel is fully started
+ endpoint.getComponent().startConsumer(() -> executor.submit(task));
tasks.add(task);
}
