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 <[email protected]> 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 <[email protected]> Signed-off-by: Claus Ibsen <[email protected]> --- .../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
