You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@camel.apache.org by da...@apache.org on 2026/07/27 13:08:20 UTC
(camel) branch fix/CAMEL-24271 created (now 7fb56b6a543c)
This is an automated email from the ASF dual-hosted git repository.
davsclaus pushed a change to branch fix/CAMEL-24271
in repository https://gitbox.apache.org/repos/asf/camel.git
at 7fb56b6a543c CAMEL-24271: camel-kafka - Use BackgroundTask for reconnection visibility
This branch includes the following new commits:
new 7fb56b6a543c CAMEL-24271: camel-kafka - Use BackgroundTask for reconnection visibility
The 1 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
(camel) 01/01: CAMEL-24271: camel-kafka - Use BackgroundTask for reconnection visibility
Posted by da...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
davsclaus pushed a commit to branch fix/CAMEL-24271
in repository https://gitbox.apache.org/repos/asf/camel.git
commit 7fb56b6a543cfd0d4742db0751a25999c9276ac5
Author: Claus Ibsen <cl...@gmail.com>
AuthorDate: Mon Jul 27 15:08:05 2026 +0200
CAMEL-24271: camel-kafka - Use BackgroundTask for reconnection visibility
Switch from ForegroundTask to BackgroundTask so that Kafka consumer
reconnection stays registered in the internal task registry for the full
duration, making it visible via TUI Internal Tasks tab and CLI.
Also fix IllegalMonitorStateException in SingleNodeKafkaResumeStrategyTest
where a releaser thread tried to unlock a ReentrantLock owned by another thread.
Co-Authored-By: Claude Opus 4.6 <no...@anthropic.com>
Signed-off-by: Claus Ibsen <cl...@gmail.com>
---
.../camel/component/kafka/KafkaFetchRecords.java | 100 +++++++--------------
.../kafka/SingleNodeKafkaResumeStrategyTest.java | 5 --
2 files changed, 31 insertions(+), 74 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 493fa245fa71..4eb36383adb4 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
@@ -19,6 +19,7 @@ package org.apache.camel.component.kafka;
import java.time.Duration;
import java.util.Collection;
import java.util.Properties;
+import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicReference;
import java.util.concurrent.locks.ReentrantLock;
@@ -43,7 +44,7 @@ import org.apache.camel.component.kafka.consumer.support.subcription.DefaultSubs
import org.apache.camel.component.kafka.consumer.support.subcription.SubscribeAdapter;
import org.apache.camel.component.kafka.consumer.support.subcription.TopicInfo;
import org.apache.camel.support.BridgeExceptionHandlerToErrorHandler;
-import org.apache.camel.support.task.ForegroundTask;
+import org.apache.camel.support.task.BackgroundTask;
import org.apache.camel.support.task.TaskRunFailureException;
import org.apache.camel.support.task.Tasks;
import org.apache.camel.support.task.budget.Budgets;
@@ -102,6 +103,7 @@ public class KafkaFetchRecords implements Runnable {
private volatile boolean connected; // this is the state (connected or not)
private final AtomicReference<State> state = new AtomicReference<>(State.RUNNING);
+ private ScheduledExecutorService reconnectPool;
private final DevConsoleMetricsCollector metricsCollector;
KafkaFetchRecords(KafkaConsumer kafkaConsumer,
@@ -139,41 +141,27 @@ public class KafkaFetchRecords implements Runnable {
safeConsumerClose();
}
- // task that deals with creating kafka consumer
+ // background task that creates and subscribes the kafka consumer
+ // this stays registered in the internal task registry for visibility via TUI/CLI
currentBackoffInterval = kafkaConsumer.getEndpoint().getComponent().getCreateConsumerBackoffInterval();
- ForegroundTask task = Tasks.foregroundTask()
- .withName("Create KafkaConsumer")
- .withBudget(Budgets.iterationBudget()
- .withMaxIterations(
- kafkaConsumer.getEndpoint().getComponent().getCreateConsumerBackoffMaxAttempts())
- .withInitialDelay(Duration.ZERO)
- .withInterval(Duration.ofMillis(currentBackoffInterval))
- .build())
- .build();
- boolean success = task.run(kafkaConsumer.getEndpoint().getCamelContext(), this::createConsumerTask);
- if (!success) {
- int max = kafkaConsumer.getEndpoint().getComponent().getCreateConsumerBackoffMaxAttempts();
- setupCreateConsumerException(task, max);
- // give up and terminate this consumer
- terminated = true;
- break;
+ int maxAttempts = kafkaConsumer.getEndpoint().getComponent().getCreateConsumerBackoffMaxAttempts();
+ if (reconnectPool == null) {
+ reconnectPool = kafkaConsumer.getEndpoint().getCamelContext().getExecutorServiceManager()
+ .newSingleThreadScheduledExecutor(this, "KafkaReconnect");
}
-
- // task that deals with subscribing kafka consumer
- currentBackoffInterval = kafkaConsumer.getEndpoint().getComponent().getSubscribeConsumerBackoffInterval();
- task = Tasks.foregroundTask()
- .withName("Subscribe KafkaConsumer")
- .withBudget(Budgets.iterationBudget()
- .withMaxIterations(
- kafkaConsumer.getEndpoint().getComponent().getSubscribeConsumerBackoffMaxAttempts())
- .withInitialDelay(Duration.ZERO)
+ BackgroundTask task = Tasks.backgroundTask()
+ .withScheduledExecutor(reconnectPool)
+ .withBudget(Budgets.iterationTimeBudget()
+ .withMaxIterations(maxAttempts)
.withInterval(Duration.ofMillis(currentBackoffInterval))
+ .withInitialDelay(Duration.ZERO)
+ .withUnlimitedDuration()
.build())
+ .withName("KafkaReconnect-" + getPrintableTopic())
.build();
- success = task.run(kafkaConsumer.getEndpoint().getCamelContext(), this::initializeConsumerTask);
+ boolean success = task.run(kafkaConsumer.getEndpoint().getCamelContext(), this::reconnectTask);
if (!success) {
- int max = kafkaConsumer.getEndpoint().getComponent().getSubscribeConsumerBackoffMaxAttempts();
- setupInitializeErrorException(task, max);
+ setupReconnectException(task, maxAttempts);
// give up and terminate this consumer
terminated = true;
break;
@@ -195,50 +183,25 @@ public class KafkaFetchRecords implements Runnable {
}
safeConsumerClose();
- }
- private void setupInitializeErrorException(ForegroundTask task, int max) {
- String time = TimeUtils.printDuration(task.elapsed(), true);
- String topic = getPrintableTopic();
- String msg = "Gave up subscribing org.apache.kafka.clients.consumer.KafkaConsumer " +
- threadId + " to " + topic + " after " + max + " attempts (elapsed: " + time + ").";
- LOG.warn(msg);
- setLastError(new KafkaConsumerFatalException(msg, lastError));
+ if (reconnectPool != null) {
+ kafkaConsumer.getEndpoint().getCamelContext().getExecutorServiceManager().shutdown(reconnectPool);
+ reconnectPool = null;
+ }
}
- private void setupCreateConsumerException(ForegroundTask task, int max) {
+ private void setupReconnectException(BackgroundTask task, int max) {
String time = TimeUtils.printDuration(task.elapsed(), true);
String topic = getPrintableTopic();
- String msg = "Gave up creating org.apache.kafka.clients.consumer.KafkaConsumer "
+ String msg = "Gave up creating/subscribing org.apache.kafka.clients.consumer.KafkaConsumer "
+ threadId + " to " + topic + " after " + max + " attempts (elapsed: " + time + ").";
-
+ LOG.warn(msg);
setLastError(new KafkaConsumerFatalException(msg, lastError));
}
- private boolean initializeConsumerTask() {
- try {
- initializeConsumer();
- } catch (Exception e) {
- setConnected(false);
- // ensure this is logged so users can see the problem
- LOG.warn("Error subscribing org.apache.kafka.clients.consumer.KafkaConsumer due to: {}", e.getMessage(),
- e);
- setLastError(e);
-
- // allow camel error handler to be aware
- if (kafkaConsumer.getEndpoint().isBridgeErrorHandler()) {
- kafkaConsumer.getExceptionHandler().handleException(e);
- }
-
- // make the task runner aware of the exception (will retry)
- throw new TaskRunFailureException(e);
- }
-
- return true;
- }
-
- private boolean createConsumerTask() {
+ private boolean reconnectTask() {
try {
+ safeConsumerClose();
createConsumer();
commitManager
= CommitManagers.createCommitManager(consumer, kafkaConsumer, threadId, getPrintableTopic());
@@ -256,19 +219,18 @@ public class KafkaFetchRecords implements Runnable {
consumerListener.setSeekPolicy(seekPolicy);
}
+
+ initializeConsumer();
} catch (Exception e) {
setConnected(false);
- // ensure this is logged so users can see the problem
- LOG.warn("Error creating org.apache.kafka.clients.consumer.KafkaConsumer due to: {}", e.getMessage(),
- e);
+ LOG.warn("Error creating/subscribing org.apache.kafka.clients.consumer.KafkaConsumer due to: {}",
+ e.getMessage(), e);
setLastError(e);
- // allow camel error handler to be aware
if (kafkaConsumer.getEndpoint().isBridgeErrorHandler()) {
kafkaConsumer.getExceptionHandler().handleException(e);
}
- // make the task runner aware of the exception (will retry)
throw new TaskRunFailureException(e);
}
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 521152ace7c9..46175720ddad 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
@@ -43,11 +43,6 @@ public class SingleNodeKafkaResumeStrategyTest {
// stop() should not throw IllegalMonitorStateException
assertDoesNotThrow(strategy::stop);
-
- // Release the lock from the holder thread
- Thread releaser = new Thread(lock::unlock);
- releaser.start();
- releaser.join();
}
@Test