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 abdb48b29960 CAMEL-23997: Fix manual commit mode auto-committing
offsets
abdb48b29960 is described below
commit abdb48b299602579c292ce6558c3247c4a2f1511
Author: Claus Ibsen <[email protected]>
AuthorDate: Wed Jul 15 12:54:37 2026 +0200
CAMEL-23997: Fix manual commit mode auto-committing offsets
When allowManualCommit=true was used with DefaultKafkaManualCommitFactory
or DefaultKafkaManualAsyncCommitFactory, the streaming processor
auto-committed offsets for every processed record — even when the route
did not call KafkaManualCommit.commit(). This defeated the purpose of
manual commit, causing silent message loss (at-most-once instead of
at-least-once).
The fix guards the framework's auto-commit call in
KafkaRecordStreamingProcessorFacade.processPolledRecords() with
!isAllowManualCommit(). The recordOffset() call (which only caches
offsets internally) remains unconditional because breakOnFirstError
and offset repository updates depend on the cached offsets.
Also fixes state repository not being updated when allowManualCommit=true
with an offsetRepository.
Closes #24696
Co-Authored-By: Claude Opus 4.6 <[email protected]>
---
.../kafka/consumer/AbstractCommitManager.java | 5 +
.../consumer/DefaultKafkaManualAsyncCommit.java | 1 +
.../KafkaRecordStreamingProcessorFacade.java | 3 +-
.../kafka/consumer/CommitManagersTest.java | 101 +++++++++++++++++++++
.../KafkaRecordStreamingProcessorTest.java | 80 ++++++++++++++++
.../ROOT/pages/camel-4x-upgrade-guide-4_22.adoc | 12 +++
6 files changed, 201 insertions(+), 1 deletion(-)
diff --git
a/components/camel-kafka/src/main/java/org/apache/camel/component/kafka/consumer/AbstractCommitManager.java
b/components/camel-kafka/src/main/java/org/apache/camel/component/kafka/consumer/AbstractCommitManager.java
index 1e7638a80fa8..e3f48c5f749b 100644
---
a/components/camel-kafka/src/main/java/org/apache/camel/component/kafka/consumer/AbstractCommitManager.java
+++
b/components/camel-kafka/src/main/java/org/apache/camel/component/kafka/consumer/AbstractCommitManager.java
@@ -92,6 +92,11 @@ public abstract class AbstractCommitManager implements
CommitManager {
consumer.commitSync(
Collections.singletonMap(partition, new
OffsetAndMetadata(partitionLastOffset + 1)),
Duration.ofMillis(timeout));
+
+ StateRepository<String, String> offsetRepository =
configuration.getOffsetRepository();
+ if (offsetRepository != null) {
+ saveStateToOffsetRepository(partition, partitionLastOffset,
offsetRepository);
+ }
}
protected void saveStateToOffsetRepository(
diff --git
a/components/camel-kafka/src/main/java/org/apache/camel/component/kafka/consumer/DefaultKafkaManualAsyncCommit.java
b/components/camel-kafka/src/main/java/org/apache/camel/component/kafka/consumer/DefaultKafkaManualAsyncCommit.java
index 7eea8cdc5f46..20a86258ad6d 100644
---
a/components/camel-kafka/src/main/java/org/apache/camel/component/kafka/consumer/DefaultKafkaManualAsyncCommit.java
+++
b/components/camel-kafka/src/main/java/org/apache/camel/component/kafka/consumer/DefaultKafkaManualAsyncCommit.java
@@ -29,5 +29,6 @@ public class DefaultKafkaManualAsyncCommit extends
DefaultKafkaManualCommit impl
@Override
public void commit() {
commitManager.recordOffset(getPartition(), getRecordOffset());
+ commitManager.commit(getPartition());
}
}
diff --git
a/components/camel-kafka/src/main/java/org/apache/camel/component/kafka/consumer/support/streaming/KafkaRecordStreamingProcessorFacade.java
b/components/camel-kafka/src/main/java/org/apache/camel/component/kafka/consumer/support/streaming/KafkaRecordStreamingProcessorFacade.java
index 0e6aa6ace087..e7fcde868ccd 100644
---
a/components/camel-kafka/src/main/java/org/apache/camel/component/kafka/consumer/support/streaming/KafkaRecordStreamingProcessorFacade.java
+++
b/components/camel-kafka/src/main/java/org/apache/camel/component/kafka/consumer/support/streaming/KafkaRecordStreamingProcessorFacade.java
@@ -116,7 +116,8 @@ public class KafkaRecordStreamingProcessorFacade extends
AbstractKafkaRecordProc
}
}
- if (!result.isBreakOnErrorHit()) {
+ if (!result.isBreakOnErrorHit()
+ &&
!camelKafkaConsumer.getEndpoint().getConfiguration().isAllowManualCommit()) {
LOG.debug("Committing offset on successful execution");
// all records processed from partition so commit them
commitManager.commit(partition);
diff --git
a/components/camel-kafka/src/test/java/org/apache/camel/component/kafka/consumer/CommitManagersTest.java
b/components/camel-kafka/src/test/java/org/apache/camel/component/kafka/consumer/CommitManagersTest.java
new file mode 100644
index 000000000000..0e151d8cf111
--- /dev/null
+++
b/components/camel-kafka/src/test/java/org/apache/camel/component/kafka/consumer/CommitManagersTest.java
@@ -0,0 +1,101 @@
+/*
+ * 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.consumer;
+
+import org.apache.camel.component.kafka.KafkaConfiguration;
+import org.apache.camel.component.kafka.KafkaConsumer;
+import org.apache.camel.component.kafka.KafkaEndpoint;
+import org.apache.kafka.clients.consumer.Consumer;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+
+import static org.junit.jupiter.api.Assertions.assertInstanceOf;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+
+class CommitManagersTest {
+
+ private final Consumer<?, ?> consumer = mock(Consumer.class);
+ private final KafkaConsumer kafkaConsumer = mock(KafkaConsumer.class);
+ private final KafkaEndpoint endpoint = mock(KafkaEndpoint.class);
+ private final KafkaConfiguration configuration =
mock(KafkaConfiguration.class);
+
+ @BeforeEach
+ void setup() {
+ when(kafkaConsumer.getEndpoint()).thenReturn(endpoint);
+ when(endpoint.getConfiguration()).thenReturn(configuration);
+ }
+
+ @Test
+ void manualCommitWithNoFactoryUsesNoop() {
+ when(configuration.isAllowManualCommit()).thenReturn(true);
+ when(endpoint.getKafkaManualCommitFactory()).thenReturn(null);
+
+ CommitManager cm = CommitManagers.createCommitManager(consumer,
kafkaConsumer, "t1", "topic");
+
+ assertInstanceOf(NoopCommitManager.class, cm);
+ }
+
+ @Test
+ void manualCommitWithSyncFactoryUsesSync() {
+ when(configuration.isAllowManualCommit()).thenReturn(true);
+ when(endpoint.getKafkaManualCommitFactory()).thenReturn(new
DefaultKafkaManualCommitFactory());
+
+ CommitManager cm = CommitManagers.createCommitManager(consumer,
kafkaConsumer, "t1", "topic");
+
+ assertInstanceOf(SyncCommitManager.class, cm);
+ }
+
+ @Test
+ void manualCommitWithAsyncFactoryUsesAsync() {
+ when(configuration.isAllowManualCommit()).thenReturn(true);
+ when(endpoint.getKafkaManualCommitFactory()).thenReturn(new
DefaultKafkaManualAsyncCommitFactory());
+
+ CommitManager cm = CommitManagers.createCommitManager(consumer,
kafkaConsumer, "t1", "topic");
+
+ assertInstanceOf(AsyncCommitManager.class, cm);
+ }
+
+ @Test
+ void autoCommitWithOffsetRepositoryUsesCommitToOffset() {
+ when(configuration.isAllowManualCommit()).thenReturn(false);
+ when(configuration.getOffsetRepository()).thenReturn(mock());
+
+ CommitManager cm = CommitManagers.createCommitManager(consumer,
kafkaConsumer, "t1", "topic");
+
+ assertInstanceOf(CommitToOffsetManager.class, cm);
+ }
+
+ @Test
+ void autoCommitWithBatchingUsesAsync() {
+ when(configuration.isAllowManualCommit()).thenReturn(false);
+ when(configuration.isBatching()).thenReturn(true);
+
+ CommitManager cm = CommitManagers.createCommitManager(consumer,
kafkaConsumer, "t1", "topic");
+
+ assertInstanceOf(AsyncCommitManager.class, cm);
+ }
+
+ @Test
+ void defaultConfigUsesNoop() {
+ when(configuration.isAllowManualCommit()).thenReturn(false);
+
+ CommitManager cm = CommitManagers.createCommitManager(consumer,
kafkaConsumer, "t1", "topic");
+
+ assertInstanceOf(NoopCommitManager.class, cm);
+ }
+}
diff --git
a/components/camel-kafka/src/test/java/org/apache/camel/component/kafka/consumer/support/streaming/KafkaRecordStreamingProcessorTest.java
b/components/camel-kafka/src/test/java/org/apache/camel/component/kafka/consumer/support/streaming/KafkaRecordStreamingProcessorTest.java
new file mode 100644
index 000000000000..10c075536164
--- /dev/null
+++
b/components/camel-kafka/src/test/java/org/apache/camel/component/kafka/consumer/support/streaming/KafkaRecordStreamingProcessorTest.java
@@ -0,0 +1,80 @@
+/*
+ * 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.consumer.support.streaming;
+
+import org.apache.camel.Exchange;
+import org.apache.camel.Message;
+import org.apache.camel.Processor;
+import org.apache.camel.component.kafka.KafkaConfiguration;
+import org.apache.camel.component.kafka.KafkaConsumer;
+import org.apache.camel.component.kafka.consumer.CommitManager;
+import org.apache.camel.component.kafka.consumer.KafkaManualCommit;
+import org.apache.camel.component.kafka.serde.KafkaHeaderDeserializer;
+import org.apache.camel.spi.HeaderFilterStrategy;
+import org.apache.kafka.clients.consumer.ConsumerRecord;
+import org.apache.kafka.common.TopicPartition;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+class KafkaRecordStreamingProcessorTest {
+
+ private final KafkaConfiguration configuration =
mock(KafkaConfiguration.class);
+ private final Processor processor = mock(Processor.class);
+ private final CommitManager commitManager = mock(CommitManager.class);
+ private final KafkaConsumer kafkaConsumer = mock(KafkaConsumer.class);
+
+ private final TopicPartition topicPartition = new
TopicPartition("test-topic", 0);
+ private final ConsumerRecord<Object, Object> record = new
ConsumerRecord<>("test-topic", 0, 42L, "key", "value");
+
+ @BeforeEach
+ void setup() {
+ Exchange exchange = mock(Exchange.class);
+ Message message = mock(Message.class);
+ when(kafkaConsumer.createExchange(false)).thenReturn(exchange);
+ when(exchange.getMessage()).thenReturn(message);
+ when(exchange.getIn()).thenReturn(message);
+
+
when(configuration.getHeaderFilterStrategy()).thenReturn(mock(HeaderFilterStrategy.class));
+
when(configuration.getHeaderDeserializer()).thenReturn(mock(KafkaHeaderDeserializer.class));
+ }
+
+ @Test
+ void manualCommitStillRecordsOffset() {
+ when(configuration.isAllowManualCommit()).thenReturn(true);
+ when(commitManager.getManualCommit(any(), any(),
any())).thenReturn(mock(KafkaManualCommit.class));
+
+ KafkaRecordStreamingProcessor proc = new
KafkaRecordStreamingProcessor(configuration, processor, commitManager);
+ proc.processExchange(kafkaConsumer, topicPartition, false, false,
record);
+
+ verify(commitManager).recordOffset(topicPartition, 42L);
+ }
+
+ @Test
+ void autoCommitRecordsOffset() {
+ when(configuration.isAllowManualCommit()).thenReturn(false);
+
+ KafkaRecordStreamingProcessor proc = new
KafkaRecordStreamingProcessor(configuration, processor, commitManager);
+ proc.processExchange(kafkaConsumer, topicPartition, false, false,
record);
+
+ verify(commitManager).recordOffset(topicPartition, 42L);
+ }
+}
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 859c222605bc..80325fece7c1 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
@@ -295,6 +295,18 @@ The `queueBufferingMaxMessages` producer option has been
deprecated. This option
since the old Scala Kafka producer was removed in Camel 2.17 (CAMEL-9467). Use
`bufferMemorySize`
or `maxBlockMs` instead.
+=== camel-kafka - Manual commit no longer auto-commits offsets
+
+When using `allowManualCommit=true` together with a
`DefaultKafkaManualCommitFactory` or
+`DefaultKafkaManualAsyncCommitFactory`, the framework previously
auto-committed offsets for every
+processed record — even when the route did not call
`KafkaManualCommit.commit()`. This defeated the
+purpose of manual commit mode and could cause message loss.
+
+The framework no longer auto-commits offsets after processing each partition
when manual commit
+is enabled. Offsets are only committed when the route explicitly calls
`KafkaManualCommit.commit()`
+on the exchange header. The configured factory still controls whether that
explicit commit executes
+synchronously or asynchronously.
+
=== camel-jbang catalog tables fill the terminal width
The `camel catalog` commands (`camel catalog component`, `camel catalog
dataformat`,
