(camel) branch main updated: CAMEL-24463: Improve KeyValueRepository infrastructure (#26086)

[gnodet]

This is an automated email from the ASF dual-hosted git repository.

gnodet pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/camel.git


The following commit(s) were added to refs/heads/main by this push:
     new 2d6568c14c87 CAMEL-24463: Improve KeyValueRepository infrastructure 
(#26086)
2d6568c14c87 is described below

commit 2d6568c14c871e47d5496507c9c75ae995aee541
Author: Guillaume Nodet <gno...@gmail.com>
AuthorDate: Fri Sep 4 00:16:59 2026 +0200

    CAMEL-24463: Improve KeyValueRepository infrastructure (#26086)
    
    * 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
    
    * 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
    
    * CAMEL-24463: Add distributed deployment guide and fix atomicity table
    
    - 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
    
    * Regen
---
 .../catalog/beans/JdbcKeyValueRepository.json      |    2 +-
 .../org/apache/camel/catalog/docs.properties       |    1 +
 .../camel/catalog/docs/keyValueRepository.adoc     | 1161 ++++++++++++++++++++
 .../cassandra/CassandraKeyValueRepository.java     |    7 +
 .../processor/EhcacheKeyValueRepository.java       |   58 +-
 .../processor/EhcacheKeyValueRepositoryTest.java   |    3 +-
 .../jcache/processor/JCacheKeyValueRepository.java |   62 +-
 .../processor/JCacheKeyValueRepositoryTest.java    |    5 +-
 .../keyvalue/kafka/KafkaKeyValueRepository.java    |   26 +
 .../component/redis/RedisKeyValueRepository.java   |    6 +-
 .../jdbc/JdbcKeyValueRepositoryConfigurer.java     |   12 +
 .../apache/camel/bean/JdbcKeyValueRepository.json  |    2 +-
 .../keyvalue/jdbc/JdbcKeyValueRepository.java      |   73 ++
 .../src/main/docs/modules/eips/nav.adoc            |    1 +
 .../modules/eips/pages/keyValueRepository.adoc     | 1161 ++++++++++++++++++++
 .../apache/camel/processor/CacheProcessorTest.java |   24 +
 .../org/apache/camel/support/KeyValueTtlValue.java |   52 +
 .../support/KeyValueRepositoryHelperTest.java      |  319 ++++++
 18 files changed, 2889 insertions(+), 86 deletions(-)

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 a38b669e8cca..07bc8667d665 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.springf [...]
+    "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.springf [...]
   }
 }
 
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 554fa8908a3d..835c5e512f64 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 000000000000..d12e4d332e72
--- /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]
+----
+<dependency>
+  <groupId>org.apache.camel</groupId>
+  <artifactId>camel-caffeine</artifactId>
+</dependency>
+----
+
+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]
+----
+<dependency>
+  <groupId>org.apache.camel</groupId>
+  <artifactId>camel-ehcache</artifactId>
+</dependency>
+----
+
+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]
+----
+<dependency>
+  <groupId>org.apache.camel</groupId>
+  <artifactId>camel-jcache</artifactId>
+</dependency>
+<!-- Plus a JSR-107 provider, e.g.: -->
+<dependency>
+  <groupId>org.ehcache</groupId>
+  <artifactId>ehcache</artifactId>
+</dependency>
+----
+
+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<String, Object> jcacheConfig() {
+    return new MutableConfiguration<String, Object>()
+        .setTypes(String.class, Object.class)
+        .setExpiryPolicyFactory(
+            CreatedExpiryPolicy.factoryOf(new 
Duration(java.util.concurrent.TimeUnit.MINUTES, 30)))
+        .setStatisticsEnabled(true);
+}
+
+@BindToRegistry("jcacheKvr")
+public KeyValueRepository jcacheKvr(MutableConfiguration<String, Object> 
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]
+----
+<dependency>
+  <groupId>org.apache.camel</groupId>
+  <artifactId>camel-hazelcast</artifactId>
+</dependency>
+----
+
+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]
+----
+<dependency>
+  <groupId>org.apache.camel</groupId>
+  <artifactId>camel-redis</artifactId>
+</dependency>
+----
+
+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]
+----
+<dependency>
+  <groupId>org.apache.camel</groupId>
+  <artifactId>camel-infinispan</artifactId>
+</dependency>
+----
+
+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]
+----
+<dependency>
+  <groupId>org.apache.camel</groupId>
+  <artifactId>camel-sql</artifactId>
+</dependency>
+----
+
+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]
+----
+<dependency>
+  <groupId>org.apache.camel</groupId>
+  <artifactId>camel-jpa</artifactId>
+</dependency>
+----
+
+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]
+----
+<dependency>
+  <groupId>org.apache.camel</groupId>
+  <artifactId>camel-cassandraql</artifactId>
+</dependency>
+----
+
+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]
+----
+<dependency>
+  <groupId>org.apache.camel</groupId>
+  <artifactId>camel-kafka</artifactId>
+</dependency>
+----
+
+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 <seconds>` 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]
+----
+<route>
+  <from uri="jms:queue:orders"/>
+  <idempotentConsumer idempotentRepository="idempotentRepo">
+    <header>JMSMessageID</header>
+    <to uri="direct:processOrder"/>
+  </idempotentConsumer>
+</route>
+----
+
+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]
+----
+<route>
+  <from uri="direct:start"/>
+  <aggregate strategyRef="groupedBodyStrategy"
+             aggregationRepositoryRef="aggregationRepo"
+             completionSize="10">
+    <correlationExpression>
+      <header>orderId</header>
+    </correlationExpression>
+    <to uri="direct:process"/>
+  </aggregate>
+</route>
+----
+
+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-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 b365dd6f2fa2..74136447a9d4 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 class CassandraKeyValueRepository extends 
ServiceSupport implements KeyVa
         return buffer != null ? KeyValueRepositoryHelper.deserialize(buffer) : 
null;
     }
 
+    /**
+     * Stores a value with optional TTL.
+     * <p/>
+     * <b>Note:</b> 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 35849594320b..30fc8ae237ef 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.component.ehcache.EhcacheManager;
 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 @@ import org.ehcache.CacheManager;
  * A {@link KeyValueRepository} implementation backed by an Ehcache {@link 
Cache}.
  * <p/>
  * 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.
  * <p/>
  * 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 @@ import org.ehcache.CacheManager;
 @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<String, TtlValue> cache;
+    private Cache<String, KeyValueTtlValue> cache;
     private EhcacheManager ehcacheManager;
 
     @Metadata(description = "Name of cache", defaultValue = 
"EhcacheKeyValueRepository")
@@ -129,7 +103,7 @@ public class EhcacheKeyValueRepository extends 
ServiceSupport implements KeyValu
     @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;
         }
@@ -140,12 +114,20 @@ public class EhcacheKeyValueRepository extends 
ServiceSupport implements KeyValu
         return entry.value();
     }
 
+    /**
+     * Stores a value with optional TTL.
+     * <p/>
+     * <b>Note:</b> 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<K,V>} API.
+     */
     @Override
     @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 +137,7 @@ public class EhcacheKeyValueRepository extends 
ServiceSupport implements KeyValu
     @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 +148,7 @@ public class EhcacheKeyValueRepository extends 
ServiceSupport implements KeyValu
     @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 +162,9 @@ public class EhcacheKeyValueRepository extends 
ServiceSupport implements KeyValu
     @Override
     public Set<String> keys() {
         Set<String> keys = new HashSet<>();
-        Iterator<Cache.Entry<String, TtlValue>> it = cache.iterator();
+        Iterator<Cache.Entry<String, KeyValueTtlValue>> it = cache.iterator();
         while (it.hasNext()) {
-            Cache.Entry<String, TtlValue> entry = it.next();
+            Cache.Entry<String, KeyValueTtlValue> entry = it.next();
             if (!entry.getValue().isExpired()) {
                 keys.add(entry.getKey());
             } else {
@@ -213,7 +195,7 @@ public class EhcacheKeyValueRepository extends 
ServiceSupport implements KeyValu
         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 ce97b3abb13d..9dbde03ca1ae 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.time.Duration;
 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 @@ class EhcacheKeyValueRepositoryTest {
                 .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 85bf205f329a..5212d6fcb47c 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.component.jcache.JCacheManager;
 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 @@ import org.apache.camel.util.ObjectHelper;
  * A {@link KeyValueRepository} implementation backed by a JCache (JSR-107) 
{@link Cache}.
  * <p/>
  * 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.
  * <p/>
  * 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 @@ import org.apache.camel.util.ObjectHelper;
 @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<String, TtlValue> cache;
-    private JCacheManager<String, TtlValue> cacheManager;
+    private Cache<String, KeyValueTtlValue> cache;
+    private JCacheManager<String, KeyValueTtlValue> cacheManager;
 
     @Metadata(description = "Configuration for JCache")
     private JCacheConfiguration configuration;
@@ -115,11 +89,11 @@ public class JCacheKeyValueRepository extends 
ServiceSupport implements CamelCon
         this.configuration = configuration;
     }
 
-    public Cache<String, TtlValue> getCache() {
+    public Cache<String, KeyValueTtlValue> getCache() {
         return cache;
     }
 
-    public void setCache(Cache<String, TtlValue> cache) {
+    public void setCache(Cache<String, KeyValueTtlValue> cache) {
         this.cache = cache;
     }
 
@@ -135,7 +109,7 @@ public class JCacheKeyValueRepository extends 
ServiceSupport implements CamelCon
     @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;
         }
@@ -146,12 +120,20 @@ public class JCacheKeyValueRepository extends 
ServiceSupport implements CamelCon
         return entry.value();
     }
 
+    /**
+     * Stores a value with optional TTL.
+     * <p/>
+     * <b>Note:</b> 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) {
         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 +143,7 @@ public class JCacheKeyValueRepository extends 
ServiceSupport implements CamelCon
     @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 +154,7 @@ public class JCacheKeyValueRepository extends 
ServiceSupport implements CamelCon
     @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 +168,9 @@ public class JCacheKeyValueRepository extends 
ServiceSupport implements CamelCon
     @Override
     public Set<String> keys() {
         Set<String> keys = new HashSet<>();
-        Iterator<Cache.Entry<String, TtlValue>> it = cache.iterator();
+        Iterator<Cache.Entry<String, KeyValueTtlValue>> it = cache.iterator();
         while (it.hasNext()) {
-            Cache.Entry<String, TtlValue> entry = it.next();
+            Cache.Entry<String, KeyValueTtlValue> 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 f7077c08eb69..bdc009bc3213 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.JCacheConfiguration;
 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 @@ import static org.awaitility.Awaitility.await;
 @HazelcastTest
 class JCacheKeyValueRepositoryTest extends CamelTestSupport {
 
-    private JCacheManager<String, JCacheKeyValueRepository.TtlValue> 
cacheManager;
-    private Cache<String, JCacheKeyValueRepository.TtlValue> cache;
+    private JCacheManager<String, KeyValueTtlValue> cacheManager;
+    private Cache<String, KeyValueTtlValue> cache;
     private JCacheKeyValueRepository repository;
 
     @Override
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 5cc946841d02..b793689919e8 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 @@ import org.slf4j.LoggerFactory;
  * <p/>
  * 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.
+ * <p/>
+ * <b>Note on TTL and topic growth:</b> 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 class KafkaKeyValueRepository extends 
ServiceSupport implements KeyValueR
         }
     }
 
+    /**
+     * 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.
+     *
+     * <p/>
+     * <b>Note on atomicity:</b> 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 <em>not</em> 
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-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 3bedccad69fb..687194885cdb 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 class RedisKeyValueRepository extends 
ServiceSupport implements KeyValueR
     public @Nullable Object put(String key, Object value, Duration ttl) {
         RBucket<byte[]> 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/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 91704ed283be..377981ec7fcc 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 class JdbcKeyValueRepositoryConfigurer extends 
org.apache.camel.support.c
         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 class JdbcKeyValueRepositoryConfigurer extends 
org.apache.camel.support.c
         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 JdbcKeyValueRepositoryConfigurer extends 
org.apache.camel.support.c
         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 JdbcKeyValueRepositoryConfigurer extends 
org.apache.camel.support.c
         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 class JdbcKeyValueRepositoryConfigurer extends 
org.apache.camel.support.c
         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 class JdbcKeyValueRepositoryConfigurer extends 
org.apache.camel.support.c
         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 a38b669e8cca..07bc8667d665 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.springf [...]
+    "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.springf [...]
   }
 }
 
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 2988e16c9925..47eadd3efd77 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 @@ public class JdbcKeyValueRepository extends ServiceSupport 
implements KeyValueRe
             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 class JdbcKeyValueRepository extends ServiceSupport 
implements KeyValueRe
         });
     }
 
+    /**
+     * Atomically replaces the value for {@code key} only if the current 
stored value equals {@code expectedOldValue}.
+     * <p/>
+     * 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}.
+     * <p/>
+     * 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 class JdbcKeyValueRepository extends ServiceSupport 
implements KeyValueRe
     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/nav.adoc 
b/core/camel-core-engine/src/main/docs/modules/eips/nav.adoc
index 4f477263bb5e..f4817f74d9ad 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]
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 000000000000..d12e4d332e72
--- /dev/null
+++ 
b/core/camel-core-engine/src/main/docs/modules/eips/pages/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]
+----
+<dependency>
+  <groupId>org.apache.camel</groupId>
+  <artifactId>camel-caffeine</artifactId>
+</dependency>
+----
+
+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]
+----
+<dependency>
+  <groupId>org.apache.camel</groupId>
+  <artifactId>camel-ehcache</artifactId>
+</dependency>
+----
+
+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]
+----
+<dependency>
+  <groupId>org.apache.camel</groupId>
+  <artifactId>camel-jcache</artifactId>
+</dependency>
+<!-- Plus a JSR-107 provider, e.g.: -->
+<dependency>
+  <groupId>org.ehcache</groupId>
+  <artifactId>ehcache</artifactId>
+</dependency>
+----
+
+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<String, Object> jcacheConfig() {
+    return new MutableConfiguration<String, Object>()
+        .setTypes(String.class, Object.class)
+        .setExpiryPolicyFactory(
+            CreatedExpiryPolicy.factoryOf(new 
Duration(java.util.concurrent.TimeUnit.MINUTES, 30)))
+        .setStatisticsEnabled(true);
+}
+
+@BindToRegistry("jcacheKvr")
+public KeyValueRepository jcacheKvr(MutableConfiguration<String, Object> 
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]
+----
+<dependency>
+  <groupId>org.apache.camel</groupId>
+  <artifactId>camel-hazelcast</artifactId>
+</dependency>
+----
+
+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]
+----
+<dependency>
+  <groupId>org.apache.camel</groupId>
+  <artifactId>camel-redis</artifactId>
+</dependency>
+----
+
+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]
+----
+<dependency>
+  <groupId>org.apache.camel</groupId>
+  <artifactId>camel-infinispan</artifactId>
+</dependency>
+----
+
+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]
+----
+<dependency>
+  <groupId>org.apache.camel</groupId>
+  <artifactId>camel-sql</artifactId>
+</dependency>
+----
+
+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]
+----
+<dependency>
+  <groupId>org.apache.camel</groupId>
+  <artifactId>camel-jpa</artifactId>
+</dependency>
+----
+
+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]
+----
+<dependency>
+  <groupId>org.apache.camel</groupId>
+  <artifactId>camel-cassandraql</artifactId>
+</dependency>
+----
+
+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]
+----
+<dependency>
+  <groupId>org.apache.camel</groupId>
+  <artifactId>camel-kafka</artifactId>
+</dependency>
+----
+
+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 <seconds>` 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]
+----
+<route>
+  <from uri="jms:queue:orders"/>
+  <idempotentConsumer idempotentRepository="idempotentRepo">
+    <header>JMSMessageID</header>
+    <to uri="direct:processOrder"/>
+  </idempotentConsumer>
+</route>
+----
+
+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]
+----
+<route>
+  <from uri="direct:start"/>
+  <aggregate strategyRef="groupedBodyStrategy"
+             aggregationRepositoryRef="aggregationRepo"
+             completionSize="10">
+    <correlationExpression>
+      <header>orderId</header>
+    </correlationExpression>
+    <to uri="direct:process"/>
+  </aggregate>
+</route>
+----
+
+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/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 07a99beb1b05..eb3e385e3881 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 @@ class CacheProcessorTest extends ContextTestSupport {
         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 @@ class CacheProcessorTest extends ContextTestSupport {
                         .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 000000000000..59a46b2de3c0
--- /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 000000000000..0977b39c41b0
--- /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<String, Integer> 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<String> items;
+
+        ComplexPayload(String id, int quantity, List<String> 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 + "}";
+        }
+    }
+}