From 7506538ba95a4f104f059697ba1334809d0d0c22 Mon Sep 17 00:00:00 2001 From: Guillaume Nodet Date: Thu, 3 Sep 2026 11:46:20 +0000 Subject: [PATCH 1/4] CAMEL-24463: Improve KeyValueRepository infrastructure - Add KeyValueRepositoryHelper unit tests (22 tests covering serialize/deserialize roundtrips, ByteBuffer, offset/length, error cases, null handling) - Add CacheProcessor test for cacheNull=true path - Fix RedisKeyValueRepository.put() to use atomic getAndSet() instead of non-atomic get+set - Extract shared KeyValueTtlValue from duplicate TtlValue inner classes in Ehcache and JCache backends - Fix import ordering in Ehcache/JCache source and test files --- .../processor/EhcacheKeyValueRepository.java | 50 +-- .../EhcacheKeyValueRepositoryTest.java | 3 +- .../processor/JCacheKeyValueRepository.java | 54 +-- .../JCacheKeyValueRepositoryTest.java | 5 +- .../redis/RedisKeyValueRepository.java | 6 +- .../camel/processor/CacheProcessorTest.java | 24 ++ .../camel/support/KeyValueTtlValue.java | 52 +++ .../support/KeyValueRepositoryHelperTest.java | 319 ++++++++++++++++++ 8 files changed, 429 insertions(+), 84 deletions(-) create mode 100644 core/camel-support/src/main/java/org/apache/camel/support/KeyValueTtlValue.java create mode 100644 core/camel-support/src/test/java/org/apache/camel/support/KeyValueRepositoryHelperTest.java diff --git a/components/camel-ehcache/src/main/java/org/apache/camel/component/ehcache/processor/EhcacheKeyValueRepository.java b/components/camel-ehcache/src/main/java/org/apache/camel/component/ehcache/processor/EhcacheKeyValueRepository.java index 35849594320b3..bea5636f695e6 100644 --- a/components/camel-ehcache/src/main/java/org/apache/camel/component/ehcache/processor/EhcacheKeyValueRepository.java +++ b/components/camel-ehcache/src/main/java/org/apache/camel/component/ehcache/processor/EhcacheKeyValueRepository.java @@ -16,8 +16,6 @@ */ package org.apache.camel.component.ehcache.processor; -import java.io.Serial; -import java.io.Serializable; import java.time.Duration; import java.util.HashSet; import java.util.Iterator; @@ -30,6 +28,7 @@ import org.apache.camel.spi.Configurer; import org.apache.camel.spi.KeyValueRepository; import org.apache.camel.spi.Metadata; +import org.apache.camel.support.KeyValueTtlValue; import org.apache.camel.support.service.ServiceSupport; import org.apache.camel.util.ObjectHelper; import org.ehcache.Cache; @@ -39,8 +38,8 @@ * A {@link KeyValueRepository} implementation backed by an Ehcache {@link Cache}. *

* Ehcache does not support per-entry TTL natively (TTL is set at the cache configuration level). This implementation - * wraps each value in a {@link TtlValue} that records the entry's expiration timestamp. Expired entries are removed - * lazily on access and during key scans, similar to how {@code MemoryKeyValueRepository} handles TTL. + * wraps each value in a {@link KeyValueTtlValue} that records the entry's expiration timestamp. Expired entries are + * removed lazily on access and during key scans, similar to how {@code MemoryKeyValueRepository} handles TTL. *

