This is an automated email from the ASF dual-hosted git repository. davsclaus pushed a commit to branch fix/CAMEL-23997-manual-commit-noop in repository https://gitbox.apache.org/repos/asf/camel.git
commit fa3eb6fe89df2c2f54f7bfc6ee5f2c2ae630fc48 Author: Claus Ibsen <[email protected]> AuthorDate: Tue Jul 14 21:12:30 2026 +0200 CAMEL-23997: Fix manual commit mode auto-committing offsets with Sync/Async commit managers When allowManualCommit=true was used with a DefaultKafkaManualCommitFactory or DefaultKafkaManualAsyncCommitFactory, the CommitManagers factory selected a SyncCommitManager or AsyncCommitManager that auto-committed offsets for every processed record — even when the route did not call KafkaManualCommit.commit(). This defeated the purpose of manual commit. The fix always uses NoopCommitManager when allowManualCommit=true. The configured factory still controls whether the route's explicit commit() call executes synchronously or asynchronously. Co-Authored-By: Claude Opus 4.6 <[email protected]> Signed-off-by: Claus Ibsen <[email protected]> --- .../component/kafka/consumer/CommitManagers.java | 21 ++--- .../kafka/consumer/CommitManagersTest.java | 101 +++++++++++++++++++++ .../ROOT/pages/camel-4x-upgrade-guide-4_22.adoc | 15 +++ 3 files changed, 122 insertions(+), 15 deletions(-) diff --git a/components/camel-kafka/src/main/java/org/apache/camel/component/kafka/consumer/CommitManagers.java b/components/camel-kafka/src/main/java/org/apache/camel/component/kafka/consumer/CommitManagers.java index 476435f0562e..41a2897f6cd6 100644 --- a/components/camel-kafka/src/main/java/org/apache/camel/component/kafka/consumer/CommitManagers.java +++ b/components/camel-kafka/src/main/java/org/apache/camel/component/kafka/consumer/CommitManagers.java @@ -34,21 +34,12 @@ public final class CommitManagers { KafkaConfiguration configuration = kafkaConsumer.getEndpoint().getConfiguration(); if (configuration.isAllowManualCommit()) { - LOG.debug("Allowing manual commit management"); - KafkaManualCommitFactory manualCommitFactory = kafkaConsumer.getEndpoint().getKafkaManualCommitFactory(); - if (manualCommitFactory instanceof DefaultKafkaManualAsyncCommitFactory) { - LOG.debug("Using an async commit manager for manual commit management"); - return new AsyncCommitManager(consumer, kafkaConsumer, threadId, printableTopic); - } else { - if (manualCommitFactory instanceof DefaultKafkaManualCommitFactory) { - LOG.debug("Using a sync commit manager for manual commit management"); - return new SyncCommitManager(consumer, kafkaConsumer, threadId, printableTopic); - } else { - // This has been the default behavior for Camel - LOG.debug("Using an NO-OP commit manager for manual commit management"); - return new NoopCommitManager(consumer, kafkaConsumer, threadId, printableTopic); - } - } + // Manual commit mode: the framework must never auto-commit offsets. + // The configured KafkaManualCommitFactory (sync or async) controls how + // the route's explicit KafkaManualCommit.commit() call executes — that + // is independent of the commit manager selection. + LOG.debug("Using a NO-OP commit manager for manual commit management"); + return new NoopCommitManager(consumer, kafkaConsumer, threadId, printableTopic); } else { if (configuration.getOffsetRepository() != null) { LOG.debug("Using a commit-to-offset manager for commit management"); 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..5e47df91ebab --- /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 manualCommitWithSyncFactoryUsesNoop() { + when(configuration.isAllowManualCommit()).thenReturn(true); + when(endpoint.getKafkaManualCommitFactory()).thenReturn(new DefaultKafkaManualCommitFactory()); + + CommitManager cm = CommitManagers.createCommitManager(consumer, kafkaConsumer, "t1", "topic"); + + assertInstanceOf(NoopCommitManager.class, cm); + } + + @Test + void manualCommitWithAsyncFactoryUsesNoop() { + when(configuration.isAllowManualCommit()).thenReturn(true); + when(endpoint.getKafkaManualCommitFactory()).thenReturn(new DefaultKafkaManualAsyncCommitFactory()); + + CommitManager cm = CommitManagers.createCommitManager(consumer, kafkaConsumer, "t1", "topic"); + + assertInstanceOf(NoopCommitManager.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/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 0454ac8a835a..8633c006d116 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 @@ -243,6 +243,21 @@ Scalar value nodes are now sent as a single record. Only container nodes (`Array are still split, which is unchanged. Any `convertBodyTo(...)` previously used as a workaround is no longer required. +=== 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 now uses a no-op commit manager when manual commit is enabled, regardless of which +commit factory is configured. 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. + +Users relying on the previous (unintended) auto-commit behavior alongside manual commit should +remove `allowManualCommit=true` and use Kafka's native auto-commit instead. + === camel-jbang catalog tables fill the terminal width The `camel catalog` commands (`camel catalog component`, `camel catalog dataformat`,
