This is an automated email from the ASF dual-hosted git repository. davsclaus pushed a commit to branch fix/CAMEL-23996-idempotent in repository https://gitbox.apache.org/repos/asf/camel.git
commit 02b521dca0afb84c1daa36b629ea86b13d5aec1a Author: Claus Ibsen <[email protected]> AuthorDate: Tue Jul 14 20:04:04 2026 +0200 CAMEL-23996: Fix KafkaIdempotentRepository correctness bugs Co-Authored-By: Claude Opus 4.6 <[email protected]> Signed-off-by: Claus Ibsen <[email protected]> --- .../kafka/KafkaIdempotentRepository.java | 23 ++-- .../kafka/KafkaIdempotentRepositoryTest.java | 146 +++++++++++++++++++++ 2 files changed, 161 insertions(+), 8 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 79f59508a977..4cd733efb073 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 @@ -53,6 +53,7 @@ import org.apache.kafka.clients.producer.ProducerConfig; import org.apache.kafka.clients.producer.ProducerRecord; import org.apache.kafka.common.PartitionInfo; import org.apache.kafka.common.TopicPartition; +import org.apache.kafka.common.errors.WakeupException; import org.apache.kafka.common.serialization.StringDeserializer; import org.apache.kafka.common.serialization.StringSerializer; import org.slf4j.Logger; @@ -382,11 +383,14 @@ public class KafkaIdempotentRepository extends ServiceSupport implements Idempot @Override protected void doStop() throws Exception { + ServiceHelper.stopService(poller); + if (consumer != null) { + consumer.wakeup(); + } if (executorService != null && camelContext != null) { - camelContext.getExecutorServiceManager().shutdown(executorService); + camelContext.getExecutorServiceManager().shutdownNow(executorService); executorService = null; } - ServiceHelper.stopService(poller); IOHelper.close(consumer, "consumer", LOG); IOHelper.close(producer, "producer", LOG); LOG.debug("Stopped KafkaIdempotentRepository. Cache counter: {}", cacheCounter.get()); @@ -436,6 +440,8 @@ public class KafkaIdempotentRepository extends ServiceSupport implements Idempot for (ConsumerRecord<String, String> consumerRecord : consumerRecords) { addToCache(consumerRecord); } + } catch (WakeupException e) { + LOG.debug("TopicPoller woken up during shutdown"); } catch (Exception e) { LOG.warn("TopicPoller error syncing due to: " + e.getMessage() + ". This exception is ignored.", e); } @@ -471,15 +477,15 @@ public class KafkaIdempotentRepository extends ServiceSupport implements Idempot @Override public boolean add(String key) { - if (cache.containsKey(key)) { + if (cache.putIfAbsent(key, key) != null) { return false; - } else { - // update the local cache and broadcast the addition on the topic, - // which will be reflected - // at a later point in any peers - cache.put(key, key); + } + try { broadcastAction(key, CacheAction.add); return true; + } catch (Exception e) { + cache.remove(key, key); + throw e; } } @@ -521,6 +527,7 @@ public class KafkaIdempotentRepository extends ServiceSupport implements Idempot @Override public void clear() { + cache.clear(); broadcastAction(null, CacheAction.clear); } 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 new file mode 100644 index 000000000000..919667fdd2a7 --- /dev/null +++ b/components/camel-kafka/src/test/java/org/apache/camel/processor/idempotent/kafka/KafkaIdempotentRepositoryTest.java @@ -0,0 +1,146 @@ +/* + * 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.processor.idempotent.kafka; + +import java.lang.reflect.Field; +import java.util.Map; +import java.util.concurrent.ConcurrentHashMap; + +import org.apache.camel.RuntimeCamelException; +import org.apache.kafka.clients.producer.MockProducer; +import org.apache.kafka.clients.producer.ProducerRecord; +import org.apache.kafka.common.serialization.StringSerializer; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; + +import static org.junit.jupiter.api.Assertions.*; + +/** + * Unit tests for {@link KafkaIdempotentRepository} cache logic, without requiring a Kafka broker. + */ +public class KafkaIdempotentRepositoryTest { + + private Map<String, Object> cache; + private MockProducer<String, String> mockProducer; + private KafkaIdempotentRepository repository; + + @BeforeEach + void setUp() throws Exception { + cache = new ConcurrentHashMap<>(); + mockProducer = new MockProducer<>(true, null, new StringSerializer(), new StringSerializer()); + + repository = new KafkaIdempotentRepository(); + repository.setTopic("test-topic"); + + setField(repository, "cache", cache); + setField(repository, "producer", mockProducer); + } + + @Test + void testAddReturnsTrueForNewKey() { + assertTrue(repository.add("key1")); + assertTrue(cache.containsKey("key1")); + } + + @Test + void testAddReturnsFalseForDuplicateKey() { + assertTrue(repository.add("key1")); + assertFalse(repository.add("key1")); + } + + @Test + void testAddBroadcastsToKafka() { + repository.add("key1"); + + assertEquals(1, mockProducer.history().size()); + ProducerRecord<String, String> record = mockProducer.history().get(0); + assertEquals("test-topic", record.topic()); + assertEquals("key1", record.key()); + assertEquals("add", record.value()); + } + + @Test + void testAddRollsBackCacheOnBroadcastFailure() throws Exception { + MockProducer<String, String> failingProducer + = new MockProducer<>(false, null, new StringSerializer(), new StringSerializer()); + setField(repository, "producer", failingProducer); + + Thread sender = new Thread(() -> { + try { + repository.add("key1"); + } catch (RuntimeCamelException e) { + // expected + } + }); + sender.start(); + + Thread.sleep(100); + failingProducer.errorNext(new RuntimeException("Simulated send failure")); + sender.join(5000); + + assertFalse(cache.containsKey("key1")); + } + + @Test + void testContainsReflectsCacheState() { + assertFalse(repository.contains("key1")); + cache.put("key1", "key1"); + assertTrue(repository.contains("key1")); + } + + @Test + void testClearEmptiesLocalCache() { + repository.add("key1"); + repository.add("key2"); + assertEquals(2, cache.size()); + + repository.clear(); + assertTrue(cache.isEmpty()); + } + + @Test + void testClearBroadcastsToKafka() { + repository.clear(); + + assertEquals(1, mockProducer.history().size()); + ProducerRecord<String, String> record = mockProducer.history().get(0); + assertNull(record.key()); + assertEquals("clear", record.value()); + } + + @Test + void testRemoveDeletesFromCache() { + repository.add("key1"); + assertTrue(repository.contains("key1")); + + repository.remove("key1"); + assertFalse(repository.contains("key1")); + } + + @Test + void testRemoveThenAddSucceeds() { + repository.add("key1"); + repository.remove("key1"); + assertTrue(repository.add("key1")); + } + + private static void setField(Object target, String fieldName, Object value) throws Exception { + Field field = target.getClass().getDeclaredField(fieldName); + field.setAccessible(true); + field.set(target, value); + } +}
