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 e61ba08a726a CAMEL-24271: camel-kafka - Use BackgroundTask for
reconnection visibility
e61ba08a726a is described below
commit e61ba08a726a93bd75485c16d350f15b6e3ecc7d
Author: Claus Ibsen <[email protected]>
AuthorDate: Mon Jul 27 19:50:19 2026 +0200
CAMEL-24271: camel-kafka - Use BackgroundTask for reconnection visibility
Switch Kafka consumer reconnection from two separate ForegroundTasks
(create + subscribe) to a single BackgroundTask, making it visible via
Camel management tooling (TUI, CLI, Hawtio). Deprecate the now-redundant
subscribeConsumerBackoff* options in favor of createConsumerBackoff*.
Also fix IllegalMonitorStateException in SingleNodeKafkaResumeStrategyTest.
Closes #25156
Co-Authored-By: Claude Opus 4.6 <[email protected]>
Signed-off-by: Claus Ibsen <[email protected]>
---
.../org/apache/camel/catalog/components/kafka.json | 6 +-
.../org/apache/camel/component/kafka/kafka.json | 6 +-
.../camel/component/kafka/KafkaComponent.java | 27 +++-
.../camel/component/kafka/KafkaFetchRecords.java | 159 ++++++++-------------
.../kafka/SingleNodeKafkaResumeStrategyTest.java | 5 -
.../ROOT/pages/camel-4x-upgrade-guide-4_22.adoc | 16 +++
.../dsl/KafkaComponentBuilderFactory.java | 6 +-
7 files changed, 114 insertions(+), 111 deletions(-)
diff --git
a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/components/kafka.json
b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/components/kafka.json
index 20079d364ed0..58033c0f5352 100644
---
a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/components/kafka.json
+++
b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/components/kafka.json
@@ -74,9 +74,9 @@
"isolationLevel": { "index": 47, "kind": "property", "displayName":
"Isolation Level", "group": "consumer (advanced)", "label":
"consumer,advanced", "required": false, "type": "enum", "javaType":
"java.lang.String", "enum": [ "read_uncommitted", "read_committed" ],
"deprecated": false, "autowired": false, "secret": false, "defaultValue":
"read_uncommitted", "configurationClass":
"org.apache.camel.component.kafka.KafkaConfiguration", "configurationField":
"configuration", "description [...]
"kafkaManualCommitFactory": { "index": 48, "kind": "property",
"displayName": "Kafka Manual Commit Factory", "group": "consumer (advanced)",
"label": "consumer,advanced", "required": false, "type": "object", "javaType":
"org.apache.camel.component.kafka.consumer.KafkaManualCommitFactory",
"deprecated": false, "autowired": true, "secret": false, "description":
"Factory to use for creating KafkaManualCommit instances. This allows to plugin
a custom factory to create custom KafkaManualC [...]
"pollExceptionStrategy": { "index": 49, "kind": "property", "displayName":
"Poll Exception Strategy", "group": "consumer (advanced)", "label":
"consumer,advanced", "required": false, "type": "object", "javaType":
"org.apache.camel.component.kafka.PollExceptionStrategy", "deprecated": false,
"autowired": true, "secret": false, "description": "To use a custom strategy
with the consumer to control how to handle exceptions thrown from the Kafka
broker while pooling messages." },
- "subscribeConsumerBackoffInterval": { "index": 50, "kind": "property",
"displayName": "Subscribe Consumer Backoff Interval", "group": "consumer
(advanced)", "label": "consumer,advanced", "required": false, "type":
"integer", "javaType": "long", "deprecated": false, "autowired": false,
"secret": false, "defaultValue": 5000, "description": "The delay in millis
seconds to wait before trying again to subscribe to the kafka broker." },
- "subscribeConsumerBackoffMaxAttempts": { "index": 51, "kind": "property",
"displayName": "Subscribe Consumer Backoff Max Attempts", "group": "consumer
(advanced)", "label": "consumer,advanced", "required": false, "type":
"integer", "javaType": "int", "deprecated": false, "autowired": false,
"secret": false, "description": "Maximum number the kafka consumer will attempt
to subscribe to the kafka broker, before eventually giving up and failing.
Error during subscribing the consumer to [...]
- "subscribeConsumerTopicMustExists": { "index": 52, "kind": "property",
"displayName": "Subscribe Consumer Topic Must Exists", "group": "consumer
(advanced)", "label": "consumer,advanced", "required": false, "type":
"boolean", "javaType": "boolean", "deprecated": false, "autowired": false,
"secret": false, "defaultValue": false, "description": "Whether when a Camel
Kafka consumer is subscribing to a Kafka broker then check whether a topic
already exist on the broker, and fail if it do [...]
+ "subscribeConsumerBackoffInterval": { "index": 50, "kind": "property",
"displayName": "Subscribe Consumer Backoff Interval", "group": "consumer
(advanced)", "label": "consumer,advanced", "required": false, "type":
"integer", "javaType": "long", "deprecated": true, "autowired": false,
"secret": false, "defaultValue": 5000, "description": "The delay in millis
seconds to wait before trying again to subscribe to the kafka broker." },
+ "subscribeConsumerBackoffMaxAttempts": { "index": 51, "kind": "property",
"displayName": "Subscribe Consumer Backoff Max Attempts", "group": "consumer
(advanced)", "label": "consumer,advanced", "required": false, "type":
"integer", "javaType": "int", "deprecated": true, "autowired": false, "secret":
false, "description": "Maximum number the kafka consumer will attempt to
subscribe to the kafka broker, before eventually giving up and failing. Error
during subscribing the consumer to t [...]
+ "subscribeConsumerTopicMustExists": { "index": 52, "kind": "property",
"displayName": "Subscribe Consumer Topic Must Exists", "group": "consumer
(advanced)", "label": "consumer,advanced", "required": false, "type":
"boolean", "javaType": "boolean", "deprecated": false, "autowired": false,
"secret": false, "defaultValue": false, "description": "Whether when a Camel
Kafka consumer is subscribing to a Kafka broker then check whether a topic
already exist on the broker, and fail if it do [...]
"batchWithIndividualHeaders": { "index": 53, "kind": "property",
"displayName": "Batch With Individual Headers", "group": "producer", "label":
"producer", "required": false, "type": "boolean", "javaType": "boolean",
"deprecated": false, "autowired": false, "secret": false, "defaultValue":
false, "configurationClass":
"org.apache.camel.component.kafka.KafkaConfiguration", "configurationField":
"configuration", "description": "If this feature is enabled and a single
element of a batch [...]
"bufferMemorySize": { "index": 54, "kind": "property", "displayName":
"Buffer Memory Size", "group": "producer", "label": "producer", "required":
false, "type": "integer", "javaType": "java.lang.Integer", "deprecated": false,
"autowired": false, "secret": false, "defaultValue": 33554432,
"configurationClass": "org.apache.camel.component.kafka.KafkaConfiguration",
"configurationField": "configuration", "description": "The total bytes of
memory the producer can use to buffer records wa [...]
"compressionCodec": { "index": 55, "kind": "property", "displayName":
"Compression Codec", "group": "producer", "label": "producer", "required":
false, "type": "enum", "javaType": "java.lang.String", "enum": [ "none",
"gzip", "snappy", "lz4", "zstd" ], "deprecated": false, "autowired": false,
"secret": false, "defaultValue": "none", "configurationClass":
"org.apache.camel.component.kafka.KafkaConfiguration", "configurationField":
"configuration", "description": "This parameter allows [...]
diff --git
a/components/camel-kafka/src/generated/resources/META-INF/org/apache/camel/component/kafka/kafka.json
b/components/camel-kafka/src/generated/resources/META-INF/org/apache/camel/component/kafka/kafka.json
index 20079d364ed0..58033c0f5352 100644
---
a/components/camel-kafka/src/generated/resources/META-INF/org/apache/camel/component/kafka/kafka.json
+++
b/components/camel-kafka/src/generated/resources/META-INF/org/apache/camel/component/kafka/kafka.json
@@ -74,9 +74,9 @@
"isolationLevel": { "index": 47, "kind": "property", "displayName":
"Isolation Level", "group": "consumer (advanced)", "label":
"consumer,advanced", "required": false, "type": "enum", "javaType":
"java.lang.String", "enum": [ "read_uncommitted", "read_committed" ],
"deprecated": false, "autowired": false, "secret": false, "defaultValue":
"read_uncommitted", "configurationClass":
"org.apache.camel.component.kafka.KafkaConfiguration", "configurationField":
"configuration", "description [...]
"kafkaManualCommitFactory": { "index": 48, "kind": "property",
"displayName": "Kafka Manual Commit Factory", "group": "consumer (advanced)",
"label": "consumer,advanced", "required": false, "type": "object", "javaType":
"org.apache.camel.component.kafka.consumer.KafkaManualCommitFactory",
"deprecated": false, "autowired": true, "secret": false, "description":
"Factory to use for creating KafkaManualCommit instances. This allows to plugin
a custom factory to create custom KafkaManualC [...]
"pollExceptionStrategy": { "index": 49, "kind": "property", "displayName":
"Poll Exception Strategy", "group": "consumer (advanced)", "label":
"consumer,advanced", "required": false, "type": "object", "javaType":
"org.apache.camel.component.kafka.PollExceptionStrategy", "deprecated": false,
"autowired": true, "secret": false, "description": "To use a custom strategy
with the consumer to control how to handle exceptions thrown from the Kafka
broker while pooling messages." },
- "subscribeConsumerBackoffInterval": { "index": 50, "kind": "property",
"displayName": "Subscribe Consumer Backoff Interval", "group": "consumer
(advanced)", "label": "consumer,advanced", "required": false, "type":
"integer", "javaType": "long", "deprecated": false, "autowired": false,
"secret": false, "defaultValue": 5000, "description": "The delay in millis
seconds to wait before trying again to subscribe to the kafka broker." },
- "subscribeConsumerBackoffMaxAttempts": { "index": 51, "kind": "property",
"displayName": "Subscribe Consumer Backoff Max Attempts", "group": "consumer
(advanced)", "label": "consumer,advanced", "required": false, "type":
"integer", "javaType": "int", "deprecated": false, "autowired": false,
"secret": false, "description": "Maximum number the kafka consumer will attempt
to subscribe to the kafka broker, before eventually giving up and failing.
Error during subscribing the consumer to [...]
- "subscribeConsumerTopicMustExists": { "index": 52, "kind": "property",
"displayName": "Subscribe Consumer Topic Must Exists", "group": "consumer
(advanced)", "label": "consumer,advanced", "required": false, "type":
"boolean", "javaType": "boolean", "deprecated": false, "autowired": false,
"secret": false, "defaultValue": false, "description": "Whether when a Camel
Kafka consumer is subscribing to a Kafka broker then check whether a topic
already exist on the broker, and fail if it do [...]
+ "subscribeConsumerBackoffInterval": { "index": 50, "kind": "property",
"displayName": "Subscribe Consumer Backoff Interval", "group": "consumer
(advanced)", "label": "consumer,advanced", "required": false, "type":
"integer", "javaType": "long", "deprecated": true, "autowired": false,
"secret": false, "defaultValue": 5000, "description": "The delay in millis
seconds to wait before trying again to subscribe to the kafka broker." },
+ "subscribeConsumerBackoffMaxAttempts": { "index": 51, "kind": "property",
"displayName": "Subscribe Consumer Backoff Max Attempts", "group": "consumer
(advanced)", "label": "consumer,advanced", "required": false, "type":
"integer", "javaType": "int", "deprecated": true, "autowired": false, "secret":
false, "description": "Maximum number the kafka consumer will attempt to
subscribe to the kafka broker, before eventually giving up and failing. Error
during subscribing the consumer to t [...]
+ "subscribeConsumerTopicMustExists": { "index": 52, "kind": "property",
"displayName": "Subscribe Consumer Topic Must Exists", "group": "consumer
(advanced)", "label": "consumer,advanced", "required": false, "type":
"boolean", "javaType": "boolean", "deprecated": false, "autowired": false,
"secret": false, "defaultValue": false, "description": "Whether when a Camel
Kafka consumer is subscribing to a Kafka broker then check whether a topic
already exist on the broker, and fail if it do [...]
"batchWithIndividualHeaders": { "index": 53, "kind": "property",
"displayName": "Batch With Individual Headers", "group": "producer", "label":
"producer", "required": false, "type": "boolean", "javaType": "boolean",
"deprecated": false, "autowired": false, "secret": false, "defaultValue":
false, "configurationClass":
"org.apache.camel.component.kafka.KafkaConfiguration", "configurationField":
"configuration", "description": "If this feature is enabled and a single
element of a batch [...]
"bufferMemorySize": { "index": 54, "kind": "property", "displayName":
"Buffer Memory Size", "group": "producer", "label": "producer", "required":
false, "type": "integer", "javaType": "java.lang.Integer", "deprecated": false,
"autowired": false, "secret": false, "defaultValue": 33554432,
"configurationClass": "org.apache.camel.component.kafka.KafkaConfiguration",
"configurationField": "configuration", "description": "The total bytes of
memory the producer can use to buffer records wa [...]
"compressionCodec": { "index": 55, "kind": "property", "displayName":
"Compression Codec", "group": "producer", "label": "producer", "required":
false, "type": "enum", "javaType": "java.lang.String", "enum": [ "none",
"gzip", "snappy", "lz4", "zstd" ], "deprecated": false, "autowired": false,
"secret": false, "defaultValue": "none", "configurationClass":
"org.apache.camel.component.kafka.KafkaConfiguration", "configurationField":
"configuration", "description": "This parameter allows [...]
diff --git
a/components/camel-kafka/src/main/java/org/apache/camel/component/kafka/KafkaComponent.java
b/components/camel-kafka/src/main/java/org/apache/camel/component/kafka/KafkaComponent.java
index 9a8413d368f4..3883d192bb20 100644
---
a/components/camel-kafka/src/main/java/org/apache/camel/component/kafka/KafkaComponent.java
+++
b/components/camel-kafka/src/main/java/org/apache/camel/component/kafka/KafkaComponent.java
@@ -56,8 +56,10 @@ public class KafkaComponent extends HealthCheckComponent
implements SSLContextPa
private int createConsumerBackoffMaxAttempts;
@Metadata(label = "consumer,advanced", defaultValue = "5000")
private long createConsumerBackoffInterval = 5000;
+ @Deprecated
@Metadata(label = "consumer,advanced")
private int subscribeConsumerBackoffMaxAttempts;
+ @Deprecated
@Metadata(label = "consumer,advanced", defaultValue = "5000")
private long subscribeConsumerBackoffInterval = 5000;
@Metadata(label = "consumer,advanced")
@@ -207,6 +209,12 @@ public class KafkaComponent extends HealthCheckComponent
implements SSLContextPa
this.createConsumerBackoffInterval = createConsumerBackoffInterval;
}
+ /**
+ * @deprecated Use {@link #getCreateConsumerBackoffMaxAttempts()} instead.
Since Camel 4.22, the consumer creation
+ * and subscription are handled by a single reconnection task
that uses the createConsumerBackoff*
+ * options.
+ */
+ @Deprecated
public int getSubscribeConsumerBackoffMaxAttempts() {
return subscribeConsumerBackoffMaxAttempts;
}
@@ -221,18 +229,33 @@ public class KafkaComponent extends HealthCheckComponent
implements SSLContextPa
* Camel will by default retry forever, and therefore never give up. If
you want to give up after many attempts,
* then set this option and Camel will then when giving up terminate the
consumer. You can manually restart the
* consumer by stopping and starting the route, to try again.
+ *
+ * @deprecated Use {@link #setCreateConsumerBackoffMaxAttempts(int)}
instead. Since Camel 4.22, the consumer
+ * creation and subscription are handled by a single
reconnection task that uses the
+ * createConsumerBackoff* options.
*/
+ @Deprecated
public void setSubscribeConsumerBackoffMaxAttempts(int
subscribeConsumerBackoffMaxAttempts) {
this.subscribeConsumerBackoffMaxAttempts =
subscribeConsumerBackoffMaxAttempts;
}
+ /**
+ * @deprecated Use {@link #getCreateConsumerBackoffInterval()} instead.
Since Camel 4.22, the consumer creation and
+ * subscription are handled by a single reconnection task that
uses the createConsumerBackoff* options.
+ */
+ @Deprecated
public long getSubscribeConsumerBackoffInterval() {
return subscribeConsumerBackoffInterval;
}
/**
* The delay in millis seconds to wait before trying again to subscribe to
the kafka broker.
+ *
+ * @deprecated Use {@link #setCreateConsumerBackoffInterval(long)}
instead. Since Camel 4.22, the consumer creation
+ * and subscription are handled by a single reconnection task
that uses the createConsumerBackoff*
+ * options.
*/
+ @Deprecated
public void setSubscribeConsumerBackoffInterval(long
subscribeConsumerBackoffInterval) {
this.subscribeConsumerBackoffInterval =
subscribeConsumerBackoffInterval;
}
@@ -247,8 +270,8 @@ public class KafkaComponent extends HealthCheckComponent
implements SSLContextPa
* topic, until it's created on the Kafka broker; and until then the Camel
Kafka consumer will fail and log a WARN
* about UNKNOWN_TOPIC_OR_PARTITION.
*
- * The option subscribeConsumerBackoffMaxAttempts can be configured to
give up trying to subscribe after a given
- * number of attempts.
+ * The option createConsumerBackoffMaxAttempts can be configured to give
up trying to subscribe after a given number
+ * of attempts.
*/
public void setSubscribeConsumerTopicMustExists(boolean
subscribeConsumerTopicMustExists) {
this.subscribeConsumerTopicMustExists =
subscribeConsumerTopicMustExists;
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..d371189a8025 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,
@@ -129,116 +131,82 @@ public class KafkaFetchRecords implements Runnable {
return;
}
- do {
- terminated = false;
+ try {
+ do {
+ terminated = false;
- if (!isConnected()) {
+ if (!isConnected()) {
- // shutdown existing consumer instance to release resources
(heartbeat)
- if (this.consumer != null) {
- safeConsumerClose();
- }
+ // shutdown existing consumer instance to release
resources (heartbeat)
+ if (this.consumer != null) {
+ safeConsumerClose();
+ }
+
+ // background task that creates and subscribes the kafka
consumer
+ // this stays registered in the internal task registry for
visibility via management (TUI/CLI/Hawtio)
+ currentBackoffInterval
+ =
kafkaConsumer.getEndpoint().getComponent().getCreateConsumerBackoffInterval();
+ int maxAttempts
+ =
kafkaConsumer.getEndpoint().getComponent().getCreateConsumerBackoffMaxAttempts();
+ if (reconnectPool == null) {
+ reconnectPool =
kafkaConsumer.getEndpoint().getCamelContext().getExecutorServiceManager()
+ .newSingleThreadScheduledExecutor(this,
"KafkaReconnect");
+ }
+ BackgroundTask task = Tasks.backgroundTask()
+ .withScheduledExecutor(reconnectPool)
+ .withBudget(Budgets.iterationTimeBudget()
+ .withMaxIterations(maxAttempts)
+
.withInterval(Duration.ofMillis(currentBackoffInterval))
+ .withInitialDelay(Duration.ZERO)
+ .withUnlimitedDuration()
+ .build())
+ .withName("KafkaReconnect-" + getPrintableTopic())
+ .build();
+ boolean success
+ =
task.run(kafkaConsumer.getEndpoint().getCamelContext(), this::reconnectTask);
+ if (!success) {
+ setupReconnectException(task, maxAttempts);
+ // give up and terminate this consumer
+ terminated = true;
+ break;
+ }
- // task that deals with creating kafka consumer
- 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;
+ setConnected(true);
}
- // 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)
-
.withInterval(Duration.ofMillis(currentBackoffInterval))
- .build())
- .build();
- success =
task.run(kafkaConsumer.getEndpoint().getCamelContext(),
this::initializeConsumerTask);
- if (!success) {
- int max =
kafkaConsumer.getEndpoint().getComponent().getSubscribeConsumerBackoffMaxAttempts();
- setupInitializeErrorException(task, max);
- // give up and terminate this consumer
- terminated = true;
- break;
+ if (isConnected()) {
+ metricsCollector.storeMetadata(consumer);
}
- setConnected(true);
- }
+ setLastError(null);
+ startPolling();
+ } while ((pollExceptionStrategy.canContinue() || isReconnect()) &&
isKafkaConsumerRunnable());
- if (isConnected()) {
- metricsCollector.storeMetadata(consumer);
+ if (LOG.isInfoEnabled()) {
+ LOG.info("Terminating KafkaConsumer thread {} receiving from
{}", threadId, getPrintableTopic());
}
- setLastError(null);
- startPolling();
- } while ((pollExceptionStrategy.canContinue() || isReconnect()) &&
isKafkaConsumerRunnable());
-
- if (LOG.isInfoEnabled()) {
- LOG.info("Terminating KafkaConsumer thread {} receiving from {}",
threadId, getPrintableTopic());
+ safeConsumerClose();
+ } finally {
+ if (reconnectPool != null) {
+
kafkaConsumer.getEndpoint().getCamelContext().getExecutorServiceManager().shutdown(reconnectPool);
+ reconnectPool = null;
+ }
}
-
- 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));
}
- 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 +224,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
diff --git
a/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_22.adoc
b/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_22.adoc
index 798509c98709..3e1b236e3a9c 100644
--- a/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_22.adoc
+++ b/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_22.adoc
@@ -885,6 +885,22 @@ that naturally differ, such as `CamelKafkaOffset`, are not
set on the batch exch
This is additive — these headers were previously absent from the batch
exchange, so nothing that worked before
changes. The headers on the individual record exchanges in the body are
unchanged.
+=== camel-kafka - Unified reconnection task
+
+The Kafka consumer's internal reconnection logic has been simplified.
Previously, creating a
+`KafkaConsumer` (kafka-client) and subscribing to topics were handled by two
separate foreground
+tasks with independent backoff settings. These have been merged into a single
background
+reconnection task that is visible via Camel management, including the TUI,
CLI, Hawtio, and other tooling.
+
+As a result, the `subscribeConsumerBackoffMaxAttempts` and
`subscribeConsumerBackoffInterval`
+component options are now deprecated and no longer used. The unified
reconnection task uses the
+`createConsumerBackoffMaxAttempts` and `createConsumerBackoffInterval` options
to control retry
+behavior for both consumer creation and topic subscription.
+
+If you previously configured `subscribeConsumerBackoffMaxAttempts` or
+`subscribeConsumerBackoffInterval`, migrate those settings to use
+`createConsumerBackoffMaxAttempts` and `createConsumerBackoffInterval` instead.
+
=== camel-core - Multicast UseOriginalAggregationStrategy fix
The Multicast EIP now correctly honors `UseOriginalAggregationStrategy`,
consistent with the Splitter
diff --git
a/dsl/camel-componentdsl/src/generated/java/org/apache/camel/builder/component/dsl/KafkaComponentBuilderFactory.java
b/dsl/camel-componentdsl/src/generated/java/org/apache/camel/builder/component/dsl/KafkaComponentBuilderFactory.java
index 6de8d52368be..d10ab6998e24 100644
---
a/dsl/camel-componentdsl/src/generated/java/org/apache/camel/builder/component/dsl/KafkaComponentBuilderFactory.java
+++
b/dsl/camel-componentdsl/src/generated/java/org/apache/camel/builder/component/dsl/KafkaComponentBuilderFactory.java
@@ -1084,6 +1084,7 @@ public interface KafkaComponentBuilderFactory {
* @param subscribeConsumerBackoffInterval the value to set
* @return the dsl builder
*/
+ @Deprecated
default KafkaComponentBuilder subscribeConsumerBackoffInterval(long
subscribeConsumerBackoffInterval) {
doSetProperty("subscribeConsumerBackoffInterval",
subscribeConsumerBackoffInterval);
return this;
@@ -1107,6 +1108,7 @@ public interface KafkaComponentBuilderFactory {
* @param subscribeConsumerBackoffMaxAttempts the value to set
* @return the dsl builder
*/
+ @Deprecated
default KafkaComponentBuilder subscribeConsumerBackoffMaxAttempts(int
subscribeConsumerBackoffMaxAttempts) {
doSetProperty("subscribeConsumerBackoffMaxAttempts",
subscribeConsumerBackoffMaxAttempts);
return this;
@@ -1120,8 +1122,8 @@ public interface KafkaComponentBuilderFactory {
* consume from the topic, until it's created on the Kafka broker; and
* until then the Camel Kafka consumer will fail and log a WARN about
* UNKNOWN_TOPIC_OR_PARTITION. The option
- * subscribeConsumerBackoffMaxAttempts can be configured to give up
- * trying to subscribe after a given number of attempts.
+ * createConsumerBackoffMaxAttempts can be configured to give up trying
+ * to subscribe after a given number of attempts.
*
* The option is a: <code>boolean</code> type.
*
