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 e5a57075b479 CAMEL-26711: camel-kafka - KafkaIdempotentRepository add
rollback must not remove a key added by a peer (#29407)
e5a57075b479 is described below
commit e5a57075b479b26d5c88f59424209e03a91ec7a3
Author: allthingssecurity <[email protected]>
AuthorDate: Sun Nov 8 13:32:42 2026 +0530
CAMEL-26711: camel-kafka - KafkaIdempotentRepository add rollback must not
remove a key added by a peer (#29407)
Co-Authored-By: Claude Opus 5.5 <[email protected]>
---
.../kafka/KafkaIdempotentRepository.java | 10 +++---
.../kafka/KafkaIdempotentRepositoryTest.java | 37 ++++++++++++++++++++++
2 files changed, 43 insertions(+), 4 deletions(-)
diff --git
a/components/camel-kafka/src/main/java/org/apache/camel/processor/idempotent/kafka/KafkaIdempotentRepository.java
b/components/camel-kafka/src/main/java/org/apache/camel/processor/idempotent/kafka/KafkaIdempotentRepository.java
index 438b94245c6e..b4163ee69fb8 100644
---
a/components/camel-kafka/src/main/java/org/apache/camel/processor/idempotent/kafka/KafkaIdempotentRepository.java
+++
b/components/camel-kafka/src/main/java/org/apache/camel/processor/idempotent/kafka/KafkaIdempotentRepository.java
@@ -350,14 +350,16 @@ public class KafkaIdempotentRepository extends
ServiceSupport implements Idempot
@Override
public boolean add(String key) {
- if (cache.putIfAbsent(key, key) != null) {
+ // the poller stores key -> key for an add of a peer, so roll back
only the value put by this call
+ Object token = new Object();
+ if (cache.putIfAbsent(key, token) != null) {
return false;
}
try {
broadcastAction(key, CacheAction.add);
return true;
} catch (Exception e) {
- cache.remove(key, key);
+ cache.remove(key, token);
throw e;
}
}
@@ -377,10 +379,10 @@ public class KafkaIdempotentRepository extends
ServiceSupport implements Idempot
@Override
@ManagedOperation(description = "Remove the key from the store")
public boolean remove(String key) {
- // update the local cache and broadcast the addition on the topic,
which
+ // update the local cache and broadcast the removal on the topic, which
// will be reflected
// at a later point in any peers
- cache.remove(key, key);
+ cache.remove(key);
broadcastAction(key, CacheAction.remove);
return true;
}
diff --git
a/components/camel-kafka/src/test/java/org/apache/camel/processor/idempotent/kafka/KafkaIdempotentRepositoryTest.java
b/components/camel-kafka/src/test/java/org/apache/camel/processor/idempotent/kafka/KafkaIdempotentRepositoryTest.java
index d40db9e53974..a275b340e597 100644
---
a/components/camel-kafka/src/test/java/org/apache/camel/processor/idempotent/kafka/KafkaIdempotentRepositoryTest.java
+++
b/components/camel-kafka/src/test/java/org/apache/camel/processor/idempotent/kafka/KafkaIdempotentRepositoryTest.java
@@ -17,12 +17,15 @@
package org.apache.camel.processor.idempotent.kafka;
import java.lang.reflect.Field;
+import java.lang.reflect.Method;
import java.util.Map;
import java.util.Properties;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicReference;
import org.apache.camel.RuntimeCamelException;
+import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.producer.MockProducer;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.common.serialization.StringDeserializer;
@@ -101,6 +104,40 @@ public class KafkaIdempotentRepositoryTest {
assertFalse(cache.containsKey("key1"));
}
+ @Test
+ void testAddRollbackKeepsTheKeyAddedByAPeer() throws Exception {
+ MockProducer<String, String> failingProducer
+ = new MockProducer<>(false, null, new StringSerializer(), new
StringSerializer());
+ setField(repository, "changelog", changelog(failingProducer));
+
+ AtomicReference<Exception> failure = new AtomicReference<>();
+ Thread sender = new Thread(() -> {
+ try {
+ repository.add("key1");
+ } catch (RuntimeCamelException e) {
+ failure.set(e);
+ }
+ });
+ sender.start();
+
+ // the add record is sent and add() waits for the result
+ await().atMost(5, TimeUnit.SECONDS).until(() ->
failingProducer.history().size() == 1);
+
+ // meanwhile the poller applies the add of the same key by a peer
+ Method addToCache =
KafkaIdempotentRepository.class.getDeclaredMethod("addToCache",
ConsumerRecord.class);
+ addToCache.setAccessible(true);
+ addToCache.invoke(repository, new ConsumerRecord<>("test-topic", 0,
0L, "key1", "add"));
+
+ failingProducer.errorNext(new RuntimeException("Simulated send
failure"));
+ sender.join(5000);
+
+ assertFalse(sender.isAlive());
+ assertNotNull(failure.get());
+ // the rollback only undoes the add of this call, the key of the peer
stays
+ assertTrue(cache.containsKey("key1"));
+ assertFalse(repository.add("key1"));
+ }
+
@Test
void testContainsReflectsCacheState() {
assertFalse(repository.contains("key1"));
