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 2018/08/16 13:32:02 UTC
[camel] branch camel-2.22.x updated: CAMEL-12732 Take into account
isAllowManualCommit (#2478)
This is an automated email from the ASF dual-hosted git repository.
davsclaus pushed a commit to branch camel-2.22.x
in repository https://gitbox.apache.org/repos/asf/camel.git
The following commit(s) were added to refs/heads/camel-2.22.x by this push:
new 7b9d03e CAMEL-12732 Take into account isAllowManualCommit (#2478)
7b9d03e is described below
commit 7b9d03e726af00f78ceeed597c5bd83e5ba8d6d7
Author: elbaz michael <mi...@outlook.fr>
AuthorDate: Thu Aug 16 15:27:59 2018 +0200
CAMEL-12732 Take into account isAllowManualCommit (#2478)
Check isAllowManualCommit before setting the state
(Fix the manual commit for the state)
---
.../src/main/java/org/apache/camel/component/kafka/KafkaConsumer.java | 2 +-
1 file changed, 1 insertion(+), 1 deletion(-)
diff --git a/components/camel-kafka/src/main/java/org/apache/camel/component/kafka/KafkaConsumer.java b/components/camel-kafka/src/main/java/org/apache/camel/component/kafka/KafkaConsumer.java
index 5f56270..21dd228 100644
--- a/components/camel-kafka/src/main/java/org/apache/camel/component/kafka/KafkaConsumer.java
+++ b/components/camel-kafka/src/main/java/org/apache/camel/component/kafka/KafkaConsumer.java
@@ -378,7 +378,7 @@ public class KafkaConsumer extends DefaultConsumer {
private void commitOffset(StateRepository<String, String> offsetRepository, TopicPartition partition, long partitionLastOffset, boolean forceCommit) {
if (partitionLastOffset != -1) {
- if (offsetRepository != null) {
+ if (!endpoint.getConfiguration().isAllowManualCommit() && offsetRepository != null) {
log.debug("Saving offset repository state {} from topic {} with offset: {}", threadId, topicName, partitionLastOffset);
offsetRepository.setState(serializeOffsetKey(partition), serializeOffsetValue(partitionLastOffset));
} else if (forceCommit) {