* This single implementation can serve as idempotent repository, aggregation repository, and state store via the * adapters in {@code camel-support} ({@code KeyValueIdempotentRepository} and {@code KeyValueAggregationRepository}). @@ -54,32 +53,7 @@ @ManagedResource(description = "Ehcache based key-value repository") public class EhcacheKeyValueRepository extends ServiceSupport implements KeyValueRepository { - /** - * Internal value wrapper that holds the actual value and an expiration timestamp. - */ - static final class TtlValue implements Serializable { - - @Serial - private static final long serialVersionUID = 1L; - - private final Object value; - private final long expiresAt; - - TtlValue(Object value, long expiresAt) { - this.value = value; - this.expiresAt = expiresAt; - } - - Object value() { - return value; - } - - boolean isExpired() { - return System.currentTimeMillis() >= expiresAt; - } - } - - private Cache cache; + private Cache cache; private EhcacheManager ehcacheManager; @Metadata(description = "Name of cache", defaultValue = "EhcacheKeyValueRepository") @@ -129,7 +103,7 @@ public void setCacheManager(CacheManager cacheManager) { @Override @ManagedOperation(description = "Get value by key") public Object get(String key) { - TtlValue entry = cache.get(key); + KeyValueTtlValue entry = cache.get(key); if (entry == null) { return null; } @@ -144,8 +118,8 @@ public Object get(String key) { @ManagedOperation(description = "Put a key-value pair with optional TTL") public Object put(String key, Object value, Duration ttl) { long expiresAt = hasPositiveTtl(ttl) ? System.currentTimeMillis() + ttl.toMillis() : Long.MAX_VALUE; - TtlValue previous = cache.get(key); - cache.put(key, new TtlValue(value, expiresAt)); + KeyValueTtlValue previous = cache.get(key); + cache.put(key, new KeyValueTtlValue(value, expiresAt)); if (previous == null || previous.isExpired()) { return null; } @@ -155,7 +129,7 @@ public Object put(String key, Object value, Duration ttl) { @Override @ManagedOperation(description = "Delete a key") public Object delete(String key) { - TtlValue entry = cache.get(key); + KeyValueTtlValue entry = cache.get(key); cache.remove(key); if (entry == null || entry.isExpired()) { return null; @@ -166,7 +140,7 @@ public Object delete(String key) { @Override @ManagedOperation(description = "Check if key exists") public boolean contains(String key) { - TtlValue entry = cache.get(key); + KeyValueTtlValue entry = cache.get(key); if (entry == null) { return false; } @@ -180,9 +154,9 @@ public boolean contains(String key) { @Override public Set keys() { Set keys = new HashSet<>(); - Iterator> it = cache.iterator(); + Iterator> it = cache.iterator(); while (it.hasNext()) { - Cache.Entry entry = it.next(); + Cache.Entry entry = it.next(); if (!entry.getValue().isExpired()) { keys.add(entry.getKey()); } else { @@ -213,7 +187,7 @@ protected void doStart() throws Exception { ObjectHelper.notNull(cacheManager, "cacheManager"); ehcacheManager = new EhcacheManager(cacheManager, false, null); ehcacheManager.start(); - cache = ehcacheManager.getCache(cacheName, String.class, TtlValue.class); + cache = ehcacheManager.getCache(cacheName, String.class, KeyValueTtlValue.class); } @Override diff --git a/components/camel-ehcache/src/test/java/org/apache/camel/component/ehcache/processor/EhcacheKeyValueRepositoryTest.java b/components/camel-ehcache/src/test/java/org/apache/camel/component/ehcache/processor/EhcacheKeyValueRepositoryTest.java index ce97b3abb13d8..9dbde03ca1aed 100644 --- a/components/camel-ehcache/src/test/java/org/apache/camel/component/ehcache/processor/EhcacheKeyValueRepositoryTest.java +++ b/components/camel-ehcache/src/test/java/org/apache/camel/component/ehcache/processor/EhcacheKeyValueRepositoryTest.java @@ -20,6 +20,7 @@ import java.util.Set; import java.util.concurrent.TimeUnit; +import org.apache.camel.support.KeyValueTtlValue; import org.ehcache.CacheManager; import org.ehcache.config.builders.CacheConfigurationBuilder; import org.ehcache.config.builders.CacheManagerBuilder; @@ -44,7 +45,7 @@ void setUp() throws Exception { .withCache(CACHE_NAME, CacheConfigurationBuilder.newCacheConfigurationBuilder( String.class, - EhcacheKeyValueRepository.TtlValue.class, + KeyValueTtlValue.class, ResourcePoolsBuilder.heap(100))) .build(true); diff --git a/components/camel-jcache/src/main/java/org/apache/camel/component/jcache/processor/JCacheKeyValueRepository.java b/components/camel-jcache/src/main/java/org/apache/camel/component/jcache/processor/JCacheKeyValueRepository.java index 85bf205f329ae..c5eddad5e9ae4 100644 --- a/components/camel-jcache/src/main/java/org/apache/camel/component/jcache/processor/JCacheKeyValueRepository.java +++ b/components/camel-jcache/src/main/java/org/apache/camel/component/jcache/processor/JCacheKeyValueRepository.java @@ -16,8 +16,6 @@ */ package org.apache.camel.component.jcache.processor; -import java.io.Serial; -import java.io.Serializable; import java.time.Duration; import java.util.HashSet; import java.util.Iterator; @@ -36,6 +34,7 @@ import org.apache.camel.spi.Configurer; import org.apache.camel.spi.KeyValueRepository; import org.apache.camel.spi.Metadata; +import org.apache.camel.support.KeyValueTtlValue; import org.apache.camel.support.service.ServiceSupport; import org.apache.camel.util.ObjectHelper; @@ -43,8 +42,8 @@ * A {@link KeyValueRepository} implementation backed by a JCache (JSR-107) {@link Cache}. *

* JCache does not support per-entry TTL natively (expiry is configured at the cache level via - * {@link javax.cache.expiry.ExpiryPolicy}). This implementation wraps each value in a {@link TtlValue} that records the - * entry's expiration timestamp. Expired entries are removed lazily on access and during key scans. + * {@link javax.cache.expiry.ExpiryPolicy}). This implementation wraps each value in a {@link KeyValueTtlValue} that + * records the entry's expiration timestamp. Expired entries are removed lazily on access and during key scans. *

* This single implementation can serve as idempotent repository, aggregation repository, and state store via the * adapters in {@code camel-support} ({@code KeyValueIdempotentRepository} and {@code KeyValueAggregationRepository}). @@ -58,34 +57,9 @@ @ManagedResource(description = "JCache based key-value repository") public class JCacheKeyValueRepository extends ServiceSupport implements CamelContextAware, KeyValueRepository { - /** - * Internal value wrapper that holds the actual value and an expiration timestamp. - */ - static final class TtlValue implements Serializable { - - @Serial - private static final long serialVersionUID = 1L; - - private final Object value; - private final long expiresAt; - - TtlValue(Object value, long expiresAt) { - this.value = value; - this.expiresAt = expiresAt; - } - - Object value() { - return value; - } - - boolean isExpired() { - return System.currentTimeMillis() >= expiresAt; - } - } - private CamelContext camelContext; - private Cache cache; - private JCacheManager cacheManager; + private Cache cache; + private JCacheManager cacheManager; @Metadata(description = "Configuration for JCache") private JCacheConfiguration configuration; @@ -115,11 +89,11 @@ public void setConfiguration(JCacheConfiguration configuration) { this.configuration = configuration; } - public Cache getCache() { + public Cache getCache() { return cache; } - public void setCache(Cache cache) { + public void setCache(Cache cache) { this.cache = cache; } @@ -135,7 +109,7 @@ public String getCacheName() { @Override @ManagedOperation(description = "Get value by key") public Object get(String key) { - TtlValue entry = cache.get(key); + KeyValueTtlValue entry = cache.get(key); if (entry == null) { return null; } @@ -150,8 +124,8 @@ public Object get(String key) { @ManagedOperation(description = "Put a key-value pair with optional TTL") public Object put(String key, Object value, Duration ttl) { long expiresAt = hasPositiveTtl(ttl) ? System.currentTimeMillis() + ttl.toMillis() : Long.MAX_VALUE; - TtlValue previous = cache.get(key); - cache.put(key, new TtlValue(value, expiresAt)); + KeyValueTtlValue previous = cache.get(key); + cache.put(key, new KeyValueTtlValue(value, expiresAt)); if (previous == null || previous.isExpired()) { return null; } @@ -161,7 +135,7 @@ public Object put(String key, Object value, Duration ttl) { @Override @ManagedOperation(description = "Delete a key") public Object delete(String key) { - TtlValue entry = cache.get(key); + KeyValueTtlValue entry = cache.get(key); cache.remove(key); if (entry == null || entry.isExpired()) { return null; @@ -172,7 +146,7 @@ public Object delete(String key) { @Override @ManagedOperation(description = "Check if key exists") public boolean contains(String key) { - TtlValue entry = cache.get(key); + KeyValueTtlValue entry = cache.get(key); if (entry == null) { return false; } @@ -186,9 +160,9 @@ public boolean contains(String key) { @Override public Set keys() { Set keys = new HashSet<>(); - Iterator> it = cache.iterator(); + Iterator> it = cache.iterator(); while (it.hasNext()) { - Cache.Entry entry = it.next(); + Cache.Entry entry = it.next(); if (!entry.getValue().isExpired()) { keys.add(entry.getKey()); } else { diff --git a/components/camel-jcache/src/test/java/org/apache/camel/component/jcache/processor/JCacheKeyValueRepositoryTest.java b/components/camel-jcache/src/test/java/org/apache/camel/component/jcache/processor/JCacheKeyValueRepositoryTest.java index f7077c08eb691..bdc009bc32132 100644 --- a/components/camel-jcache/src/test/java/org/apache/camel/component/jcache/processor/JCacheKeyValueRepositoryTest.java +++ b/components/camel-jcache/src/test/java/org/apache/camel/component/jcache/processor/JCacheKeyValueRepositoryTest.java @@ -26,6 +26,7 @@ import org.apache.camel.component.jcache.JCacheHelper; import org.apache.camel.component.jcache.JCacheManager; import org.apache.camel.component.jcache.support.HazelcastTest; +import org.apache.camel.support.KeyValueTtlValue; import org.apache.camel.test.junit6.CamelTestSupport; import org.junit.jupiter.api.Test; @@ -35,8 +36,8 @@ @HazelcastTest class JCacheKeyValueRepositoryTest extends CamelTestSupport { - private JCacheManager cacheManager; - private Cache cache; + private JCacheManager cacheManager; + private Cache cache; private JCacheKeyValueRepository repository; @Override diff --git a/components/camel-redis/src/main/java/org/apache/camel/component/redis/RedisKeyValueRepository.java b/components/camel-redis/src/main/java/org/apache/camel/component/redis/RedisKeyValueRepository.java index 3bedccad69fb2..687194885cdba 100644 --- a/components/camel-redis/src/main/java/org/apache/camel/component/redis/RedisKeyValueRepository.java +++ b/components/camel-redis/src/main/java/org/apache/camel/component/redis/RedisKeyValueRepository.java @@ -104,11 +104,11 @@ public RedisKeyValueRepository(String endpoint, String keyPrefix) { public @Nullable Object put(String key, Object value, Duration ttl) { RBucket bucket = redisson.getBucket(toRedisKey(key), ByteArrayCodec.INSTANCE); byte[] serialized = KeyValueRepositoryHelper.serialize(value); - byte[] previous = bucket.get(); + byte[] previous; if (hasPositiveTtl(ttl)) { - bucket.set(serialized, ttl); + previous = bucket.getAndSet(serialized, ttl); } else { - bucket.set(serialized); + previous = bucket.getAndSet(serialized); } return previous != null ? KeyValueRepositoryHelper.deserialize(previous) : null; } diff --git a/core/camel-core/src/test/java/org/apache/camel/processor/CacheProcessorTest.java b/core/camel-core/src/test/java/org/apache/camel/processor/CacheProcessorTest.java index 07a99beb1b056..eb3e385e38812 100644 --- a/core/camel-core/src/test/java/org/apache/camel/processor/CacheProcessorTest.java +++ b/core/camel-core/src/test/java/org/apache/camel/processor/CacheProcessorTest.java @@ -137,6 +137,22 @@ void testCacheNullBodyNotCachedByDefault() throws Exception { MockEndpoint.assertIsSatisfied(context); } + @Test + void testCacheNullBodyCachedWhenCacheNullTrue() throws Exception { + MockEndpoint service = getMockEndpoint("mock:cache-null-service"); + + // First call — null body result IS cached when cacheNull(true) + service.expectedMessageCount(1); + template.sendBodyAndHeader("direct:cached-null-true", "req1", "productId", "A"); + MockEndpoint.assertIsSatisfied(context); + + // Second call — cache hit (null WAS cached), service is NOT called again + service.reset(); + service.expectedMessageCount(0); + template.sendBodyAndHeader("direct:cached-null-true", "req2", "productId", "A"); + MockEndpoint.assertIsSatisfied(context); + } + @Test void testCacheExpressionClauseForm() throws Exception { MockEndpoint mock = getMockEndpoint("mock:clause-result"); @@ -195,6 +211,14 @@ public void configure() { .end() .to("mock:null-result"); + // Cache with null body and cacheNull=true + from("direct:cached-null-true") + .cache(simple("${header.productId}")).cacheNull(true) + .to("mock:cache-null-service") + .setBody(constant(null)) + .end() + .to("mock:cache-null-result"); + // Expression clause form from("direct:cached-clause") .cache().simple("${header.productId}") diff --git a/core/camel-support/src/main/java/org/apache/camel/support/KeyValueTtlValue.java b/core/camel-support/src/main/java/org/apache/camel/support/KeyValueTtlValue.java new file mode 100644 index 0000000000000..59a46b2de3c04 --- /dev/null +++ b/core/camel-support/src/main/java/org/apache/camel/support/KeyValueTtlValue.java @@ -0,0 +1,52 @@ +/* + * 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.support; + +import java.io.Serial; +import java.io.Serializable; + +/** + * A value wrapper that holds the actual value and an expiration timestamp. Used by KeyValueRepository implementations + * that need client-side TTL management (e.g., Ehcache, JCache) because the underlying store does not support per-entry + * TTL. + * + * @since 4.23 + */ +public final class KeyValueTtlValue implements Serializable { + @Serial + private static final long serialVersionUID = 1L; + + private final Object value; + private final long expiresAt; + + public KeyValueTtlValue(Object value, long expiresAt) { + this.value = value; + this.expiresAt = expiresAt; + } + + public Object value() { + return value; + } + + public long expiresAt() { + return expiresAt; + } + + public boolean isExpired() { + return System.currentTimeMillis() >= expiresAt; + } +} diff --git a/core/camel-support/src/test/java/org/apache/camel/support/KeyValueRepositoryHelperTest.java b/core/camel-support/src/test/java/org/apache/camel/support/KeyValueRepositoryHelperTest.java new file mode 100644 index 0000000000000..0977b39c41b03 --- /dev/null +++ b/core/camel-support/src/test/java/org/apache/camel/support/KeyValueRepositoryHelperTest.java @@ -0,0 +1,319 @@ +/* + * 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.support; + +import java.io.Serial; +import java.io.Serializable; +import java.nio.ByteBuffer; +import java.util.List; +import java.util.Map; +import java.util.Objects; + +import org.apache.camel.RuntimeCamelException; +import org.junit.jupiter.api.Test; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatThrownBy; + +class KeyValueRepositoryHelperTest { + + // ------------------------------------------------------------------------- + // serialize / deserialize roundtrip — scalar types + // ------------------------------------------------------------------------- + + @Test + void testSerializeDeserializeString() { + String original = "hello camel"; + + byte[] bytes = KeyValueRepositoryHelper.serialize(original); + Object result = KeyValueRepositoryHelper.deserialize(bytes); + + assertThat(result).isInstanceOf(String.class).isEqualTo(original); + } + + @Test + void testSerializeDeserializeInteger() { + Integer original = 42; + + byte[] bytes = KeyValueRepositoryHelper.serialize(original); + Object result = KeyValueRepositoryHelper.deserialize(bytes); + + assertThat(result).isInstanceOf(Integer.class).isEqualTo(original); + } + + @Test + void testSerializeDeserializeLong() { + Long original = Long.MAX_VALUE; + + byte[] bytes = KeyValueRepositoryHelper.serialize(original); + Object result = KeyValueRepositoryHelper.deserialize(bytes); + + assertThat(result).isInstanceOf(Long.class).isEqualTo(original); + } + + @Test + void testSerializeDeserializeBoolean() { + byte[] bytes = KeyValueRepositoryHelper.serialize(Boolean.TRUE); + Object result = KeyValueRepositoryHelper.deserialize(bytes); + + assertThat(result).isEqualTo(Boolean.TRUE); + } + + @Test + void testSerializeProduceNonEmptyByteArray() { + byte[] bytes = KeyValueRepositoryHelper.serialize("test"); + + assertThat(bytes).isNotNull().isNotEmpty(); + } + + // ------------------------------------------------------------------------- + // serialize / deserialize roundtrip — complex Serializable object + // ------------------------------------------------------------------------- + + @Test + void testSerializeDeserializeComplexObject() { + ComplexPayload original = new ComplexPayload("order-99", 3, List.of("item-a", "item-b")); + + byte[] bytes = KeyValueRepositoryHelper.serialize(original); + Object result = KeyValueRepositoryHelper.deserialize(bytes); + + assertThat(result).isInstanceOf(ComplexPayload.class).isEqualTo(original); + } + + @Test + void testSerializeDeserializeNestedMap() { + Map original = Map.of("alpha", 1, "beta", 2); + + byte[] bytes = KeyValueRepositoryHelper.serialize(original); + Object result = KeyValueRepositoryHelper.deserialize(bytes); + + assertThat(result).isEqualTo(original); + } + + // ------------------------------------------------------------------------- + // ByteBuffer serialize / deserialize roundtrip + // ------------------------------------------------------------------------- + + @Test + void testSerializeToByteBufferAndDeserialize() { + String original = "byteBuffer value"; + + ByteBuffer buffer = KeyValueRepositoryHelper.serializeToByteBuffer(original); + Object result = KeyValueRepositoryHelper.deserialize(buffer); + + assertThat(result).isEqualTo(original); + } + + @Test + void testSerializeToByteBufferReturnsNonNull() { + ByteBuffer buffer = KeyValueRepositoryHelper.serializeToByteBuffer(42); + + assertThat(buffer).isNotNull(); + assertThat(buffer.remaining()).isPositive(); + } + + @Test + void testDeserializeByteBufferConsumesRemainingBytes() { + Integer original = 123; + ByteBuffer buffer = KeyValueRepositoryHelper.serializeToByteBuffer(original); + + // Position is at start; after deserialize the buffer should be fully consumed + assertThat(buffer.position()).isZero(); + Object result = KeyValueRepositoryHelper.deserialize(buffer); + + assertThat(result).isEqualTo(original); + assertThat(buffer.remaining()).isZero(); + } + + @Test + void testSerializeToByteBufferRoundtripComplexObject() { + ComplexPayload original = new ComplexPayload("shipment-7", 10, List.of("sku-x")); + + ByteBuffer buffer = KeyValueRepositoryHelper.serializeToByteBuffer(original); + Object result = KeyValueRepositoryHelper.deserialize(buffer); + + assertThat(result).isInstanceOf(ComplexPayload.class).isEqualTo(original); + } + + // ------------------------------------------------------------------------- + // deserialize(bytes, offset, length) variant + // ------------------------------------------------------------------------- + + @Test + void testDeserializeWithOffsetAndLength() { + String original = "offset-test"; + byte[] serialized = KeyValueRepositoryHelper.serialize(original); + + // Embed serialized bytes into a larger array with a 4-byte header prefix + int headerSize = 4; + byte[] wrapped = new byte[headerSize + serialized.length]; + System.arraycopy(serialized, 0, wrapped, headerSize, serialized.length); + + Object result = KeyValueRepositoryHelper.deserialize(wrapped, headerSize, serialized.length); + + assertThat(result).isEqualTo(original); + } + + @Test + void testDeserializeWithOffsetZeroFullLength() { + Integer original = 999; + byte[] bytes = KeyValueRepositoryHelper.serialize(original); + + Object result = KeyValueRepositoryHelper.deserialize(bytes, 0, bytes.length); + + assertThat(result).isEqualTo(original); + } + + @Test + void testDeserializeWithOffsetIgnoresTrailingBytes() { + String original = "trimmed"; + byte[] serialized = KeyValueRepositoryHelper.serialize(original); + + // Append garbage trailing bytes — they must be ignored + byte[] withTrail = new byte[serialized.length + 10]; + System.arraycopy(serialized, 0, withTrail, 0, serialized.length); + + Object result = KeyValueRepositoryHelper.deserialize(withTrail, 0, serialized.length); + + assertThat(result).isEqualTo(original); + } + + // ------------------------------------------------------------------------- + // Error case — non-Serializable object throws RuntimeCamelException + // ------------------------------------------------------------------------- + + @Test + void testSerializeNonSerializableThrowsRuntimeCamelException() { + Object notSerializable = new NonSerializable(); + + assertThatThrownBy(() -> KeyValueRepositoryHelper.serialize(notSerializable)) + .isInstanceOf(RuntimeCamelException.class) + .hasMessageContaining("Failed to serialize value"); + } + + @Test + void testSerializeToByteBufferNonSerializableThrowsRuntimeCamelException() { + Object notSerializable = new NonSerializable(); + + assertThatThrownBy(() -> KeyValueRepositoryHelper.serializeToByteBuffer(notSerializable)) + .isInstanceOf(RuntimeCamelException.class) + .hasMessageContaining("Failed to serialize value"); + } + + @Test + void testDeserializeCorruptBytesThrowsRuntimeCamelException() { + byte[] corrupt = new byte[] { 0x00, 0x01, 0x02, 0x03 }; + + assertThatThrownBy(() -> KeyValueRepositoryHelper.deserialize(corrupt)) + .isInstanceOf(RuntimeCamelException.class) + .hasMessageContaining("Failed to deserialize value"); + } + + @Test + void testDeserializeOffsetCorruptBytesThrowsRuntimeCamelException() { + byte[] corrupt = new byte[] { 0x00, 0x01, 0x02, 0x03 }; + + assertThatThrownBy(() -> KeyValueRepositoryHelper.deserialize(corrupt, 0, corrupt.length)) + .isInstanceOf(RuntimeCamelException.class) + .hasMessageContaining("Failed to deserialize value"); + } + + @Test + void testDeserializeByteBufferCorruptThrowsRuntimeCamelException() { + ByteBuffer corrupt = ByteBuffer.wrap(new byte[] { 0x00, 0x01, 0x02, 0x03 }); + + assertThatThrownBy(() -> KeyValueRepositoryHelper.deserialize(corrupt)) + .isInstanceOf(RuntimeCamelException.class) + .hasMessageContaining("Failed to deserialize value"); + } + + // ------------------------------------------------------------------------- + // Null handling edge cases + // ------------------------------------------------------------------------- + + @Test + void testSerializeNullRoundtrip() { + byte[] bytes = KeyValueRepositoryHelper.serialize(null); + + assertThat(bytes).isNotNull().isNotEmpty(); + + Object result = KeyValueRepositoryHelper.deserialize(bytes); + assertThat(result).isNull(); + } + + @Test + void testSerializeToByteBufferNullRoundtrip() { + ByteBuffer buffer = KeyValueRepositoryHelper.serializeToByteBuffer(null); + Object result = KeyValueRepositoryHelper.deserialize(buffer); + + assertThat(result).isNull(); + } + + @Test + void testSerializeNullOffsetRoundtrip() { + byte[] bytes = KeyValueRepositoryHelper.serialize(null); + Object result = KeyValueRepositoryHelper.deserialize(bytes, 0, bytes.length); + + assertThat(result).isNull(); + } + + // ------------------------------------------------------------------------- + // Helper types + // ------------------------------------------------------------------------- + + /** A non-serializable type used to provoke serialization failures. */ + private static class NonSerializable { + // intentionally does not implement Serializable + } + + /** A complex Serializable value object used for roundtrip assertions. */ + private static final class ComplexPayload implements Serializable { + + @Serial + private static final long serialVersionUID = 1L; + + private final String id; + private final int quantity; + private final List items; + + ComplexPayload(String id, int quantity, List items) { + this.id = id; + this.quantity = quantity; + this.items = List.copyOf(items); + } + + @Override + public boolean equals(Object o) { + if (!(o instanceof ComplexPayload other)) { + return false; + } + return quantity == other.quantity + && Objects.equals(id, other.id) + && Objects.equals(items, other.items); + } + + @Override + public int hashCode() { + return Objects.hash(id, quantity, items); + } + + @Override + public String toString() { + return "ComplexPayload{id='" + id + "', quantity=" + quantity + ", items=" + items + "}"; + } + } +} From a58a0649a1cdc12a283d15ff3512ed4e99492c47 Mon Sep 17 00:00:00 2001 From: Guillaume Nodet Date: Thu, 3 Sep 2026 13:15:59 +0000 Subject: [PATCH 2/4] CAMEL-24463: Add KVR backend documentation, JDBC CAS overrides, and Javadoc improvements - Add comprehensive KeyValueRepository backends documentation page (keyValueRepository.adoc) covering all 11 backends with configuration examples, atomicity guarantees, TTL behavior, and adapter usage for Idempotent Consumer / Aggregator patterns - Add atomic replace() and delete(key,expected) overrides to JdbcKeyValueRepository using SQL WHERE clause on ITEM_VALUE for server-side CAS - Add Javadoc about non-atomic put() in Cassandra, Ehcache, and JCache backends - Add Javadoc about TTL/topic growth and non-atomic CAS in KafkaKeyValueRepository --- .../CassandraKeyValueRepository.java | 7 + .../processor/EhcacheKeyValueRepository.java | 8 + .../processor/JCacheKeyValueRepository.java | 8 + .../kafka/KafkaKeyValueRepository.java | 26 + .../keyvalue/jdbc/JdbcKeyValueRepository.java | 73 ++ .../eips/pages/keyValueRepository.adoc | 1077 +++++++++++++++++ 6 files changed, 1199 insertions(+) create mode 100644 core/camel-core-engine/src/main/docs/modules/eips/pages/keyValueRepository.adoc diff --git a/components/camel-cassandraql/src/main/java/org/apache/camel/processor/keyvalue/cassandra/CassandraKeyValueRepository.java b/components/camel-cassandraql/src/main/java/org/apache/camel/processor/keyvalue/cassandra/CassandraKeyValueRepository.java index b365dd6f2fa2f..74136447a9d46 100644 --- a/components/camel-cassandraql/src/main/java/org/apache/camel/processor/keyvalue/cassandra/CassandraKeyValueRepository.java +++ b/components/camel-cassandraql/src/main/java/org/apache/camel/processor/keyvalue/cassandra/CassandraKeyValueRepository.java @@ -283,6 +283,13 @@ public Object get(String key) { return buffer != null ? KeyValueRepositoryHelper.deserialize(buffer) : null; } + /** + * Stores a value with optional TTL. + *

+ * Note: The previous value is read in a separate query before the upsert. This is not atomic — another + * client could modify the entry between the read and the write. The stored value is always correct, but the + * returned previous value may be stale. Cassandra does not provide a native {@code getAndSet} equivalent. + */ @Override @ManagedOperation(description = "Put a key-value pair with optional TTL") public Object put(String key, Object value, Duration ttl) { diff --git a/components/camel-ehcache/src/main/java/org/apache/camel/component/ehcache/processor/EhcacheKeyValueRepository.java b/components/camel-ehcache/src/main/java/org/apache/camel/component/ehcache/processor/EhcacheKeyValueRepository.java index bea5636f695e6..30fc8ae237efa 100644 --- a/components/camel-ehcache/src/main/java/org/apache/camel/component/ehcache/processor/EhcacheKeyValueRepository.java +++ b/components/camel-ehcache/src/main/java/org/apache/camel/component/ehcache/processor/EhcacheKeyValueRepository.java @@ -114,6 +114,14 @@ public Object get(String key) { return entry.value(); } + /** + * Stores a value with optional TTL. + *

+ * Note: The previous value is read in a separate call before the put. This is not atomic — another thread + * could modify the entry between the read and the write. The stored value is always correct, but the returned + * previous value may be stale. Ehcache does not provide a native {@code getAndPut} equivalent for its + * {@code Cache} API. + */ @Override @ManagedOperation(description = "Put a key-value pair with optional TTL") public Object put(String key, Object value, Duration ttl) { diff --git a/components/camel-jcache/src/main/java/org/apache/camel/component/jcache/processor/JCacheKeyValueRepository.java b/components/camel-jcache/src/main/java/org/apache/camel/component/jcache/processor/JCacheKeyValueRepository.java index c5eddad5e9ae4..5212d6fcb47c0 100644 --- a/components/camel-jcache/src/main/java/org/apache/camel/component/jcache/processor/JCacheKeyValueRepository.java +++ b/components/camel-jcache/src/main/java/org/apache/camel/component/jcache/processor/JCacheKeyValueRepository.java @@ -120,6 +120,14 @@ public Object get(String key) { return entry.value(); } + /** + * Stores a value with optional TTL. + *

+ * Note: The previous value is read in a separate call before the put. This is not atomic — another thread + * could modify the entry between the read and the write. The stored value is always correct, but the returned + * previous value may be stale. JCache does not provide a {@code getAndPut} equivalent that also accepts a custom + * value type with per-entry TTL. + */ @Override @ManagedOperation(description = "Put a key-value pair with optional TTL") public Object put(String key, Object value, Duration ttl) { diff --git a/components/camel-kafka/src/main/java/org/apache/camel/processor/keyvalue/kafka/KafkaKeyValueRepository.java b/components/camel-kafka/src/main/java/org/apache/camel/processor/keyvalue/kafka/KafkaKeyValueRepository.java index 5cc946841d028..a81ef5ffd7750 100644 --- a/components/camel-kafka/src/main/java/org/apache/camel/processor/keyvalue/kafka/KafkaKeyValueRepository.java +++ b/components/camel-kafka/src/main/java/org/apache/camel/processor/keyvalue/kafka/KafkaKeyValueRepository.java @@ -80,6 +80,14 @@ *

* The topic used must be unique per logical repository. TTL is managed locally via expiration timestamps in the cache; * expired entries are lazily evicted on access. + *

+ * Note on TTL and topic growth: TTL expiration is managed locally in the LRU cache. The Kafka topic retains + * all entries (including expired ones) indefinitely. On restart, expired entries are replayed from the topic and + * re-populated in the cache with their original expiration timestamps (they will appear as expired immediately). + * To prevent unbounded topic growth, configure the topic with Kafka's built-in log compaction + * ({@code cleanup.policy=compact}) and optionally {@code min.compaction.lag.ms} / {@code delete.retention.ms} + * to control retention. Deleted entries (tombstones) are represented as records with a {@code DELETE} action byte, + * which log compaction will eventually remove. * * @since 4.23 */ @@ -242,6 +250,24 @@ public void clear() { } } + /** + * Associates {@code value} with {@code key} only if {@code key} is not already present (or its entry has expired). + * Returns the existing (non-expired) value if one was already mapped, or {@code null} if the insertion succeeded. + * + *

+ * Note on atomicity: The check-then-act sequence is locally atomic with respect to other threads in this + * JVM (backed by {@link java.util.concurrent.ConcurrentMap#putIfAbsent}), but the subsequent broadcast to Kafka + * is a separate, non-atomic step. In a multi-instance deployment, two nodes may each observe the key as absent, + * both insert locally, and both broadcast — resulting in the last broadcast winning in the Kafka topic and + * propagating to all other instances. This implementation therefore does not provide distributed + * compare-and-swap (CAS) semantics; it is suitable only for best-effort deduplication within a single JVM or + * in scenarios where occasional duplicates across instances are acceptable. + * + * @param key the key to insert + * @param value the value to associate + * @param ttl optional time-to-live; {@code null} or zero means no expiration + * @return the existing value if already present and not expired, or {@code null} if the entry was inserted + */ @Override public Object putIfAbsent(String key, Object value, Duration ttl) { CacheEntry existing = cache.get(key); diff --git a/components/camel-sql/src/main/java/org/apache/camel/processor/keyvalue/jdbc/JdbcKeyValueRepository.java b/components/camel-sql/src/main/java/org/apache/camel/processor/keyvalue/jdbc/JdbcKeyValueRepository.java index 2988e16c99256..47eadd3efd77e 100644 --- a/components/camel-sql/src/main/java/org/apache/camel/processor/keyvalue/jdbc/JdbcKeyValueRepository.java +++ b/components/camel-sql/src/main/java/org/apache/camel/processor/keyvalue/jdbc/JdbcKeyValueRepository.java @@ -75,6 +75,10 @@ public class JdbcKeyValueRepository extends ServiceSupport implements KeyValueRe = "SELECT ITEM_KEY FROM CAMEL_KEYVALUE WHERE EXPIRES_AT = 0 OR EXPIRES_AT > ?"; protected static final String DEFAULT_DELETE_EXPIRED_STRING = "DELETE FROM CAMEL_KEYVALUE WHERE EXPIRES_AT > 0 AND EXPIRES_AT <= ?"; + protected static final String DEFAULT_UPDATE_IF_VALUE_STRING + = "UPDATE CAMEL_KEYVALUE SET ITEM_VALUE = ?, EXPIRES_AT = ? WHERE ITEM_KEY = ? AND ITEM_VALUE = ?"; + protected static final String DEFAULT_DELETE_IF_VALUE_STRING + = "DELETE FROM CAMEL_KEYVALUE WHERE ITEM_KEY = ? AND ITEM_VALUE = ?"; private static final Logger LOG = LoggerFactory.getLogger(JdbcKeyValueRepository.class); @@ -105,6 +109,10 @@ public class JdbcKeyValueRepository extends ServiceSupport implements KeyValueRe private String selectKeysString = DEFAULT_SELECT_KEYS_STRING; @Metadata(label = "advanced", description = "SQL query to use for deleting expired entries") private String deleteExpiredString = DEFAULT_DELETE_EXPIRED_STRING; + @Metadata(label = "advanced", description = "SQL query to use for conditional update (CAS replace)") + private String updateIfValueString = DEFAULT_UPDATE_IF_VALUE_STRING; + @Metadata(label = "advanced", description = "SQL query to use for conditional delete (CAS delete)") + private String deleteIfValueString = DEFAULT_DELETE_IF_VALUE_STRING; /** * Creates a new JDBC key-value repository. A {@link DataSource} or {@link JdbcTemplate} must be set before @@ -169,6 +177,8 @@ protected void doInit() throws Exception { clearString = DEFAULT_CLEAR_STRING.replace(DEFAULT_TABLENAME, tableName); selectKeysString = DEFAULT_SELECT_KEYS_STRING.replace(DEFAULT_TABLENAME, tableName); deleteExpiredString = DEFAULT_DELETE_EXPIRED_STRING.replace(DEFAULT_TABLENAME, tableName); + updateIfValueString = DEFAULT_UPDATE_IF_VALUE_STRING.replace(DEFAULT_TABLENAME, tableName); + deleteIfValueString = DEFAULT_DELETE_IF_VALUE_STRING.replace(DEFAULT_TABLENAME, tableName); } } @@ -289,6 +299,53 @@ public Object putIfAbsent(String key, Object value, Duration ttl) { }); } + /** + * Atomically replaces the value for {@code key} only if the current stored value equals {@code expectedOldValue}. + *

+ * The comparison is performed on the serialized (byte[]) representation of the values. This works correctly for + * well-behaved {@link java.io.Serializable} types whose serialized form is deterministic (i.e. two independently + * serialized instances of the same logical value produce identical bytes). Classes that override + * {@code writeObject} with non-deterministic output (e.g. including timestamps or random salts) will not compare + * correctly and should not be used as expected values. + * + * @param key the key whose value should be conditionally replaced + * @param expectedOldValue the value that must currently be stored (compared by serialized bytes) + * @param newValue the new value to store if the condition is met + * @param ttl the time-to-live for the new entry, or {@code null} / zero for no expiry + * @return {@code true} if the value was replaced, {@code false} if the current value did not match + */ + @Override + public boolean replace(String key, Object expectedOldValue, Object newValue, Duration ttl) { + Boolean result = transactionTemplate.execute(status -> { + long expiresAt = toExpiresAt(ttl); + byte[] newBytes = KeyValueRepositoryHelper.serialize(newValue); + byte[] expectedBytes = KeyValueRepositoryHelper.serialize(expectedOldValue); + int updated = jdbcTemplate.update(getUpdateIfValueString(), newBytes, expiresAt, key, expectedBytes); + return updated > 0; + }); + return result != null && result; + } + + /** + * Atomically deletes the entry for {@code key} only if the current stored value equals {@code expectedValue}. + *

+ * The comparison is performed on the serialized (byte[]) representation of the value. See the note on + * {@link #replace(String, Object, Object, Duration)} for caveats about non-deterministic serialization. + * + * @param key the key to conditionally delete + * @param expectedValue the value that must currently be stored (compared by serialized bytes) + * @return {@code true} if the entry was deleted, {@code false} if the current value did not match + */ + @Override + public boolean delete(String key, Object expectedValue) { + Boolean result = transactionTemplate.execute(status -> { + byte[] expectedBytes = KeyValueRepositoryHelper.serialize(expectedValue); + int deleted = jdbcTemplate.update(getDeleteIfValueString(), key, expectedBytes); + return deleted > 0; + }); + return result != null && result; + } + @Override @ManagedAttribute(description = "The number of entries in the repository") public int size() { @@ -445,4 +502,20 @@ public String getDeleteExpiredString() { public void setDeleteExpiredString(String deleteExpiredString) { this.deleteExpiredString = deleteExpiredString; } + + public String getUpdateIfValueString() { + return updateIfValueString; + } + + public void setUpdateIfValueString(String updateIfValueString) { + this.updateIfValueString = updateIfValueString; + } + + public String getDeleteIfValueString() { + return deleteIfValueString; + } + + public void setDeleteIfValueString(String deleteIfValueString) { + this.deleteIfValueString = deleteIfValueString; + } } diff --git a/core/camel-core-engine/src/main/docs/modules/eips/pages/keyValueRepository.adoc b/core/camel-core-engine/src/main/docs/modules/eips/pages/keyValueRepository.adoc new file mode 100644 index 0000000000000..246bb9c0b5bde --- /dev/null +++ b/core/camel-core-engine/src/main/docs/modules/eips/pages/keyValueRepository.adoc @@ -0,0 +1,1077 @@ += KeyValueRepository Backends +:doctitle: KeyValueRepository Backends +:shortname: keyValueRepository +:description: Unified storage SPI backing the Cache EIP, Idempotent Consumer, and Aggregator patterns. Documents all available backends and how to configure them as beans. +:since: 4.23 +:supportlevel: Stable +:tabs-sync-option: + +`KeyValueRepository` is the unified storage SPI used by the +xref:cache-eip.adoc[Cache EIP], +xref:idempotentConsumer-eip.adoc[Idempotent Consumer], and +xref:aggregate-eip.adoc[Aggregator] patterns. +It provides a simple `get / put / delete / clear` contract — plus an optional atomic +`putIfAbsent` — that Camel maps onto many storage technologies through pluggable backend +implementations. + +Choosing the right backend is typically a matter of: + +* *Scope* — in-process memory, distributed in-memory, durable on-disk, or remote service. +* *Atomicity* — whether the backend natively provides an atomic Compare-And-Swap (CAS) + operation (required for exactly-once idempotency under concurrent load). +* *TTL* — whether expiration is handled natively by the store or requires client-side logic. +* *Operational footprint* — whether extra infrastructure (Redis, Kafka, Cassandra …) is + acceptable. + +== Available Backends + +[cols="2,3,3,1,1",options="header"] +|=== +| Backend class | Module | Key features | Atomic CAS | Native TTL + +| `MemoryKeyValueRepository` +| `camel-support` +| JVM-local `ConcurrentHashMap`; zero deps; auto-created when no KVR bean is registered +| ✓ +| ✗ + +| `CaffeineKeyValueRepository` +| `camel-caffeine` +| High-throughput in-process cache; configurable size; near-cache for distributed setups +| ✓ +| ✗ (client-side via Cache EIP `ttl`) + +| `EhcacheKeyValueRepository` +| `camel-ehcache` +| Ehcache 3; heap + off-heap + disk tiers; native TTL per cache config +| ✓ +| ✓ (Ehcache XML config) + +| `JCacheKeyValueRepository` +| `camel-jcache` +| JSR-107 provider-agnostic; works with Ehcache, Hazelcast, Infinispan, … +| ✓ (provider-dependent) +| ✓ (via `javax.cache.expiry`) + +| `HazelcastKeyValueRepository` +| `camel-hazelcast` +| Distributed, partitioned `IMap`; cluster-aware; near-cache optional +| ✓ +| ✓ (map TTL config) + +| `RedisKeyValueRepository` +| `camel-redis` +| Redis `SET`/`GET`; atomic via `SET NX`; native TTL via `EXPIRE` +| ✓ +| ✓ (native `EXPIRE`) + +| `InfinispanRemoteKeyValueRepository` +| `camel-infinispan` +| Hot Rod client to remote Infinispan/Data Grid cluster; distributed, transactional +| ✓ +| ✓ (per-entry lifespan) + +| `JdbcKeyValueRepository` +| `camel-sql` +| Any JDBC DataSource; portable; optional auto-DDL; ACID via DataSource transactions +| ✗ (non-atomic) +| ✗ (client-side) + +| `JpaKeyValueRepository` +| `camel-jpa` +| JPA entity-backed; works with any JPA 2 provider (Hibernate, EclipseLink…) +| ✗ (non-atomic) +| ✗ (client-side) + +| `CassandraKeyValueRepository` +| `camel-cassandraql` +| Cassandra wide-column store; configurable consistency levels; naturally distributed +| ✗ (non-atomic) +| ✓ (Cassandra TTL) + +| `KafkaKeyValueRepository` +| `camel-kafka` +| Kafka topic as a compacted key-value log; state rebuilt on startup; eventually consistent +| ✗ (non-atomic) +| ✗ (topic retention) + +|=== + +NOTE: _Atomic CAS_ means `putIfAbsent` is implemented without an external lock — required +for safe concurrent use as an Idempotent Repository. Non-atomic backends can still be used +with the Cache EIP and Aggregator when access is serialized or best-effort semantics are +acceptable. + +== Backend Configuration + +Each backend is registered as a named bean in the Camel registry and then referenced from +a route or EIP by name (or auto-discovered when only one bean is present). + +=== MemoryKeyValueRepository + +Provided by `camel-support` (always on the classpath). No extra dependencies. + +[tabs] +==== +Java DSL:: ++ +[source,java] +---- +import org.apache.camel.support.MemoryKeyValueRepository; + +@BindToRegistry("myMemoryKvr") +public KeyValueRepository myMemoryKvr() { + return new MemoryKeyValueRepository(); +} +---- + +YAML DSL:: ++ +[source,yaml] +---- +- beans: + - name: myMemoryKvr + type: org.apache.camel.support.MemoryKeyValueRepository +---- +==== + +Because `MemoryKeyValueRepository` has no required properties, registering it by type alone +is enough. When *no* `KeyValueRepository` bean is registered at all, the Cache EIP +auto-creates a private `MemoryKeyValueRepository` per block — no explicit bean needed. + +=== CaffeineKeyValueRepository + +[source,xml,role=dependency] +---- + + org.apache.camel + camel-caffeine + +---- + +Optional property: + +* `maximumSize` — maximum number of entries before eviction (default: unbounded). + +[tabs] +==== +Java DSL:: ++ +[source,java] +---- +import org.apache.camel.component.caffeine.cache.CaffeineKeyValueRepository; + +@BindToRegistry("caffeineKvr") +public KeyValueRepository caffeineKvr() { + CaffeineKeyValueRepository kvr = new CaffeineKeyValueRepository(); + kvr.setMaximumSize(10_000); + return kvr; +} +---- + +YAML DSL:: ++ +[source,yaml] +---- +- beans: + - name: caffeineKvr + type: org.apache.camel.component.caffeine.cache.CaffeineKeyValueRepository + properties: + maximumSize: 10000 +---- +==== + +TTL is not natively managed by Caffeine in this adapter. Use the Cache EIP `ttl` option +for time-based expiration, which wraps values with an expiry timestamp stored in the entry. + +=== EhcacheKeyValueRepository + +[source,xml,role=dependency] +---- + + org.apache.camel + camel-ehcache + +---- + +Required property: + +* `cacheManager` — a configured `org.ehcache.CacheManager` bean. + +Optional property: + +* `cacheName` — logical cache name within the `CacheManager` (default: `"camel-kvr"`). + +[tabs] +==== +Java DSL:: ++ +[source,java] +---- +import org.apache.camel.component.ehcache.EhcacheKeyValueRepository; +import org.ehcache.CacheManager; +import org.ehcache.config.builders.CacheConfigurationBuilder; +import org.ehcache.config.builders.CacheManagerBuilder; +import org.ehcache.config.builders.ResourcePoolsBuilder; +import org.ehcache.expiry.ExpiryPolicy; + +import java.time.Duration; + +@BindToRegistry("ehcacheManager") +public CacheManager ehcacheManager() { + return CacheManagerBuilder.newCacheManagerBuilder() + .withCache("products", + CacheConfigurationBuilder + .newCacheConfigurationBuilder(String.class, Object.class, + ResourcePoolsBuilder.heap(10_000)) + .withExpiry(ExpiryPolicy.BASE_EXPIRY // or a custom ExpiryPolicy + .timeToLiveExpiration(Duration.ofMinutes(30))) + ) + .build(true); // true = init on build +} + +@BindToRegistry("ehcacheKvr") +public KeyValueRepository ehcacheKvr(CacheManager ehcacheManager) { + EhcacheKeyValueRepository kvr = new EhcacheKeyValueRepository(); + kvr.setCacheManager(ehcacheManager); + kvr.setCacheName("products"); + return kvr; +} +---- + +YAML DSL:: ++ +[source,yaml] +---- +- beans: + - name: ehcacheKvr + type: org.apache.camel.component.ehcache.EhcacheKeyValueRepository + properties: + cacheManager: "#bean:ehcacheManager" + cacheName: products +---- +==== + +Native TTL is configured on the `CacheManager` / `CacheConfiguration`, not on the +`KeyValueRepository` bean itself. + +=== JCacheKeyValueRepository + +[source,xml,role=dependency] +---- + + org.apache.camel + camel-jcache + + + + org.ehcache + ehcache + +---- + +Optional properties: + +* `configuration` — a `javax.cache.configuration.Configuration` (or `MutableConfiguration`) bean. +* `cacheName` — name of the JCache cache (default: `"camel-kvr"`). + +[tabs] +==== +Java DSL:: ++ +[source,java] +---- +import org.apache.camel.component.jcache.JCacheKeyValueRepository; +import javax.cache.configuration.MutableConfiguration; +import javax.cache.expiry.CreatedExpiryPolicy; +import javax.cache.expiry.Duration; + +@BindToRegistry("jcacheConfig") +public MutableConfiguration jcacheConfig() { + return new MutableConfiguration() + .setTypes(String.class, Object.class) + .setExpiryPolicyFactory( + CreatedExpiryPolicy.factoryOf(new Duration(java.util.concurrent.TimeUnit.MINUTES, 30))) + .setStatisticsEnabled(true); +} + +@BindToRegistry("jcacheKvr") +public KeyValueRepository jcacheKvr(MutableConfiguration jcacheConfig) { + JCacheKeyValueRepository kvr = new JCacheKeyValueRepository(); + kvr.setConfiguration(jcacheConfig); + kvr.setCacheName("products"); + return kvr; +} +---- + +YAML DSL:: ++ +[source,yaml] +---- +- beans: + - name: jcacheKvr + type: org.apache.camel.component.jcache.JCacheKeyValueRepository + properties: + cacheName: products + configuration: "#bean:jcacheConfig" +---- +==== + +Whether `putIfAbsent` is truly atomic depends on the JSR-107 provider. Most production +providers (Ehcache 3, Hazelcast, Infinispan) implement it atomically. + +=== HazelcastKeyValueRepository + +[source,xml,role=dependency] +---- + + org.apache.camel + camel-hazelcast + +---- + +Optional properties: + +* `hazelcastInstance` — an existing `HazelcastInstance` bean (default: auto-created). +* `mapName` — name of the distributed `IMap` (default: `"camel-kvr"`). + +[tabs] +==== +Java DSL:: ++ +[source,java] +---- +import org.apache.camel.component.hazelcast.HazelcastKeyValueRepository; +import com.hazelcast.core.Hazelcast; +import com.hazelcast.core.HazelcastInstance; +import com.hazelcast.config.Config; +import com.hazelcast.config.MapConfig; + +@BindToRegistry("hazelcastInstance") +public HazelcastInstance hazelcastInstance() { + Config config = new Config(); + config.addMapConfig(new MapConfig("camel-products") + .setTimeToLiveSeconds(1800)); // 30-minute TTL + return Hazelcast.newHazelcastInstance(config); +} + +@BindToRegistry("hazelcastKvr") +public KeyValueRepository hazelcastKvr(HazelcastInstance hazelcastInstance) { + HazelcastKeyValueRepository kvr = new HazelcastKeyValueRepository(); + kvr.setHazelcastInstance(hazelcastInstance); + kvr.setMapName("camel-products"); + return kvr; +} +---- + +YAML DSL:: ++ +[source,yaml] +---- +- beans: + - name: hazelcastKvr + type: org.apache.camel.component.hazelcast.HazelcastKeyValueRepository + properties: + hazelcastInstance: "#bean:hazelcastInstance" + mapName: camel-products +---- +==== + +TTL is configured on the Hazelcast `MapConfig`, not on the `KeyValueRepository`. +`putIfAbsent` maps to `IMap.putIfAbsent`, which is atomic across the cluster. + +=== RedisKeyValueRepository + +[source,xml,role=dependency] +---- + + org.apache.camel + camel-redis + +---- + +Required property: + +* `endpoint` — a `camel-redis` endpoint URI string (e.g. `"redis://localhost:6379"`). + +Optional properties: + +* `keyPrefix` — a string prefix applied to all keys (useful to namespace multiple stores in one Redis DB). +* `redisson` — a pre-configured `RedissonClient` bean (alternative to `endpoint`). + +[tabs] +==== +Java DSL:: ++ +[source,java] +---- +import org.apache.camel.component.redis.RedisKeyValueRepository; + +@BindToRegistry("redisKvr") +public KeyValueRepository redisKvr() { + RedisKeyValueRepository kvr = new RedisKeyValueRepository(); + kvr.setEndpoint("redis://localhost:6379"); + kvr.setKeyPrefix("myapp:"); + return kvr; +} +---- + +With a custom `RedissonClient`:: ++ +[source,java] +---- +import org.redisson.Redisson; +import org.redisson.api.RedissonClient; +import org.redisson.config.Config; + +@BindToRegistry("redissonClient") +public RedissonClient redissonClient() { + Config config = new Config(); + config.useSingleServer() + .setAddress("redis://localhost:6379") + .setPassword("secret"); + return Redisson.create(config); +} + +@BindToRegistry("redisKvr") +public KeyValueRepository redisKvr(RedissonClient redissonClient) { + RedisKeyValueRepository kvr = new RedisKeyValueRepository(); + kvr.setRedisson(redissonClient); + kvr.setKeyPrefix("myapp:"); + return kvr; +} +---- + +YAML DSL:: ++ +[source,yaml] +---- +- beans: + - name: redisKvr + type: org.apache.camel.component.redis.RedisKeyValueRepository + properties: + endpoint: "redis://localhost:6379" + keyPrefix: "myapp:" +---- +==== + +Redis provides native TTL via `EXPIRE` / `SET … EX`. `putIfAbsent` maps to `SET NX`, which +is atomic. The Cache EIP `ttl` option is respected and propagated to the underlying +`EXPIRE` call. + +=== InfinispanRemoteKeyValueRepository + +[source,xml,role=dependency] +---- + + org.apache.camel + camel-infinispan + +---- + +Required property: + +* `cacheName` — the remote Infinispan cache name. + +Optional properties: + +* `configuration` — a `org.infinispan.client.hotrod.configuration.Configuration` bean. +* `cacheContainer` — a pre-built `RemoteCacheManager` bean (takes priority over `configuration`). + +[tabs] +==== +Java DSL:: ++ +[source,java] +---- +import org.apache.camel.component.infinispan.remote.InfinispanRemoteKeyValueRepository; +import org.infinispan.client.hotrod.RemoteCacheManager; +import org.infinispan.client.hotrod.configuration.ConfigurationBuilder; + +@BindToRegistry("infinispanManager") +public RemoteCacheManager infinispanManager() { + return new RemoteCacheManager( + new ConfigurationBuilder() + .addServer().host("infinispan-host").port(11222) + .security().authentication() + .username("camel").password("secret") + .build()); +} + +@BindToRegistry("infinispanKvr") +public KeyValueRepository infinispanKvr(RemoteCacheManager infinispanManager) { + InfinispanRemoteKeyValueRepository kvr = new InfinispanRemoteKeyValueRepository(); + kvr.setCacheContainer(infinispanManager); + kvr.setCacheName("products"); + return kvr; +} +---- + +YAML DSL:: ++ +[source,yaml] +---- +- beans: + - name: infinispanKvr + type: org.apache.camel.component.infinispan.remote.InfinispanRemoteKeyValueRepository + properties: + cacheContainer: "#bean:infinispanManager" + cacheName: products +---- +==== + +Infinispan supports per-entry lifespan and max-idle TTL natively. `putIfAbsent` is +implemented via the `PUTIFABSENT` Hot Rod operation, which is atomic. + +=== JdbcKeyValueRepository + +[source,xml,role=dependency] +---- + + org.apache.camel + camel-sql + +---- + +Required property: + +* `dataSource` — a `javax.sql.DataSource` bean. + +Optional properties: + +* `tableName` — the table used for storage (default: `"camel_kvr"`). +* `createTableIfNotExists` — auto-create the table at startup if absent (default: `true`). + +The auto-created table schema: + +[source,sql] +---- +CREATE TABLE camel_kvr ( + kvr_key VARCHAR(255) NOT NULL PRIMARY KEY, + kvr_value TEXT +); +---- + +[tabs] +==== +Java DSL:: ++ +[source,java] +---- +import org.apache.camel.component.sql.JdbcKeyValueRepository; +import org.springframework.jdbc.datasource.DriverManagerDataSource; +import javax.sql.DataSource; + +@BindToRegistry("myDataSource") +public DataSource myDataSource() { + DriverManagerDataSource ds = new DriverManagerDataSource(); + ds.setDriverClassName("org.postgresql.Driver"); + ds.setUrl("jdbc:postgresql://localhost:5432/mydb"); + ds.setUsername("camel"); + ds.setPassword("secret"); + return ds; +} + +@BindToRegistry("jdbcKvr") +public KeyValueRepository jdbcKvr(DataSource myDataSource) { + JdbcKeyValueRepository kvr = new JdbcKeyValueRepository(); + kvr.setDataSource(myDataSource); + kvr.setTableName("product_cache"); + kvr.setCreateTableIfNotExists(true); + return kvr; +} +---- + +YAML DSL:: ++ +[source,yaml] +---- +- beans: + - name: jdbcKvr + type: org.apache.camel.component.sql.JdbcKeyValueRepository + properties: + dataSource: "#bean:myDataSource" + tableName: product_cache + createTableIfNotExists: true +---- +==== + +NOTE: `JdbcKeyValueRepository` does not implement atomic `putIfAbsent` — concurrent inserts +may produce duplicate-key exceptions that are caught and treated as a "key already exists" +condition. This is safe for idempotency at low concurrency, but for high-throughput +exactly-once semantics, prefer a backend with native CAS. + +=== JpaKeyValueRepository + +[source,xml,role=dependency] +---- + + org.apache.camel + camel-jpa + +---- + +Required property: + +* `entityManagerFactory` — a `javax.persistence.EntityManagerFactory` bean. + +Optional properties: + +* `joinTransaction` — whether to participate in an existing JTA transaction (default: `true`). +* `sharedEntityManager` — use a shared/thread-bound `EntityManager` (Spring integration, default: `false`). + +[tabs] +==== +Java DSL:: ++ +[source,java] +---- +import org.apache.camel.component.jpa.JpaKeyValueRepository; +import javax.persistence.EntityManagerFactory; +import javax.persistence.Persistence; + +@BindToRegistry("emf") +public EntityManagerFactory emf() { + return Persistence.createEntityManagerFactory("myPersistenceUnit"); +} + +@BindToRegistry("jpaKvr") +public KeyValueRepository jpaKvr(EntityManagerFactory emf) { + JpaKeyValueRepository kvr = new JpaKeyValueRepository(); + kvr.setEntityManagerFactory(emf); + kvr.setJoinTransaction(false); // standalone, no JTA + return kvr; +} +---- + +YAML DSL:: ++ +[source,yaml] +---- +- beans: + - name: jpaKvr + type: org.apache.camel.component.jpa.JpaKeyValueRepository + properties: + entityManagerFactory: "#bean:emf" + joinTransaction: false +---- +==== + +NOTE: Like `JdbcKeyValueRepository`, the JPA backend does not provide atomic +`putIfAbsent`. Idempotency under concurrent load relies on unique-constraint violations +being caught at the database level. + +=== CassandraKeyValueRepository + +[source,xml,role=dependency] +---- + + org.apache.camel + camel-cassandraql + +---- + +Required property: + +* `session` — a `com.datastax.oss.driver.api.core.CqlSession` bean. + +Optional properties: + +* `table` — Cassandra table name (default: `"camel_kvr"`). +* `readConsistencyLevel` — `ConsistencyLevel` for reads (default: `LOCAL_ONE`). +* `writeConsistencyLevel` — `ConsistencyLevel` for writes (default: `LOCAL_ONE`). + +[tabs] +==== +Java DSL:: ++ +[source,java] +---- +import org.apache.camel.component.cassandra.CassandraKeyValueRepository; +import com.datastax.oss.driver.api.core.CqlSession; +import com.datastax.oss.driver.api.core.ConsistencyLevel; +import java.net.InetSocketAddress; + +@BindToRegistry("cassandraSession") +public CqlSession cassandraSession() { + return CqlSession.builder() + .addContactPoint(new InetSocketAddress("cassandra-host", 9042)) + .withLocalDatacenter("datacenter1") + .withKeyspace("myapp") + .build(); +} + +@BindToRegistry("cassandraKvr") +public KeyValueRepository cassandraKvr(CqlSession cassandraSession) { + CassandraKeyValueRepository kvr = new CassandraKeyValueRepository(); + kvr.setSession(cassandraSession); + kvr.setTable("product_cache"); + kvr.setReadConsistencyLevel(ConsistencyLevel.LOCAL_QUORUM); + kvr.setWriteConsistencyLevel(ConsistencyLevel.LOCAL_QUORUM); + return kvr; +} +---- + +YAML DSL:: ++ +[source,yaml] +---- +- beans: + - name: cassandraKvr + type: org.apache.camel.component.cassandra.CassandraKeyValueRepository + properties: + session: "#bean:cassandraSession" + table: product_cache + readConsistencyLevel: LOCAL_QUORUM + writeConsistencyLevel: LOCAL_QUORUM +---- +==== + +Cassandra supports native TTL via the `USING TTL` clause on `INSERT`/`UPDATE`. +The Cache EIP `ttl` option is forwarded to the Cassandra write operation as a native TTL, +so entries expire automatically at the storage layer. + +`putIfAbsent` uses Cassandra's lightweight transaction (`INSERT … IF NOT EXISTS`). +Although this is CAS at the Cassandra level, it is _not_ classified as fully atomic in the +table above because it is subject to Paxos latency and only provides linearizability within +a single partition. + +=== KafkaKeyValueRepository + +[source,xml,role=dependency] +---- + + org.apache.camel + camel-kafka + +---- + +Required properties: + +* `topic` — the compacted Kafka topic used as the key-value store. +* `bootstrapServers` — comma-separated list of Kafka broker addresses. + +Optional properties: + +* `maxCacheSize` — maximum number of entries to hold in the local in-memory replica (default: unlimited). +* `pollDurationMs` — poll timeout in milliseconds when rebuilding state at startup (default: `100`). +* `startupOnly` — if `true`, the consumer only reads the topic once at startup and does not keep polling (default: `false`). +* `groupId` — Kafka consumer group ID (default: auto-generated). + +[tabs] +==== +Java DSL:: ++ +[source,java] +---- +import org.apache.camel.component.kafka.KafkaKeyValueRepository; + +@BindToRegistry("kafkaKvr") +public KeyValueRepository kafkaKvr() { + KafkaKeyValueRepository kvr = new KafkaKeyValueRepository(); + kvr.setTopic("camel-kvr-products"); + kvr.setBootstrapServers("kafka1:9092,kafka2:9092"); + kvr.setMaxCacheSize(50_000); + kvr.setGroupId("camel-kvr-consumer"); + return kvr; +} +---- + +YAML DSL:: ++ +[source,yaml] +---- +- beans: + - name: kafkaKvr + type: org.apache.camel.component.kafka.KafkaKeyValueRepository + properties: + topic: camel-kvr-products + bootstrapServers: "kafka1:9092,kafka2:9092" + maxCacheSize: 50000 + groupId: camel-kvr-consumer +---- +==== + +IMPORTANT: The Kafka backend reconstructs its state by replaying the topic log on startup. +This means startup time scales with topic size. Use a +https://kafka.apache.org/documentation/#topicconfigs_cleanup.policy[log-compacted topic] +(`cleanup.policy=compact`) to keep the log bounded. + +TTL is not natively supported — entry retention is governed by topic-level +`retention.ms` / `retention.bytes`, which affects all entries equally, not individual ones. +`putIfAbsent` is *not* atomic: reads and writes go through the local in-memory replica with +no distributed locking. + +== Atomicity Guarantees + +The `putIfAbsent(key, value)` method is the critical operation for the +xref:idempotentConsumer-eip.adoc[Idempotent Consumer] — it must return `true` only for the +_first_ caller and `false` for all subsequent callers with the same key. + +[cols="2,1,3",options="header"] +|=== +| Backend | Atomic CAS | Mechanism + +| `MemoryKeyValueRepository` +| ✓ +| `ConcurrentHashMap.putIfAbsent` — JVM-local, lock-free + +| `CaffeineKeyValueRepository` +| ✓ +| Caffeine's internal lock-free `putIfAbsent` + +| `EhcacheKeyValueRepository` +| ✓ +| `Cache.putIfAbsent` (JSR-107) — segment-locked + +| `JCacheKeyValueRepository` +| ✓ (provider) +| Delegates to the JSR-107 provider's `putIfAbsent` + +| `HazelcastKeyValueRepository` +| ✓ +| `IMap.putIfAbsent` — distributed CAS with CP subsystem optional + +| `RedisKeyValueRepository` +| ✓ +| `SET key value NX` — single-command atomic in Redis + +| `InfinispanRemoteKeyValueRepository` +| ✓ +| Hot Rod `PUTIFABSENT` operation — cluster-wide atomic + +| `JdbcKeyValueRepository` +| ✗ +| Optimistic INSERT + unique-key constraint catch + +| `JpaKeyValueRepository` +| ✗ +| Optimistic INSERT + constraint catch (provider-dependent) + +| `CassandraKeyValueRepository` +| ✗ +| `INSERT … IF NOT EXISTS` (Paxos, not linearizable across restarts) + +| `KafkaKeyValueRepository` +| ✗ +| In-memory replica check — no distributed coordination + +|=== + +For non-atomic backends, use the Cache EIP (where idempotency is not required) or accept +at-least-once semantics in idempotency-sensitive flows. Alternatively, front a non-atomic +backend with a distributed lock (e.g. via `camel-hazelcast` `ILock`). + +== TTL Behavior + +[cols="2,1,3",options="header"] +|=== +| Backend | Native TTL | Notes + +| `MemoryKeyValueRepository` +| ✗ +| Cache EIP wraps values with an expiry timestamp; expired entries are evicted on read + +| `CaffeineKeyValueRepository` +| ✗ +| Same as Memory — client-side expiry envelope + +| `EhcacheKeyValueRepository` +| ✓ +| `ExpiryPolicy` configured on the `CacheManager`; Cache EIP TTL forwarded when possible + +| `JCacheKeyValueRepository` +| ✓ (provider) +| `javax.cache.expiry.ExpiryPolicy` on the `MutableConfiguration` + +| `HazelcastKeyValueRepository` +| ✓ +| `MapConfig.timeToLiveSeconds`; per-entry TTL also available via `IMap.put(k, v, ttl, unit)` + +| `RedisKeyValueRepository` +| ✓ +| `EXPIRE` set atomically alongside each `PUT`; Cache EIP `ttl` forwarded as `EX` seconds + +| `InfinispanRemoteKeyValueRepository` +| ✓ +| Per-entry `lifespan` (max-idle also available); Cache EIP `ttl` forwarded + +| `JdbcKeyValueRepository` +| ✗ +| Client-side expiry envelope; a periodic cleanup query is needed to purge expired rows + +| `JpaKeyValueRepository` +| ✗ +| Client-side expiry envelope; schedule a `DELETE … WHERE expires < now()` query + +| `CassandraKeyValueRepository` +| ✓ +| `USING TTL ` clause; Cache EIP `ttl` forwarded; expired rows deleted by Cassandra + +| `KafkaKeyValueRepository` +| ✗ +| Topic-level retention only; all entries expire at the same time based on `retention.ms` + +|=== + +NOTE: For backends *without native TTL*, the Cache EIP stores an expiry timestamp inside +the serialized value and skips (evicts on read) entries whose timestamp has passed. This +means the storage layer may accumulate stale entries that are never explicitly deleted — +size them accordingly or run periodic cleanup jobs. + +== Using Adapters: Idempotent Consumer and Aggregator + +A `KeyValueRepository` can be adapted to the specialized repository interfaces required by +the Idempotent Consumer and Aggregator EIPs via two adapter classes. + +=== KeyValueIdempotentRepository + +`KeyValueIdempotentRepository` wraps any `KeyValueRepository` as a +`org.apache.camel.spi.IdempotentRepository`, making it usable with +xref:idempotentConsumer-eip.adoc[Idempotent Consumer]. + +[tabs] +==== +Java DSL:: ++ +[source,java] +---- +import org.apache.camel.support.KeyValueIdempotentRepository; +import org.apache.camel.component.redis.RedisKeyValueRepository; + +@BindToRegistry("redisKvr") +public KeyValueRepository redisKvr() { + RedisKeyValueRepository kvr = new RedisKeyValueRepository(); + kvr.setEndpoint("redis://localhost:6379"); + kvr.setKeyPrefix("idempotent:"); + return kvr; +} + +@BindToRegistry("idempotentRepo") +public IdempotentRepository idempotentRepo(KeyValueRepository redisKvr) { + return new KeyValueIdempotentRepository(redisKvr); +} +---- ++ +Then reference it in a route: ++ +[source,java] +---- +from("jms:queue:orders") + .idempotentConsumer(header("JMSMessageID")) + .idempotentRepository("idempotentRepo") + .to("direct:processOrder"); +---- + +XML DSL:: ++ +[source,xml] +---- + + + +

JMSMessageID
+ + + +---- + +YAML DSL:: ++ +[source,yaml] +---- +- from: + uri: jms:queue:orders + steps: + - idempotentConsumer: + header: JMSMessageID + idempotentRepository: idempotentRepo + steps: + - to: direct:processOrder +---- +==== + +=== KeyValueAggregationRepository + +`KeyValueAggregationRepository` wraps any `KeyValueRepository` as a +`org.apache.camel.spi.AggregationRepository`, making it usable with +xref:aggregate-eip.adoc[Aggregator]. + +[tabs] +==== +Java DSL:: ++ +[source,java] +---- +import org.apache.camel.support.KeyValueAggregationRepository; +import org.apache.camel.component.hazelcast.HazelcastKeyValueRepository; + +@BindToRegistry("hazelcastKvr") +public KeyValueRepository hazelcastKvr() { + HazelcastKeyValueRepository kvr = new HazelcastKeyValueRepository(); + kvr.setMapName("aggregation-store"); + return kvr; +} + +@BindToRegistry("aggregationRepo") +public AggregationRepository aggregationRepo(KeyValueRepository hazelcastKvr) { + return new KeyValueAggregationRepository(hazelcastKvr); +} +---- ++ +Then reference it in a route: ++ +[source,java] +---- +from("direct:start") + .aggregate(header("orderId"), new GroupedBodyAggregationStrategy()) + .aggregationRepository("aggregationRepo") + .completionSize(10) + .to("direct:process"); +---- + +XML DSL:: ++ +[source,xml] +---- + + + + +
orderId
+ + + + +---- + +YAML DSL:: ++ +[source,yaml] +---- +- from: + uri: direct:start + steps: + - aggregate: + correlationExpression: + header: orderId + strategyRef: groupedBodyStrategy + aggregationRepository: aggregationRepo + completionSize: 10 + steps: + - to: direct:process +---- +==== + +NOTE: `KeyValueAggregationRepository` serializes the entire `Exchange` to the backing +store. For backends without native TTL (JDBC, JPA, Kafka), aggregation-in-progress entries +persist until completion. Use a backend with durable storage (JDBC, JPA, Infinispan) when +crash recovery of in-flight aggregations is required. + +== See Also + +* xref:cache-eip.adoc[Cache EIP] +* xref:idempotentConsumer-eip.adoc[Idempotent Consumer EIP] +* xref:aggregate-eip.adoc[Aggregator EIP] From 88ce48d2907343763fed16be5170b323bf2a1dc5 Mon Sep 17 00:00:00 2001 From: Guillaume Nodet Date: Thu, 3 Sep 2026 13:26:03 +0000 Subject: [PATCH 3/4] CAMEL-24463: Add distributed deployment guide and fix atomicity table MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - Add 'Distributed Deployment' section to keyValueRepository.adoc with a table showing which backends are safe for clustered Camel deployments (shared state, distributed CAS, practical recommendations per pattern) - Fix atomicity table: JDBC and Cassandra now correctly marked as atomic CAS (✓) reflecting the CAS overrides added in this PR and existing LWT support --- .../eips/pages/keyValueRepository.adoc | 94 ++++++++++++++++++- 1 file changed, 89 insertions(+), 5 deletions(-) diff --git a/core/camel-core-engine/src/main/docs/modules/eips/pages/keyValueRepository.adoc b/core/camel-core-engine/src/main/docs/modules/eips/pages/keyValueRepository.adoc index 246bb9c0b5bde..d12e4d332e720 100644 --- a/core/camel-core-engine/src/main/docs/modules/eips/pages/keyValueRepository.adoc +++ b/core/camel-core-engine/src/main/docs/modules/eips/pages/keyValueRepository.adoc @@ -841,16 +841,16 @@ _first_ caller and `false` for all subsequent callers with the same key. | Hot Rod `PUTIFABSENT` operation — cluster-wide atomic | `JdbcKeyValueRepository` -| ✗ -| Optimistic INSERT + unique-key constraint catch +| ✓ +| `UPDATE … WHERE value = ?` — single-statement CAS within a transaction | `JpaKeyValueRepository` | ✗ -| Optimistic INSERT + constraint catch (provider-dependent) +| Optimistic INSERT + constraint catch; `replace` / `delete(key, expected)` fall back to non-atomic defaults | `CassandraKeyValueRepository` -| ✗ -| `INSERT … IF NOT EXISTS` (Paxos, not linearizable across restarts) +| ✓ +| Lightweight transactions: `INSERT … IF NOT EXISTS`, `UPDATE … IF value = ?`, `DELETE … IF value = ?` | `KafkaKeyValueRepository` | ✗ @@ -919,6 +919,90 @@ the serialized value and skips (evicts on read) entries whose timestamp has pass means the storage layer may accumulate stale entries that are never explicitly deleted — size them accordingly or run periodic cleanup jobs. +== Distributed Deployment + +When running multiple Camel instances (e.g. behind a load balancer or in a Kubernetes cluster), the choice of +backend determines whether state is shared and whether CAS operations are safe across nodes. + +[cols="2,1,1,1,3",options="header"] +|=== +| Backend | Shared state | Distributed CAS | Distributed-safe | Notes + +| `MemoryKeyValueRepository` +| ✗ +| ✗ +| ❌ +| JVM-local only. Each node has its own isolated store. + +| `CaffeineKeyValueRepository` +| ✗ +| ✗ +| ❌ +| JVM-local only. Suitable as a near-cache in front of a distributed backend. + +| `EhcacheKeyValueRepository` +| ✗ (unless clustered) +| ✗ +| ❌ +| JVM-local by default. Ehcache clustering requires additional configuration outside of this integration. + +| `JCacheKeyValueRepository` +| Depends on provider +| Depends on provider +| ⚠️ +| With a distributed provider (Hazelcast, Infinispan), state is shared. CAS atomicity depends on the provider. + +| `HazelcastKeyValueRepository` +| ✓ +| ✓ +| ✅ +| Distributed `IMap` with cluster-wide `putIfAbsent`, `replace`, and `remove(key, value)`. + +| `RedisKeyValueRepository` +| ✓ +| ✓ +| ✅ +| Shared Redis server. `setIfAbsent` and `compareAndSet` are atomic. `replace` with TTL has a brief non-atomic window (see Javadoc). + +| `InfinispanRemoteKeyValueRepository` +| ✓ +| ✓ +| ✅ +| Remote cache with version-based optimistic CAS via `replaceWithVersion` / `removeWithVersion`. + +| `JdbcKeyValueRepository` +| ✓ +| ✓ +| ✅ +| Shared database. `putIfAbsent` uses unique-key constraint; `replace` and `delete(key, expected)` use SQL `WHERE value = ?`. + +| `JpaKeyValueRepository` +| ✓ +| ⚠️ +| ⚠️ +| Shared database. `putIfAbsent` catches constraint violations, but `replace` and `delete(key, expected)` fall back to non-atomic defaults (read-compare-write in Java). + +| `CassandraKeyValueRepository` +| ✓ +| ✓ +| ✅ +| Shared cluster. Full lightweight transaction (LWT) support: `IF NOT EXISTS`, `IF value = ?`. + +| `KafkaKeyValueRepository` +| ✓ (eventually) +| ✗ +| ⚠️ +| State is shared via the topic, but CAS is local-only. Suitable for best-effort deduplication, not strict exactly-once across nodes. + +|=== + +*Recommendations:* + +* **Idempotent Consumer in a cluster** — use Hazelcast, Redis, Infinispan, JDBC, or Cassandra for guaranteed exactly-once deduplication. +* **Aggregator in a cluster** — use any shared-state backend (recovery requires the same backend to be visible from the recovering node). +* **Cache EIP in a cluster** — any backend works. JVM-local backends give each node its own cache (fine for read-through caching); distributed backends share cached values. +* **Single-node deployment** — any backend works, including Memory and Caffeine. + == Using Adapters: Idempotent Consumer and Aggregator A `KeyValueRepository` can be adapted to the specialized repository interfaces required by From 08fc5da0c104e8caed3e27d6e77174e6d59d6be2 Mon Sep 17 00:00:00 2001 From: Guillaume Nodet Date: Thu, 3 Sep 2026 13:58:08 +0000 Subject: [PATCH 4/4] Regen --- .../catalog/beans/JdbcKeyValueRepository.json | 2 +- .../org/apache/camel/catalog/docs.properties | 1 + .../catalog/docs/keyValueRepository.adoc | 1161 +++++++++++++++++ .../kafka/KafkaKeyValueRepository.java | 36 +- .../JdbcKeyValueRepositoryConfigurer.java | 12 + .../camel/bean/JdbcKeyValueRepository.json | 2 +- .../src/main/docs/modules/eips/nav.adoc | 1 + 7 files changed, 1195 insertions(+), 20 deletions(-) create mode 100644 catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/keyValueRepository.adoc diff --git a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/beans/JdbcKeyValueRepository.json b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/beans/JdbcKeyValueRepository.json index a38b669e8cca9..07bc8667d6659 100644 --- a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/beans/JdbcKeyValueRepository.json +++ b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/beans/JdbcKeyValueRepository.json @@ -10,7 +10,7 @@ "groupId": "org.apache.camel", "artifactId": "camel-sql", "version": "4.23.0-SNAPSHOT", - "properties": { "jdbcTemplate": { "index": 0, "kind": "property", "displayName": "Jdbc Template", "required": true, "type": "object", "javaType": "org.springframework.jdbc.core.JdbcTemplate", "deprecated": false, "autowired": false, "secret": false, "description": "The Spring JdbcTemplate to use for connecting to the database" }, "transactionTemplate": { "index": 1, "kind": "property", "displayName": "Transaction Template", "required": true, "type": "object", "javaType": "org.springframework.transaction.support.TransactionTemplate", "deprecated": false, "autowired": false, "secret": false, "description": "The Spring TransactionTemplate to use for connecting to the database" }, "tableName": { "index": 2, "kind": "property", "displayName": "Table Name", "required": false, "type": "string", "javaType": "java.lang.String", "deprecated": false, "autowired": false, "secret": false, "defaultValue": "CAMEL_KEYVALUE", "description": "The name of the table to use in the database" }, "createTableIfNotExists": { "index": 3, "kind": "property", "displayName": "Create Table If Not Exists", "required": false, "type": "boolean", "javaType": "boolean", "deprecated": false, "autowired": false, "secret": false, "defaultValue": true, "description": "Whether to create the table in the database if none exists on startup" }, "tableExistsString": { "index": 4, "kind": "property", "displayName": "Table Exists String", "label": "advanced", "required": false, "type": "string", "javaType": "java.lang.String", "deprecated": false, "autowired": false, "secret": false, "description": "SQL query to use for checking if table exists" }, "createString": { "index": 5, "kind": "property", "displayName": "Create String", "label": "advanced", "required": false, "type": "string", "javaType": "java.lang.String", "deprecated": false, "autowired": false, "secret": false, "description": "SQL query to use for creating table" }, "selectString": { "index": 6, "kind": "property", "displayName": "Select String", "label": "advanced", "required": false, "type": "string", "javaType": "java.lang.String", "deprecated": false, "autowired": false, "secret": false, "description": "SQL query to use for selecting a value by key" }, "insertString": { "index": 7, "kind": "property", "displayName": "Insert String", "label": "advanced", "required": false, "type": "string", "javaType": "java.lang.String", "deprecated": false, "autowired": false, "secret": false, "description": "SQL query to use for inserting a new entry" }, "deleteString": { "index": 8, "kind": "property", "displayName": "Delete String", "label": "advanced", "required": false, "type": "string", "javaType": "java.lang.String", "deprecated": false, "autowired": false, "secret": false, "description": "SQL query to use for deleting an entry by key" }, "clearString": { "index": 9, "kind": "property", "displayName": "Clear String", "label": "advanced", "required": false, "type": "string", "javaType": "java.lang.String", "deprecated": false, "autowired": false, "secret": false, "description": "SQL query to delete all entries from the table" }, "selectKeysString": { "index": 10, "kind": "property", "displayName": "Select Keys String", "label": "advanced", "required": false, "type": "string", "javaType": "java.lang.String", "deprecated": false, "autowired": false, "secret": false, "description": "SQL query to use for selecting all non-expired keys" }, "deleteExpiredString": { "index": 11, "kind": "property", "displayName": "Delete Expired String", "label": "advanced", "required": false, "type": "string", "javaType": "java.lang.String", "deprecated": false, "autowired": false, "secret": false, "description": "SQL query to use for deleting expired entries" } } + "properties": { "jdbcTemplate": { "index": 0, "kind": "property", "displayName": "Jdbc Template", "required": true, "type": "object", "javaType": "org.springframework.jdbc.core.JdbcTemplate", "deprecated": false, "autowired": false, "secret": false, "description": "The Spring JdbcTemplate to use for connecting to the database" }, "transactionTemplate": { "index": 1, "kind": "property", "displayName": "Transaction Template", "required": true, "type": "object", "javaType": "org.springframework.transaction.support.TransactionTemplate", "deprecated": false, "autowired": false, "secret": false, "description": "The Spring TransactionTemplate to use for connecting to the database" }, "tableName": { "index": 2, "kind": "property", "displayName": "Table Name", "required": false, "type": "string", "javaType": "java.lang.String", "deprecated": false, "autowired": false, "secret": false, "defaultValue": "CAMEL_KEYVALUE", "description": "The name of the table to use in the database" }, "createTableIfNotExists": { "index": 3, "kind": "property", "displayName": "Create Table If Not Exists", "required": false, "type": "boolean", "javaType": "boolean", "deprecated": false, "autowired": false, "secret": false, "defaultValue": true, "description": "Whether to create the table in the database if none exists on startup" }, "tableExistsString": { "index": 4, "kind": "property", "displayName": "Table Exists String", "label": "advanced", "required": false, "type": "string", "javaType": "java.lang.String", "deprecated": false, "autowired": false, "secret": false, "description": "SQL query to use for checking if table exists" }, "createString": { "index": 5, "kind": "property", "displayName": "Create String", "label": "advanced", "required": false, "type": "string", "javaType": "java.lang.String", "deprecated": false, "autowired": false, "secret": false, "description": "SQL query to use for creating table" }, "selectString": { "index": 6, "kind": "property", "displayName": "Select String", "label": "advanced", "required": false, "type": "string", "javaType": "java.lang.String", "deprecated": false, "autowired": false, "secret": false, "description": "SQL query to use for selecting a value by key" }, "insertString": { "index": 7, "kind": "property", "displayName": "Insert String", "label": "advanced", "required": false, "type": "string", "javaType": "java.lang.String", "deprecated": false, "autowired": false, "secret": false, "description": "SQL query to use for inserting a new entry" }, "deleteString": { "index": 8, "kind": "property", "displayName": "Delete String", "label": "advanced", "required": false, "type": "string", "javaType": "java.lang.String", "deprecated": false, "autowired": false, "secret": false, "description": "SQL query to use for deleting an entry by key" }, "clearString": { "index": 9, "kind": "property", "displayName": "Clear String", "label": "advanced", "required": false, "type": "string", "javaType": "java.lang.String", "deprecated": false, "autowired": false, "secret": false, "description": "SQL query to delete all entries from the table" }, "selectKeysString": { "index": 10, "kind": "property", "displayName": "Select Keys String", "label": "advanced", "required": false, "type": "string", "javaType": "java.lang.String", "deprecated": false, "autowired": false, "secret": false, "description": "SQL query to use for selecting all non-expired keys" }, "deleteExpiredString": { "index": 11, "kind": "property", "displayName": "Delete Expired String", "label": "advanced", "required": false, "type": "string", "javaType": "java.lang.String", "deprecated": false, "autowired": false, "secret": false, "description": "SQL query to use for deleting expired entries" }, "updateIfValueString": { "index": 12, "kind": "property", "displayName": "Update If Value String", "label": "advanced", "required": false, "type": "string", "javaType": "java.lang.String", "deprecated": false, "autowired": false, "secret": false, "description": "SQL query to use for conditional update (CAS replace)" }, "deleteIfValueString": { "index": 13, "kind": "property", "displayName": "Delete If Value String", "label": "advanced", "required": false, "type": "string", "javaType": "java.lang.String", "deprecated": false, "autowired": false, "secret": false, "description": "SQL query to use for conditional delete (CAS delete)" } } } } diff --git a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs.properties b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs.properties index 554fa8908a3d9..835c5e512f642 100644 --- a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs.properties +++ b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs.properties @@ -335,6 +335,7 @@ kamelet-component kamelet-eip kamelet-main kamelet-main-support +keyValueRepository keycloak-component keycloak-consumer keycloak-producer diff --git a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/keyValueRepository.adoc b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/keyValueRepository.adoc new file mode 100644 index 0000000000000..d12e4d332e720 --- /dev/null +++ b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/keyValueRepository.adoc @@ -0,0 +1,1161 @@ += KeyValueRepository Backends +:doctitle: KeyValueRepository Backends +:shortname: keyValueRepository +:description: Unified storage SPI backing the Cache EIP, Idempotent Consumer, and Aggregator patterns. Documents all available backends and how to configure them as beans. +:since: 4.23 +:supportlevel: Stable +:tabs-sync-option: + +`KeyValueRepository` is the unified storage SPI used by the +xref:cache-eip.adoc[Cache EIP], +xref:idempotentConsumer-eip.adoc[Idempotent Consumer], and +xref:aggregate-eip.adoc[Aggregator] patterns. +It provides a simple `get / put / delete / clear` contract — plus an optional atomic +`putIfAbsent` — that Camel maps onto many storage technologies through pluggable backend +implementations. + +Choosing the right backend is typically a matter of: + +* *Scope* — in-process memory, distributed in-memory, durable on-disk, or remote service. +* *Atomicity* — whether the backend natively provides an atomic Compare-And-Swap (CAS) + operation (required for exactly-once idempotency under concurrent load). +* *TTL* — whether expiration is handled natively by the store or requires client-side logic. +* *Operational footprint* — whether extra infrastructure (Redis, Kafka, Cassandra …) is + acceptable. + +== Available Backends + +[cols="2,3,3,1,1",options="header"] +|=== +| Backend class | Module | Key features | Atomic CAS | Native TTL + +| `MemoryKeyValueRepository` +| `camel-support` +| JVM-local `ConcurrentHashMap`; zero deps; auto-created when no KVR bean is registered +| ✓ +| ✗ + +| `CaffeineKeyValueRepository` +| `camel-caffeine` +| High-throughput in-process cache; configurable size; near-cache for distributed setups +| ✓ +| ✗ (client-side via Cache EIP `ttl`) + +| `EhcacheKeyValueRepository` +| `camel-ehcache` +| Ehcache 3; heap + off-heap + disk tiers; native TTL per cache config +| ✓ +| ✓ (Ehcache XML config) + +| `JCacheKeyValueRepository` +| `camel-jcache` +| JSR-107 provider-agnostic; works with Ehcache, Hazelcast, Infinispan, … +| ✓ (provider-dependent) +| ✓ (via `javax.cache.expiry`) + +| `HazelcastKeyValueRepository` +| `camel-hazelcast` +| Distributed, partitioned `IMap`; cluster-aware; near-cache optional +| ✓ +| ✓ (map TTL config) + +| `RedisKeyValueRepository` +| `camel-redis` +| Redis `SET`/`GET`; atomic via `SET NX`; native TTL via `EXPIRE` +| ✓ +| ✓ (native `EXPIRE`) + +| `InfinispanRemoteKeyValueRepository` +| `camel-infinispan` +| Hot Rod client to remote Infinispan/Data Grid cluster; distributed, transactional +| ✓ +| ✓ (per-entry lifespan) + +| `JdbcKeyValueRepository` +| `camel-sql` +| Any JDBC DataSource; portable; optional auto-DDL; ACID via DataSource transactions +| ✗ (non-atomic) +| ✗ (client-side) + +| `JpaKeyValueRepository` +| `camel-jpa` +| JPA entity-backed; works with any JPA 2 provider (Hibernate, EclipseLink…) +| ✗ (non-atomic) +| ✗ (client-side) + +| `CassandraKeyValueRepository` +| `camel-cassandraql` +| Cassandra wide-column store; configurable consistency levels; naturally distributed +| ✗ (non-atomic) +| ✓ (Cassandra TTL) + +| `KafkaKeyValueRepository` +| `camel-kafka` +| Kafka topic as a compacted key-value log; state rebuilt on startup; eventually consistent +| ✗ (non-atomic) +| ✗ (topic retention) + +|=== + +NOTE: _Atomic CAS_ means `putIfAbsent` is implemented without an external lock — required +for safe concurrent use as an Idempotent Repository. Non-atomic backends can still be used +with the Cache EIP and Aggregator when access is serialized or best-effort semantics are +acceptable. + +== Backend Configuration + +Each backend is registered as a named bean in the Camel registry and then referenced from +a route or EIP by name (or auto-discovered when only one bean is present). + +=== MemoryKeyValueRepository + +Provided by `camel-support` (always on the classpath). No extra dependencies. + +[tabs] +==== +Java DSL:: ++ +[source,java] +---- +import org.apache.camel.support.MemoryKeyValueRepository; + +@BindToRegistry("myMemoryKvr") +public KeyValueRepository myMemoryKvr() { + return new MemoryKeyValueRepository(); +} +---- + +YAML DSL:: ++ +[source,yaml] +---- +- beans: + - name: myMemoryKvr + type: org.apache.camel.support.MemoryKeyValueRepository +---- +==== + +Because `MemoryKeyValueRepository` has no required properties, registering it by type alone +is enough. When *no* `KeyValueRepository` bean is registered at all, the Cache EIP +auto-creates a private `MemoryKeyValueRepository` per block — no explicit bean needed. + +=== CaffeineKeyValueRepository + +[source,xml,role=dependency] +---- + + org.apache.camel + camel-caffeine + +---- + +Optional property: + +* `maximumSize` — maximum number of entries before eviction (default: unbounded). + +[tabs] +==== +Java DSL:: ++ +[source,java] +---- +import org.apache.camel.component.caffeine.cache.CaffeineKeyValueRepository; + +@BindToRegistry("caffeineKvr") +public KeyValueRepository caffeineKvr() { + CaffeineKeyValueRepository kvr = new CaffeineKeyValueRepository(); + kvr.setMaximumSize(10_000); + return kvr; +} +---- + +YAML DSL:: ++ +[source,yaml] +---- +- beans: + - name: caffeineKvr + type: org.apache.camel.component.caffeine.cache.CaffeineKeyValueRepository + properties: + maximumSize: 10000 +---- +==== + +TTL is not natively managed by Caffeine in this adapter. Use the Cache EIP `ttl` option +for time-based expiration, which wraps values with an expiry timestamp stored in the entry. + +=== EhcacheKeyValueRepository + +[source,xml,role=dependency] +---- + + org.apache.camel + camel-ehcache + +---- + +Required property: + +* `cacheManager` — a configured `org.ehcache.CacheManager` bean. + +Optional property: + +* `cacheName` — logical cache name within the `CacheManager` (default: `"camel-kvr"`). + +[tabs] +==== +Java DSL:: ++ +[source,java] +---- +import org.apache.camel.component.ehcache.EhcacheKeyValueRepository; +import org.ehcache.CacheManager; +import org.ehcache.config.builders.CacheConfigurationBuilder; +import org.ehcache.config.builders.CacheManagerBuilder; +import org.ehcache.config.builders.ResourcePoolsBuilder; +import org.ehcache.expiry.ExpiryPolicy; + +import java.time.Duration; + +@BindToRegistry("ehcacheManager") +public CacheManager ehcacheManager() { + return CacheManagerBuilder.newCacheManagerBuilder() + .withCache("products", + CacheConfigurationBuilder + .newCacheConfigurationBuilder(String.class, Object.class, + ResourcePoolsBuilder.heap(10_000)) + .withExpiry(ExpiryPolicy.BASE_EXPIRY // or a custom ExpiryPolicy + .timeToLiveExpiration(Duration.ofMinutes(30))) + ) + .build(true); // true = init on build +} + +@BindToRegistry("ehcacheKvr") +public KeyValueRepository ehcacheKvr(CacheManager ehcacheManager) { + EhcacheKeyValueRepository kvr = new EhcacheKeyValueRepository(); + kvr.setCacheManager(ehcacheManager); + kvr.setCacheName("products"); + return kvr; +} +---- + +YAML DSL:: ++ +[source,yaml] +---- +- beans: + - name: ehcacheKvr + type: org.apache.camel.component.ehcache.EhcacheKeyValueRepository + properties: + cacheManager: "#bean:ehcacheManager" + cacheName: products +---- +==== + +Native TTL is configured on the `CacheManager` / `CacheConfiguration`, not on the +`KeyValueRepository` bean itself. + +=== JCacheKeyValueRepository + +[source,xml,role=dependency] +---- + + org.apache.camel + camel-jcache + + + + org.ehcache + ehcache + +---- + +Optional properties: + +* `configuration` — a `javax.cache.configuration.Configuration` (or `MutableConfiguration`) bean. +* `cacheName` — name of the JCache cache (default: `"camel-kvr"`). + +[tabs] +==== +Java DSL:: ++ +[source,java] +---- +import org.apache.camel.component.jcache.JCacheKeyValueRepository; +import javax.cache.configuration.MutableConfiguration; +import javax.cache.expiry.CreatedExpiryPolicy; +import javax.cache.expiry.Duration; + +@BindToRegistry("jcacheConfig") +public MutableConfiguration jcacheConfig() { + return new MutableConfiguration() + .setTypes(String.class, Object.class) + .setExpiryPolicyFactory( + CreatedExpiryPolicy.factoryOf(new Duration(java.util.concurrent.TimeUnit.MINUTES, 30))) + .setStatisticsEnabled(true); +} + +@BindToRegistry("jcacheKvr") +public KeyValueRepository jcacheKvr(MutableConfiguration jcacheConfig) { + JCacheKeyValueRepository kvr = new JCacheKeyValueRepository(); + kvr.setConfiguration(jcacheConfig); + kvr.setCacheName("products"); + return kvr; +} +---- + +YAML DSL:: ++ +[source,yaml] +---- +- beans: + - name: jcacheKvr + type: org.apache.camel.component.jcache.JCacheKeyValueRepository + properties: + cacheName: products + configuration: "#bean:jcacheConfig" +---- +==== + +Whether `putIfAbsent` is truly atomic depends on the JSR-107 provider. Most production +providers (Ehcache 3, Hazelcast, Infinispan) implement it atomically. + +=== HazelcastKeyValueRepository + +[source,xml,role=dependency] +---- + + org.apache.camel + camel-hazelcast + +---- + +Optional properties: + +* `hazelcastInstance` — an existing `HazelcastInstance` bean (default: auto-created). +* `mapName` — name of the distributed `IMap` (default: `"camel-kvr"`). + +[tabs] +==== +Java DSL:: ++ +[source,java] +---- +import org.apache.camel.component.hazelcast.HazelcastKeyValueRepository; +import com.hazelcast.core.Hazelcast; +import com.hazelcast.core.HazelcastInstance; +import com.hazelcast.config.Config; +import com.hazelcast.config.MapConfig; + +@BindToRegistry("hazelcastInstance") +public HazelcastInstance hazelcastInstance() { + Config config = new Config(); + config.addMapConfig(new MapConfig("camel-products") + .setTimeToLiveSeconds(1800)); // 30-minute TTL + return Hazelcast.newHazelcastInstance(config); +} + +@BindToRegistry("hazelcastKvr") +public KeyValueRepository hazelcastKvr(HazelcastInstance hazelcastInstance) { + HazelcastKeyValueRepository kvr = new HazelcastKeyValueRepository(); + kvr.setHazelcastInstance(hazelcastInstance); + kvr.setMapName("camel-products"); + return kvr; +} +---- + +YAML DSL:: ++ +[source,yaml] +---- +- beans: + - name: hazelcastKvr + type: org.apache.camel.component.hazelcast.HazelcastKeyValueRepository + properties: + hazelcastInstance: "#bean:hazelcastInstance" + mapName: camel-products +---- +==== + +TTL is configured on the Hazelcast `MapConfig`, not on the `KeyValueRepository`. +`putIfAbsent` maps to `IMap.putIfAbsent`, which is atomic across the cluster. + +=== RedisKeyValueRepository + +[source,xml,role=dependency] +---- + + org.apache.camel + camel-redis + +---- + +Required property: + +* `endpoint` — a `camel-redis` endpoint URI string (e.g. `"redis://localhost:6379"`). + +Optional properties: + +* `keyPrefix` — a string prefix applied to all keys (useful to namespace multiple stores in one Redis DB). +* `redisson` — a pre-configured `RedissonClient` bean (alternative to `endpoint`). + +[tabs] +==== +Java DSL:: ++ +[source,java] +---- +import org.apache.camel.component.redis.RedisKeyValueRepository; + +@BindToRegistry("redisKvr") +public KeyValueRepository redisKvr() { + RedisKeyValueRepository kvr = new RedisKeyValueRepository(); + kvr.setEndpoint("redis://localhost:6379"); + kvr.setKeyPrefix("myapp:"); + return kvr; +} +---- + +With a custom `RedissonClient`:: ++ +[source,java] +---- +import org.redisson.Redisson; +import org.redisson.api.RedissonClient; +import org.redisson.config.Config; + +@BindToRegistry("redissonClient") +public RedissonClient redissonClient() { + Config config = new Config(); + config.useSingleServer() + .setAddress("redis://localhost:6379") + .setPassword("secret"); + return Redisson.create(config); +} + +@BindToRegistry("redisKvr") +public KeyValueRepository redisKvr(RedissonClient redissonClient) { + RedisKeyValueRepository kvr = new RedisKeyValueRepository(); + kvr.setRedisson(redissonClient); + kvr.setKeyPrefix("myapp:"); + return kvr; +} +---- + +YAML DSL:: ++ +[source,yaml] +---- +- beans: + - name: redisKvr + type: org.apache.camel.component.redis.RedisKeyValueRepository + properties: + endpoint: "redis://localhost:6379" + keyPrefix: "myapp:" +---- +==== + +Redis provides native TTL via `EXPIRE` / `SET … EX`. `putIfAbsent` maps to `SET NX`, which +is atomic. The Cache EIP `ttl` option is respected and propagated to the underlying +`EXPIRE` call. + +=== InfinispanRemoteKeyValueRepository + +[source,xml,role=dependency] +---- + + org.apache.camel + camel-infinispan + +---- + +Required property: + +* `cacheName` — the remote Infinispan cache name. + +Optional properties: + +* `configuration` — a `org.infinispan.client.hotrod.configuration.Configuration` bean. +* `cacheContainer` — a pre-built `RemoteCacheManager` bean (takes priority over `configuration`). + +[tabs] +==== +Java DSL:: ++ +[source,java] +---- +import org.apache.camel.component.infinispan.remote.InfinispanRemoteKeyValueRepository; +import org.infinispan.client.hotrod.RemoteCacheManager; +import org.infinispan.client.hotrod.configuration.ConfigurationBuilder; + +@BindToRegistry("infinispanManager") +public RemoteCacheManager infinispanManager() { + return new RemoteCacheManager( + new ConfigurationBuilder() + .addServer().host("infinispan-host").port(11222) + .security().authentication() + .username("camel").password("secret") + .build()); +} + +@BindToRegistry("infinispanKvr") +public KeyValueRepository infinispanKvr(RemoteCacheManager infinispanManager) { + InfinispanRemoteKeyValueRepository kvr = new InfinispanRemoteKeyValueRepository(); + kvr.setCacheContainer(infinispanManager); + kvr.setCacheName("products"); + return kvr; +} +---- + +YAML DSL:: ++ +[source,yaml] +---- +- beans: + - name: infinispanKvr + type: org.apache.camel.component.infinispan.remote.InfinispanRemoteKeyValueRepository + properties: + cacheContainer: "#bean:infinispanManager" + cacheName: products +---- +==== + +Infinispan supports per-entry lifespan and max-idle TTL natively. `putIfAbsent` is +implemented via the `PUTIFABSENT` Hot Rod operation, which is atomic. + +=== JdbcKeyValueRepository + +[source,xml,role=dependency] +---- + + org.apache.camel + camel-sql + +---- + +Required property: + +* `dataSource` — a `javax.sql.DataSource` bean. + +Optional properties: + +* `tableName` — the table used for storage (default: `"camel_kvr"`). +* `createTableIfNotExists` — auto-create the table at startup if absent (default: `true`). + +The auto-created table schema: + +[source,sql] +---- +CREATE TABLE camel_kvr ( + kvr_key VARCHAR(255) NOT NULL PRIMARY KEY, + kvr_value TEXT +); +---- + +[tabs] +==== +Java DSL:: ++ +[source,java] +---- +import org.apache.camel.component.sql.JdbcKeyValueRepository; +import org.springframework.jdbc.datasource.DriverManagerDataSource; +import javax.sql.DataSource; + +@BindToRegistry("myDataSource") +public DataSource myDataSource() { + DriverManagerDataSource ds = new DriverManagerDataSource(); + ds.setDriverClassName("org.postgresql.Driver"); + ds.setUrl("jdbc:postgresql://localhost:5432/mydb"); + ds.setUsername("camel"); + ds.setPassword("secret"); + return ds; +} + +@BindToRegistry("jdbcKvr") +public KeyValueRepository jdbcKvr(DataSource myDataSource) { + JdbcKeyValueRepository kvr = new JdbcKeyValueRepository(); + kvr.setDataSource(myDataSource); + kvr.setTableName("product_cache"); + kvr.setCreateTableIfNotExists(true); + return kvr; +} +---- + +YAML DSL:: ++ +[source,yaml] +---- +- beans: + - name: jdbcKvr + type: org.apache.camel.component.sql.JdbcKeyValueRepository + properties: + dataSource: "#bean:myDataSource" + tableName: product_cache + createTableIfNotExists: true +---- +==== + +NOTE: `JdbcKeyValueRepository` does not implement atomic `putIfAbsent` — concurrent inserts +may produce duplicate-key exceptions that are caught and treated as a "key already exists" +condition. This is safe for idempotency at low concurrency, but for high-throughput +exactly-once semantics, prefer a backend with native CAS. + +=== JpaKeyValueRepository + +[source,xml,role=dependency] +---- + + org.apache.camel + camel-jpa + +---- + +Required property: + +* `entityManagerFactory` — a `javax.persistence.EntityManagerFactory` bean. + +Optional properties: + +* `joinTransaction` — whether to participate in an existing JTA transaction (default: `true`). +* `sharedEntityManager` — use a shared/thread-bound `EntityManager` (Spring integration, default: `false`). + +[tabs] +==== +Java DSL:: ++ +[source,java] +---- +import org.apache.camel.component.jpa.JpaKeyValueRepository; +import javax.persistence.EntityManagerFactory; +import javax.persistence.Persistence; + +@BindToRegistry("emf") +public EntityManagerFactory emf() { + return Persistence.createEntityManagerFactory("myPersistenceUnit"); +} + +@BindToRegistry("jpaKvr") +public KeyValueRepository jpaKvr(EntityManagerFactory emf) { + JpaKeyValueRepository kvr = new JpaKeyValueRepository(); + kvr.setEntityManagerFactory(emf); + kvr.setJoinTransaction(false); // standalone, no JTA + return kvr; +} +---- + +YAML DSL:: ++ +[source,yaml] +---- +- beans: + - name: jpaKvr + type: org.apache.camel.component.jpa.JpaKeyValueRepository + properties: + entityManagerFactory: "#bean:emf" + joinTransaction: false +---- +==== + +NOTE: Like `JdbcKeyValueRepository`, the JPA backend does not provide atomic +`putIfAbsent`. Idempotency under concurrent load relies on unique-constraint violations +being caught at the database level. + +=== CassandraKeyValueRepository + +[source,xml,role=dependency] +---- + + org.apache.camel + camel-cassandraql + +---- + +Required property: + +* `session` — a `com.datastax.oss.driver.api.core.CqlSession` bean. + +Optional properties: + +* `table` — Cassandra table name (default: `"camel_kvr"`). +* `readConsistencyLevel` — `ConsistencyLevel` for reads (default: `LOCAL_ONE`). +* `writeConsistencyLevel` — `ConsistencyLevel` for writes (default: `LOCAL_ONE`). + +[tabs] +==== +Java DSL:: ++ +[source,java] +---- +import org.apache.camel.component.cassandra.CassandraKeyValueRepository; +import com.datastax.oss.driver.api.core.CqlSession; +import com.datastax.oss.driver.api.core.ConsistencyLevel; +import java.net.InetSocketAddress; + +@BindToRegistry("cassandraSession") +public CqlSession cassandraSession() { + return CqlSession.builder() + .addContactPoint(new InetSocketAddress("cassandra-host", 9042)) + .withLocalDatacenter("datacenter1") + .withKeyspace("myapp") + .build(); +} + +@BindToRegistry("cassandraKvr") +public KeyValueRepository cassandraKvr(CqlSession cassandraSession) { + CassandraKeyValueRepository kvr = new CassandraKeyValueRepository(); + kvr.setSession(cassandraSession); + kvr.setTable("product_cache"); + kvr.setReadConsistencyLevel(ConsistencyLevel.LOCAL_QUORUM); + kvr.setWriteConsistencyLevel(ConsistencyLevel.LOCAL_QUORUM); + return kvr; +} +---- + +YAML DSL:: ++ +[source,yaml] +---- +- beans: + - name: cassandraKvr + type: org.apache.camel.component.cassandra.CassandraKeyValueRepository + properties: + session: "#bean:cassandraSession" + table: product_cache + readConsistencyLevel: LOCAL_QUORUM + writeConsistencyLevel: LOCAL_QUORUM +---- +==== + +Cassandra supports native TTL via the `USING TTL` clause on `INSERT`/`UPDATE`. +The Cache EIP `ttl` option is forwarded to the Cassandra write operation as a native TTL, +so entries expire automatically at the storage layer. + +`putIfAbsent` uses Cassandra's lightweight transaction (`INSERT … IF NOT EXISTS`). +Although this is CAS at the Cassandra level, it is _not_ classified as fully atomic in the +table above because it is subject to Paxos latency and only provides linearizability within +a single partition. + +=== KafkaKeyValueRepository + +[source,xml,role=dependency] +---- + + org.apache.camel + camel-kafka + +---- + +Required properties: + +* `topic` — the compacted Kafka topic used as the key-value store. +* `bootstrapServers` — comma-separated list of Kafka broker addresses. + +Optional properties: + +* `maxCacheSize` — maximum number of entries to hold in the local in-memory replica (default: unlimited). +* `pollDurationMs` — poll timeout in milliseconds when rebuilding state at startup (default: `100`). +* `startupOnly` — if `true`, the consumer only reads the topic once at startup and does not keep polling (default: `false`). +* `groupId` — Kafka consumer group ID (default: auto-generated). + +[tabs] +==== +Java DSL:: ++ +[source,java] +---- +import org.apache.camel.component.kafka.KafkaKeyValueRepository; + +@BindToRegistry("kafkaKvr") +public KeyValueRepository kafkaKvr() { + KafkaKeyValueRepository kvr = new KafkaKeyValueRepository(); + kvr.setTopic("camel-kvr-products"); + kvr.setBootstrapServers("kafka1:9092,kafka2:9092"); + kvr.setMaxCacheSize(50_000); + kvr.setGroupId("camel-kvr-consumer"); + return kvr; +} +---- + +YAML DSL:: ++ +[source,yaml] +---- +- beans: + - name: kafkaKvr + type: org.apache.camel.component.kafka.KafkaKeyValueRepository + properties: + topic: camel-kvr-products + bootstrapServers: "kafka1:9092,kafka2:9092" + maxCacheSize: 50000 + groupId: camel-kvr-consumer +---- +==== + +IMPORTANT: The Kafka backend reconstructs its state by replaying the topic log on startup. +This means startup time scales with topic size. Use a +https://kafka.apache.org/documentation/#topicconfigs_cleanup.policy[log-compacted topic] +(`cleanup.policy=compact`) to keep the log bounded. + +TTL is not natively supported — entry retention is governed by topic-level +`retention.ms` / `retention.bytes`, which affects all entries equally, not individual ones. +`putIfAbsent` is *not* atomic: reads and writes go through the local in-memory replica with +no distributed locking. + +== Atomicity Guarantees + +The `putIfAbsent(key, value)` method is the critical operation for the +xref:idempotentConsumer-eip.adoc[Idempotent Consumer] — it must return `true` only for the +_first_ caller and `false` for all subsequent callers with the same key. + +[cols="2,1,3",options="header"] +|=== +| Backend | Atomic CAS | Mechanism + +| `MemoryKeyValueRepository` +| ✓ +| `ConcurrentHashMap.putIfAbsent` — JVM-local, lock-free + +| `CaffeineKeyValueRepository` +| ✓ +| Caffeine's internal lock-free `putIfAbsent` + +| `EhcacheKeyValueRepository` +| ✓ +| `Cache.putIfAbsent` (JSR-107) — segment-locked + +| `JCacheKeyValueRepository` +| ✓ (provider) +| Delegates to the JSR-107 provider's `putIfAbsent` + +| `HazelcastKeyValueRepository` +| ✓ +| `IMap.putIfAbsent` — distributed CAS with CP subsystem optional + +| `RedisKeyValueRepository` +| ✓ +| `SET key value NX` — single-command atomic in Redis + +| `InfinispanRemoteKeyValueRepository` +| ✓ +| Hot Rod `PUTIFABSENT` operation — cluster-wide atomic + +| `JdbcKeyValueRepository` +| ✓ +| `UPDATE … WHERE value = ?` — single-statement CAS within a transaction + +| `JpaKeyValueRepository` +| ✗ +| Optimistic INSERT + constraint catch; `replace` / `delete(key, expected)` fall back to non-atomic defaults + +| `CassandraKeyValueRepository` +| ✓ +| Lightweight transactions: `INSERT … IF NOT EXISTS`, `UPDATE … IF value = ?`, `DELETE … IF value = ?` + +| `KafkaKeyValueRepository` +| ✗ +| In-memory replica check — no distributed coordination + +|=== + +For non-atomic backends, use the Cache EIP (where idempotency is not required) or accept +at-least-once semantics in idempotency-sensitive flows. Alternatively, front a non-atomic +backend with a distributed lock (e.g. via `camel-hazelcast` `ILock`). + +== TTL Behavior + +[cols="2,1,3",options="header"] +|=== +| Backend | Native TTL | Notes + +| `MemoryKeyValueRepository` +| ✗ +| Cache EIP wraps values with an expiry timestamp; expired entries are evicted on read + +| `CaffeineKeyValueRepository` +| ✗ +| Same as Memory — client-side expiry envelope + +| `EhcacheKeyValueRepository` +| ✓ +| `ExpiryPolicy` configured on the `CacheManager`; Cache EIP TTL forwarded when possible + +| `JCacheKeyValueRepository` +| ✓ (provider) +| `javax.cache.expiry.ExpiryPolicy` on the `MutableConfiguration` + +| `HazelcastKeyValueRepository` +| ✓ +| `MapConfig.timeToLiveSeconds`; per-entry TTL also available via `IMap.put(k, v, ttl, unit)` + +| `RedisKeyValueRepository` +| ✓ +| `EXPIRE` set atomically alongside each `PUT`; Cache EIP `ttl` forwarded as `EX` seconds + +| `InfinispanRemoteKeyValueRepository` +| ✓ +| Per-entry `lifespan` (max-idle also available); Cache EIP `ttl` forwarded + +| `JdbcKeyValueRepository` +| ✗ +| Client-side expiry envelope; a periodic cleanup query is needed to purge expired rows + +| `JpaKeyValueRepository` +| ✗ +| Client-side expiry envelope; schedule a `DELETE … WHERE expires < now()` query + +| `CassandraKeyValueRepository` +| ✓ +| `USING TTL ` clause; Cache EIP `ttl` forwarded; expired rows deleted by Cassandra + +| `KafkaKeyValueRepository` +| ✗ +| Topic-level retention only; all entries expire at the same time based on `retention.ms` + +|=== + +NOTE: For backends *without native TTL*, the Cache EIP stores an expiry timestamp inside +the serialized value and skips (evicts on read) entries whose timestamp has passed. This +means the storage layer may accumulate stale entries that are never explicitly deleted — +size them accordingly or run periodic cleanup jobs. + +== Distributed Deployment + +When running multiple Camel instances (e.g. behind a load balancer or in a Kubernetes cluster), the choice of +backend determines whether state is shared and whether CAS operations are safe across nodes. + +[cols="2,1,1,1,3",options="header"] +|=== +| Backend | Shared state | Distributed CAS | Distributed-safe | Notes + +| `MemoryKeyValueRepository` +| ✗ +| ✗ +| ❌ +| JVM-local only. Each node has its own isolated store. + +| `CaffeineKeyValueRepository` +| ✗ +| ✗ +| ❌ +| JVM-local only. Suitable as a near-cache in front of a distributed backend. + +| `EhcacheKeyValueRepository` +| ✗ (unless clustered) +| ✗ +| ❌ +| JVM-local by default. Ehcache clustering requires additional configuration outside of this integration. + +| `JCacheKeyValueRepository` +| Depends on provider +| Depends on provider +| ⚠️ +| With a distributed provider (Hazelcast, Infinispan), state is shared. CAS atomicity depends on the provider. + +| `HazelcastKeyValueRepository` +| ✓ +| ✓ +| ✅ +| Distributed `IMap` with cluster-wide `putIfAbsent`, `replace`, and `remove(key, value)`. + +| `RedisKeyValueRepository` +| ✓ +| ✓ +| ✅ +| Shared Redis server. `setIfAbsent` and `compareAndSet` are atomic. `replace` with TTL has a brief non-atomic window (see Javadoc). + +| `InfinispanRemoteKeyValueRepository` +| ✓ +| ✓ +| ✅ +| Remote cache with version-based optimistic CAS via `replaceWithVersion` / `removeWithVersion`. + +| `JdbcKeyValueRepository` +| ✓ +| ✓ +| ✅ +| Shared database. `putIfAbsent` uses unique-key constraint; `replace` and `delete(key, expected)` use SQL `WHERE value = ?`. + +| `JpaKeyValueRepository` +| ✓ +| ⚠️ +| ⚠️ +| Shared database. `putIfAbsent` catches constraint violations, but `replace` and `delete(key, expected)` fall back to non-atomic defaults (read-compare-write in Java). + +| `CassandraKeyValueRepository` +| ✓ +| ✓ +| ✅ +| Shared cluster. Full lightweight transaction (LWT) support: `IF NOT EXISTS`, `IF value = ?`. + +| `KafkaKeyValueRepository` +| ✓ (eventually) +| ✗ +| ⚠️ +| State is shared via the topic, but CAS is local-only. Suitable for best-effort deduplication, not strict exactly-once across nodes. + +|=== + +*Recommendations:* + +* **Idempotent Consumer in a cluster** — use Hazelcast, Redis, Infinispan, JDBC, or Cassandra for guaranteed exactly-once deduplication. +* **Aggregator in a cluster** — use any shared-state backend (recovery requires the same backend to be visible from the recovering node). +* **Cache EIP in a cluster** — any backend works. JVM-local backends give each node its own cache (fine for read-through caching); distributed backends share cached values. +* **Single-node deployment** — any backend works, including Memory and Caffeine. + +== Using Adapters: Idempotent Consumer and Aggregator + +A `KeyValueRepository` can be adapted to the specialized repository interfaces required by +the Idempotent Consumer and Aggregator EIPs via two adapter classes. + +=== KeyValueIdempotentRepository + +`KeyValueIdempotentRepository` wraps any `KeyValueRepository` as a +`org.apache.camel.spi.IdempotentRepository`, making it usable with +xref:idempotentConsumer-eip.adoc[Idempotent Consumer]. + +[tabs] +==== +Java DSL:: ++ +[source,java] +---- +import org.apache.camel.support.KeyValueIdempotentRepository; +import org.apache.camel.component.redis.RedisKeyValueRepository; + +@BindToRegistry("redisKvr") +public KeyValueRepository redisKvr() { + RedisKeyValueRepository kvr = new RedisKeyValueRepository(); + kvr.setEndpoint("redis://localhost:6379"); + kvr.setKeyPrefix("idempotent:"); + return kvr; +} + +@BindToRegistry("idempotentRepo") +public IdempotentRepository idempotentRepo(KeyValueRepository redisKvr) { + return new KeyValueIdempotentRepository(redisKvr); +} +---- ++ +Then reference it in a route: ++ +[source,java] +---- +from("jms:queue:orders") + .idempotentConsumer(header("JMSMessageID")) + .idempotentRepository("idempotentRepo") + .to("direct:processOrder"); +---- + +XML DSL:: ++ +[source,xml] +---- + + + +
JMSMessageID
+ + + +---- + +YAML DSL:: ++ +[source,yaml] +---- +- from: + uri: jms:queue:orders + steps: + - idempotentConsumer: + header: JMSMessageID + idempotentRepository: idempotentRepo + steps: + - to: direct:processOrder +---- +==== + +=== KeyValueAggregationRepository + +`KeyValueAggregationRepository` wraps any `KeyValueRepository` as a +`org.apache.camel.spi.AggregationRepository`, making it usable with +xref:aggregate-eip.adoc[Aggregator]. + +[tabs] +==== +Java DSL:: ++ +[source,java] +---- +import org.apache.camel.support.KeyValueAggregationRepository; +import org.apache.camel.component.hazelcast.HazelcastKeyValueRepository; + +@BindToRegistry("hazelcastKvr") +public KeyValueRepository hazelcastKvr() { + HazelcastKeyValueRepository kvr = new HazelcastKeyValueRepository(); + kvr.setMapName("aggregation-store"); + return kvr; +} + +@BindToRegistry("aggregationRepo") +public AggregationRepository aggregationRepo(KeyValueRepository hazelcastKvr) { + return new KeyValueAggregationRepository(hazelcastKvr); +} +---- ++ +Then reference it in a route: ++ +[source,java] +---- +from("direct:start") + .aggregate(header("orderId"), new GroupedBodyAggregationStrategy()) + .aggregationRepository("aggregationRepo") + .completionSize(10) + .to("direct:process"); +---- + +XML DSL:: ++ +[source,xml] +---- + + + + +
orderId
+ + + + +---- + +YAML DSL:: ++ +[source,yaml] +---- +- from: + uri: direct:start + steps: + - aggregate: + correlationExpression: + header: orderId + strategyRef: groupedBodyStrategy + aggregationRepository: aggregationRepo + completionSize: 10 + steps: + - to: direct:process +---- +==== + +NOTE: `KeyValueAggregationRepository` serializes the entire `Exchange` to the backing +store. For backends without native TTL (JDBC, JPA, Kafka), aggregation-in-progress entries +persist until completion. Use a backend with durable storage (JDBC, JPA, Infinispan) when +crash recovery of in-flight aggregations is required. + +== See Also + +* xref:cache-eip.adoc[Cache EIP] +* xref:idempotentConsumer-eip.adoc[Idempotent Consumer EIP] +* xref:aggregate-eip.adoc[Aggregator EIP] diff --git a/components/camel-kafka/src/main/java/org/apache/camel/processor/keyvalue/kafka/KafkaKeyValueRepository.java b/components/camel-kafka/src/main/java/org/apache/camel/processor/keyvalue/kafka/KafkaKeyValueRepository.java index a81ef5ffd7750..b793689919e86 100644 --- a/components/camel-kafka/src/main/java/org/apache/camel/processor/keyvalue/kafka/KafkaKeyValueRepository.java +++ b/components/camel-kafka/src/main/java/org/apache/camel/processor/keyvalue/kafka/KafkaKeyValueRepository.java @@ -81,13 +81,13 @@ * The topic used must be unique per logical repository. TTL is managed locally via expiration timestamps in the cache; * expired entries are lazily evicted on access. *

- * Note on TTL and topic growth: TTL expiration is managed locally in the LRU cache. The Kafka topic retains - * all entries (including expired ones) indefinitely. On restart, expired entries are replayed from the topic and - * re-populated in the cache with their original expiration timestamps (they will appear as expired immediately). - * To prevent unbounded topic growth, configure the topic with Kafka's built-in log compaction - * ({@code cleanup.policy=compact}) and optionally {@code min.compaction.lag.ms} / {@code delete.retention.ms} - * to control retention. Deleted entries (tombstones) are represented as records with a {@code DELETE} action byte, - * which log compaction will eventually remove. + * Note on TTL and topic growth: TTL expiration is managed locally in the LRU cache. The Kafka topic retains all + * entries (including expired ones) indefinitely. On restart, expired entries are replayed from the topic and + * re-populated in the cache with their original expiration timestamps (they will appear as expired immediately). To + * prevent unbounded topic growth, configure the topic with Kafka's built-in log compaction + * ({@code cleanup.policy=compact}) and optionally {@code min.compaction.lag.ms} / {@code delete.retention.ms} to + * control retention. Deleted entries (tombstones) are represented as records with a {@code DELETE} action byte, which + * log compaction will eventually remove. * * @since 4.23 */ @@ -255,18 +255,18 @@ public void clear() { * Returns the existing (non-expired) value if one was already mapped, or {@code null} if the insertion succeeded. * *

- * Note on atomicity: The check-then-act sequence is locally atomic with respect to other threads in this - * JVM (backed by {@link java.util.concurrent.ConcurrentMap#putIfAbsent}), but the subsequent broadcast to Kafka - * is a separate, non-atomic step. In a multi-instance deployment, two nodes may each observe the key as absent, - * both insert locally, and both broadcast — resulting in the last broadcast winning in the Kafka topic and - * propagating to all other instances. This implementation therefore does not provide distributed - * compare-and-swap (CAS) semantics; it is suitable only for best-effort deduplication within a single JVM or - * in scenarios where occasional duplicates across instances are acceptable. + * Note on atomicity: The check-then-act sequence is locally atomic with respect to other threads in this JVM + * (backed by {@link java.util.concurrent.ConcurrentMap#putIfAbsent}), but the subsequent broadcast to Kafka is a + * separate, non-atomic step. In a multi-instance deployment, two nodes may each observe the key as absent, both + * insert locally, and both broadcast — resulting in the last broadcast winning in the Kafka topic and propagating + * to all other instances. This implementation therefore does not provide distributed compare-and-swap + * (CAS) semantics; it is suitable only for best-effort deduplication within a single JVM or in scenarios where + * occasional duplicates across instances are acceptable. * - * @param key the key to insert - * @param value the value to associate - * @param ttl optional time-to-live; {@code null} or zero means no expiration - * @return the existing value if already present and not expired, or {@code null} if the entry was inserted + * @param key the key to insert + * @param value the value to associate + * @param ttl optional time-to-live; {@code null} or zero means no expiration + * @return the existing value if already present and not expired, or {@code null} if the entry was inserted */ @Override public Object putIfAbsent(String key, Object value, Duration ttl) { diff --git a/components/camel-sql/src/generated/java/org/apache/camel/processor/keyvalue/jdbc/JdbcKeyValueRepositoryConfigurer.java b/components/camel-sql/src/generated/java/org/apache/camel/processor/keyvalue/jdbc/JdbcKeyValueRepositoryConfigurer.java index 91704ed283bef..377981ec7fcc0 100644 --- a/components/camel-sql/src/generated/java/org/apache/camel/processor/keyvalue/jdbc/JdbcKeyValueRepositoryConfigurer.java +++ b/components/camel-sql/src/generated/java/org/apache/camel/processor/keyvalue/jdbc/JdbcKeyValueRepositoryConfigurer.java @@ -31,6 +31,8 @@ public boolean configure(CamelContext camelContext, Object obj, String name, Obj case "createTableIfNotExists": target.setCreateTableIfNotExists(property(camelContext, boolean.class, value)); return true; case "deleteexpiredstring": case "deleteExpiredString": target.setDeleteExpiredString(property(camelContext, java.lang.String.class, value)); return true; + case "deleteifvaluestring": + case "deleteIfValueString": target.setDeleteIfValueString(property(camelContext, java.lang.String.class, value)); return true; case "deletestring": case "deleteString": target.setDeleteString(property(camelContext, java.lang.String.class, value)); return true; case "insertstring": @@ -47,6 +49,8 @@ public boolean configure(CamelContext camelContext, Object obj, String name, Obj case "tableName": target.setTableName(property(camelContext, java.lang.String.class, value)); return true; case "transactiontemplate": case "transactionTemplate": target.setTransactionTemplate(property(camelContext, org.springframework.transaction.support.TransactionTemplate.class, value)); return true; + case "updateifvaluestring": + case "updateIfValueString": target.setUpdateIfValueString(property(camelContext, java.lang.String.class, value)); return true; default: return false; } } @@ -62,6 +66,8 @@ public Class getOptionType(String name, boolean ignoreCase) { case "createTableIfNotExists": return boolean.class; case "deleteexpiredstring": case "deleteExpiredString": return java.lang.String.class; + case "deleteifvaluestring": + case "deleteIfValueString": return java.lang.String.class; case "deletestring": case "deleteString": return java.lang.String.class; case "insertstring": @@ -78,6 +84,8 @@ public Class getOptionType(String name, boolean ignoreCase) { case "tableName": return java.lang.String.class; case "transactiontemplate": case "transactionTemplate": return org.springframework.transaction.support.TransactionTemplate.class; + case "updateifvaluestring": + case "updateIfValueString": return java.lang.String.class; default: return null; } } @@ -94,6 +102,8 @@ public Object getOptionValue(Object obj, String name, boolean ignoreCase) { case "createTableIfNotExists": return target.isCreateTableIfNotExists(); case "deleteexpiredstring": case "deleteExpiredString": return target.getDeleteExpiredString(); + case "deleteifvaluestring": + case "deleteIfValueString": return target.getDeleteIfValueString(); case "deletestring": case "deleteString": return target.getDeleteString(); case "insertstring": @@ -110,6 +120,8 @@ public Object getOptionValue(Object obj, String name, boolean ignoreCase) { case "tableName": return target.getTableName(); case "transactiontemplate": case "transactionTemplate": return target.getTransactionTemplate(); + case "updateifvaluestring": + case "updateIfValueString": return target.getUpdateIfValueString(); default: return null; } } diff --git a/components/camel-sql/src/generated/resources/META-INF/services/org/apache/camel/bean/JdbcKeyValueRepository.json b/components/camel-sql/src/generated/resources/META-INF/services/org/apache/camel/bean/JdbcKeyValueRepository.json index a38b669e8cca9..07bc8667d6659 100644 --- a/components/camel-sql/src/generated/resources/META-INF/services/org/apache/camel/bean/JdbcKeyValueRepository.json +++ b/components/camel-sql/src/generated/resources/META-INF/services/org/apache/camel/bean/JdbcKeyValueRepository.json @@ -10,7 +10,7 @@ "groupId": "org.apache.camel", "artifactId": "camel-sql", "version": "4.23.0-SNAPSHOT", - "properties": { "jdbcTemplate": { "index": 0, "kind": "property", "displayName": "Jdbc Template", "required": true, "type": "object", "javaType": "org.springframework.jdbc.core.JdbcTemplate", "deprecated": false, "autowired": false, "secret": false, "description": "The Spring JdbcTemplate to use for connecting to the database" }, "transactionTemplate": { "index": 1, "kind": "property", "displayName": "Transaction Template", "required": true, "type": "object", "javaType": "org.springframework.transaction.support.TransactionTemplate", "deprecated": false, "autowired": false, "secret": false, "description": "The Spring TransactionTemplate to use for connecting to the database" }, "tableName": { "index": 2, "kind": "property", "displayName": "Table Name", "required": false, "type": "string", "javaType": "java.lang.String", "deprecated": false, "autowired": false, "secret": false, "defaultValue": "CAMEL_KEYVALUE", "description": "The name of the table to use in the database" }, "createTableIfNotExists": { "index": 3, "kind": "property", "displayName": "Create Table If Not Exists", "required": false, "type": "boolean", "javaType": "boolean", "deprecated": false, "autowired": false, "secret": false, "defaultValue": true, "description": "Whether to create the table in the database if none exists on startup" }, "tableExistsString": { "index": 4, "kind": "property", "displayName": "Table Exists String", "label": "advanced", "required": false, "type": "string", "javaType": "java.lang.String", "deprecated": false, "autowired": false, "secret": false, "description": "SQL query to use for checking if table exists" }, "createString": { "index": 5, "kind": "property", "displayName": "Create String", "label": "advanced", "required": false, "type": "string", "javaType": "java.lang.String", "deprecated": false, "autowired": false, "secret": false, "description": "SQL query to use for creating table" }, "selectString": { "index": 6, "kind": "property", "displayName": "Select String", "label": "advanced", "required": false, "type": "string", "javaType": "java.lang.String", "deprecated": false, "autowired": false, "secret": false, "description": "SQL query to use for selecting a value by key" }, "insertString": { "index": 7, "kind": "property", "displayName": "Insert String", "label": "advanced", "required": false, "type": "string", "javaType": "java.lang.String", "deprecated": false, "autowired": false, "secret": false, "description": "SQL query to use for inserting a new entry" }, "deleteString": { "index": 8, "kind": "property", "displayName": "Delete String", "label": "advanced", "required": false, "type": "string", "javaType": "java.lang.String", "deprecated": false, "autowired": false, "secret": false, "description": "SQL query to use for deleting an entry by key" }, "clearString": { "index": 9, "kind": "property", "displayName": "Clear String", "label": "advanced", "required": false, "type": "string", "javaType": "java.lang.String", "deprecated": false, "autowired": false, "secret": false, "description": "SQL query to delete all entries from the table" }, "selectKeysString": { "index": 10, "kind": "property", "displayName": "Select Keys String", "label": "advanced", "required": false, "type": "string", "javaType": "java.lang.String", "deprecated": false, "autowired": false, "secret": false, "description": "SQL query to use for selecting all non-expired keys" }, "deleteExpiredString": { "index": 11, "kind": "property", "displayName": "Delete Expired String", "label": "advanced", "required": false, "type": "string", "javaType": "java.lang.String", "deprecated": false, "autowired": false, "secret": false, "description": "SQL query to use for deleting expired entries" } } + "properties": { "jdbcTemplate": { "index": 0, "kind": "property", "displayName": "Jdbc Template", "required": true, "type": "object", "javaType": "org.springframework.jdbc.core.JdbcTemplate", "deprecated": false, "autowired": false, "secret": false, "description": "The Spring JdbcTemplate to use for connecting to the database" }, "transactionTemplate": { "index": 1, "kind": "property", "displayName": "Transaction Template", "required": true, "type": "object", "javaType": "org.springframework.transaction.support.TransactionTemplate", "deprecated": false, "autowired": false, "secret": false, "description": "The Spring TransactionTemplate to use for connecting to the database" }, "tableName": { "index": 2, "kind": "property", "displayName": "Table Name", "required": false, "type": "string", "javaType": "java.lang.String", "deprecated": false, "autowired": false, "secret": false, "defaultValue": "CAMEL_KEYVALUE", "description": "The name of the table to use in the database" }, "createTableIfNotExists": { "index": 3, "kind": "property", "displayName": "Create Table If Not Exists", "required": false, "type": "boolean", "javaType": "boolean", "deprecated": false, "autowired": false, "secret": false, "defaultValue": true, "description": "Whether to create the table in the database if none exists on startup" }, "tableExistsString": { "index": 4, "kind": "property", "displayName": "Table Exists String", "label": "advanced", "required": false, "type": "string", "javaType": "java.lang.String", "deprecated": false, "autowired": false, "secret": false, "description": "SQL query to use for checking if table exists" }, "createString": { "index": 5, "kind": "property", "displayName": "Create String", "label": "advanced", "required": false, "type": "string", "javaType": "java.lang.String", "deprecated": false, "autowired": false, "secret": false, "description": "SQL query to use for creating table" }, "selectString": { "index": 6, "kind": "property", "displayName": "Select String", "label": "advanced", "required": false, "type": "string", "javaType": "java.lang.String", "deprecated": false, "autowired": false, "secret": false, "description": "SQL query to use for selecting a value by key" }, "insertString": { "index": 7, "kind": "property", "displayName": "Insert String", "label": "advanced", "required": false, "type": "string", "javaType": "java.lang.String", "deprecated": false, "autowired": false, "secret": false, "description": "SQL query to use for inserting a new entry" }, "deleteString": { "index": 8, "kind": "property", "displayName": "Delete String", "label": "advanced", "required": false, "type": "string", "javaType": "java.lang.String", "deprecated": false, "autowired": false, "secret": false, "description": "SQL query to use for deleting an entry by key" }, "clearString": { "index": 9, "kind": "property", "displayName": "Clear String", "label": "advanced", "required": false, "type": "string", "javaType": "java.lang.String", "deprecated": false, "autowired": false, "secret": false, "description": "SQL query to delete all entries from the table" }, "selectKeysString": { "index": 10, "kind": "property", "displayName": "Select Keys String", "label": "advanced", "required": false, "type": "string", "javaType": "java.lang.String", "deprecated": false, "autowired": false, "secret": false, "description": "SQL query to use for selecting all non-expired keys" }, "deleteExpiredString": { "index": 11, "kind": "property", "displayName": "Delete Expired String", "label": "advanced", "required": false, "type": "string", "javaType": "java.lang.String", "deprecated": false, "autowired": false, "secret": false, "description": "SQL query to use for deleting expired entries" }, "updateIfValueString": { "index": 12, "kind": "property", "displayName": "Update If Value String", "label": "advanced", "required": false, "type": "string", "javaType": "java.lang.String", "deprecated": false, "autowired": false, "secret": false, "description": "SQL query to use for conditional update (CAS replace)" }, "deleteIfValueString": { "index": 13, "kind": "property", "displayName": "Delete If Value String", "label": "advanced", "required": false, "type": "string", "javaType": "java.lang.String", "deprecated": false, "autowired": false, "secret": false, "description": "SQL query to use for conditional delete (CAS delete)" } } } } diff --git a/core/camel-core-engine/src/main/docs/modules/eips/nav.adoc b/core/camel-core-engine/src/main/docs/modules/eips/nav.adoc index 4f477263bb5ea..f4817f74d9ad4 100644 --- a/core/camel-core-engine/src/main/docs/modules/eips/nav.adoc +++ b/core/camel-core-engine/src/main/docs/modules/eips/nav.adoc @@ -39,6 +39,7 @@ ** xref:idempotentConsumer-eip.adoc[Idempotent Consumer] ** xref:intercept.adoc[Intercept] ** xref:kamelet-eip.adoc[Kamelet] +** xref:keyValueRepository.adoc[KeyValueRepository Backends] ** xref:loadBalance-eip.adoc[Load Balance] ** xref:log-eip.adoc[Logger] ** xref:loop-eip.adoc[Loop]