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 86ffe1bd5575 CAMEL-24463: Add Redis and Infinispan KeyValueRepository
implementations (#25818)
86ffe1bd5575 is described below
commit 86ffe1bd55758e6e30b8a7304d2db80193ebfc1e
Author: Guillaume Nodet <[email protected]>
AuthorDate: Wed Sep 2 10:42:32 2026 +0200
CAMEL-24463: Add Redis and Infinispan KeyValueRepository implementations
(#25818)
CAMEL-24463: Add Redis and Infinispan KeyValueRepository implementations
Co-authored-by: Claude Opus 4.6 <[email protected]>
---
.../org/apache/camel/catalog/beans.properties | 2 +
.../beans/InfinispanRemoteKeyValueRepository.json | 16 ++
.../catalog/beans/RedisKeyValueRepository.json | 16 ++
...finispanRemoteKeyValueRepositoryConfigurer.java | 60 ++++
.../services/org/apache/camel/bean.properties | 2 +-
.../bean/InfinispanRemoteKeyValueRepository.json | 16 ++
...ispan.remote.InfinispanRemoteKeyValueRepository | 2 +
.../remote/InfinispanRemoteKeyValueRepository.java | 266 ++++++++++++++++++
.../InfinispanRemoteKeyValueRepositoryIT.java | 308 +++++++++++++++++++++
components/camel-redis/pom.xml | 11 +
.../redis/RedisKeyValueRepositoryConfigurer.java | 57 ++++
.../services/org/apache/camel/bean.properties | 2 +-
.../apache/camel/bean/RedisKeyValueRepository.json | 16 ++
...e.camel.component.redis.RedisKeyValueRepository | 2 +
.../component/redis/RedisKeyValueRepository.java | 262 ++++++++++++++++++
.../integration/RedisKeyValueRepositoryIT.java | 299 ++++++++++++++++++++
16 files changed, 1335 insertions(+), 2 deletions(-)
diff --git
a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/beans.properties
b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/beans.properties
index ef75beaeb981..a7cd67fa6ec4 100644
---
a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/beans.properties
+++
b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/beans.properties
@@ -24,6 +24,7 @@ InfinispanEmbeddedIdempotentRepository
InfinispanRemoteAggregationRepository
InfinispanRemoteClusterService
InfinispanRemoteIdempotentRepository
+InfinispanRemoteKeyValueRepository
JCacheAggregationRepository
JCacheIdempotentRepository
JGroupsRaftClusterService
@@ -40,6 +41,7 @@ MemoryKeyValueRepository
MongoDbIdempotentRepository
OpensearchBulkRequestAggregationStrategy
RedisAggregationRepository
+RedisKeyValueRepository
SimpleScheduledRoutePolicy
SpringCacheIdempotentRepository
SpringRedisIdempotentRepository
diff --git
a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/beans/InfinispanRemoteKeyValueRepository.json
b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/beans/InfinispanRemoteKeyValueRepository.json
new file mode 100644
index 000000000000..7a920e4f70b2
--- /dev/null
+++
b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/beans/InfinispanRemoteKeyValueRepository.json
@@ -0,0 +1,16 @@
+{
+ "bean": {
+ "kind": "bean",
+ "name": "InfinispanRemoteKeyValueRepository",
+ "javaType":
"org.apache.camel.component.infinispan.remote.InfinispanRemoteKeyValueRepository",
+ "interfaceType": "org.apache.camel.spi.KeyValueRepository",
+ "title": "Infinispan Remote Key Value Repository",
+ "description": "A KeyValueRepository backed by remote Infinispan (HotRod
client).",
+ "deprecated": false,
+ "groupId": "org.apache.camel",
+ "artifactId": "camel-infinispan",
+ "version": "4.23.0-SNAPSHOT",
+ "properties": { "cacheName": { "index": 0, "kind": "property",
"displayName": "Cache Name", "required": true, "type": "string", "javaType":
"java.lang.String", "deprecated": false, "autowired": false, "secret": false,
"description": "Name of cache" }, "configuration": { "index": 1, "kind":
"property", "displayName": "Configuration", "required": false, "type":
"object", "javaType":
"org.apache.camel.component.infinispan.remote.InfinispanRemoteConfiguration",
"deprecated": false, "auto [...]
+ }
+}
+
diff --git
a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/beans/RedisKeyValueRepository.json
b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/beans/RedisKeyValueRepository.json
new file mode 100644
index 000000000000..35fad0e6c77f
--- /dev/null
+++
b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/beans/RedisKeyValueRepository.json
@@ -0,0 +1,16 @@
+{
+ "bean": {
+ "kind": "bean",
+ "name": "RedisKeyValueRepository",
+ "javaType": "org.apache.camel.component.redis.RedisKeyValueRepository",
+ "interfaceType": "org.apache.camel.spi.KeyValueRepository",
+ "title": "Redis Key Value Repository",
+ "description": "A KeyValueRepository backed by Redis (Redisson client).",
+ "deprecated": false,
+ "groupId": "org.apache.camel",
+ "artifactId": "camel-redis",
+ "version": "4.23.0-SNAPSHOT",
+ "properties": { "redisson": { "index": 0, "kind": "property",
"displayName": "Redisson", "label": "advanced", "required": false, "type":
"object", "javaType": "org.redisson.api.RedissonClient", "deprecated": false,
"autowired": false, "secret": false, "description": "To use an existing
Redisson client to connect to Redis server" }, "endpoint": { "index": 1,
"kind": "property", "displayName": "Endpoint", "required": true, "type":
"string", "javaType": "java.lang.String", "deprecated": [...]
+ }
+}
+
diff --git
a/components/camel-infinispan/camel-infinispan/src/generated/java/org/apache/camel/component/infinispan/remote/InfinispanRemoteKeyValueRepositoryConfigurer.java
b/components/camel-infinispan/camel-infinispan/src/generated/java/org/apache/camel/component/infinispan/remote/InfinispanRemoteKeyValueRepositoryConfigurer.java
new file mode 100644
index 000000000000..74a1b158fff4
--- /dev/null
+++
b/components/camel-infinispan/camel-infinispan/src/generated/java/org/apache/camel/component/infinispan/remote/InfinispanRemoteKeyValueRepositoryConfigurer.java
@@ -0,0 +1,60 @@
+/* Generated by camel build tools - do NOT edit this file! */
+package org.apache.camel.component.infinispan.remote;
+
+import javax.annotation.processing.Generated;
+import java.util.Map;
+
+import org.apache.camel.CamelContext;
+import org.apache.camel.spi.ExtendedPropertyConfigurerGetter;
+import org.apache.camel.spi.PropertyConfigurerGetter;
+import org.apache.camel.spi.ConfigurerStrategy;
+import org.apache.camel.spi.GeneratedPropertyConfigurer;
+import org.apache.camel.util.CaseInsensitiveMap;
+import
org.apache.camel.component.infinispan.remote.InfinispanRemoteKeyValueRepository;
+
+/**
+ * Generated by camel build tools - do NOT edit this file!
+ */
+@Generated("org.apache.camel.maven.packaging.GenerateConfigurerMojo")
+@SuppressWarnings("unchecked")
+public class InfinispanRemoteKeyValueRepositoryConfigurer extends
org.apache.camel.support.component.PropertyConfigurerSupport implements
GeneratedPropertyConfigurer, PropertyConfigurerGetter {
+
+ @Override
+ public boolean configure(CamelContext camelContext, Object obj, String
name, Object value, boolean ignoreCase) {
+
org.apache.camel.component.infinispan.remote.InfinispanRemoteKeyValueRepository
target =
(org.apache.camel.component.infinispan.remote.InfinispanRemoteKeyValueRepository)
obj;
+ switch (ignoreCase ? name.toLowerCase() : name) {
+ case "cachecontainer":
+ case "cacheContainer": target.setCacheContainer(property(camelContext,
org.infinispan.client.hotrod.RemoteCacheManager.class, value)); return true;
+ case "cachename":
+ case "cacheName": target.setCacheName(property(camelContext,
java.lang.String.class, value)); return true;
+ case "configuration": target.setConfiguration(property(camelContext,
org.apache.camel.component.infinispan.remote.InfinispanRemoteConfiguration.class,
value)); return true;
+ default: return false;
+ }
+ }
+
+ @Override
+ public Class<?> getOptionType(String name, boolean ignoreCase) {
+ switch (ignoreCase ? name.toLowerCase() : name) {
+ case "cachecontainer":
+ case "cacheContainer": return
org.infinispan.client.hotrod.RemoteCacheManager.class;
+ case "cachename":
+ case "cacheName": return java.lang.String.class;
+ case "configuration": return
org.apache.camel.component.infinispan.remote.InfinispanRemoteConfiguration.class;
+ default: return null;
+ }
+ }
+
+ @Override
+ public Object getOptionValue(Object obj, String name, boolean ignoreCase) {
+
org.apache.camel.component.infinispan.remote.InfinispanRemoteKeyValueRepository
target =
(org.apache.camel.component.infinispan.remote.InfinispanRemoteKeyValueRepository)
obj;
+ switch (ignoreCase ? name.toLowerCase() : name) {
+ case "cachecontainer":
+ case "cacheContainer": return target.getCacheContainer();
+ case "cachename":
+ case "cacheName": return target.getCacheName();
+ case "configuration": return target.getConfiguration();
+ default: return null;
+ }
+ }
+}
+
diff --git
a/components/camel-infinispan/camel-infinispan/src/generated/resources/META-INF/services/org/apache/camel/bean.properties
b/components/camel-infinispan/camel-infinispan/src/generated/resources/META-INF/services/org/apache/camel/bean.properties
index 4266e31d6ea1..69f469416651 100644
---
a/components/camel-infinispan/camel-infinispan/src/generated/resources/META-INF/services/org/apache/camel/bean.properties
+++
b/components/camel-infinispan/camel-infinispan/src/generated/resources/META-INF/services/org/apache/camel/bean.properties
@@ -1,5 +1,5 @@
# Generated by camel build tools - do NOT edit this file!
-bean=InfinispanRemoteAggregationRepository InfinispanRemoteClusterService
InfinispanRemoteIdempotentRepository
+bean=InfinispanRemoteAggregationRepository InfinispanRemoteClusterService
InfinispanRemoteIdempotentRepository InfinispanRemoteKeyValueRepository
groupId=org.apache.camel
artifactId=camel-infinispan
version=4.23.0-SNAPSHOT
diff --git
a/components/camel-infinispan/camel-infinispan/src/generated/resources/META-INF/services/org/apache/camel/bean/InfinispanRemoteKeyValueRepository.json
b/components/camel-infinispan/camel-infinispan/src/generated/resources/META-INF/services/org/apache/camel/bean/InfinispanRemoteKeyValueRepository.json
new file mode 100644
index 000000000000..7a920e4f70b2
--- /dev/null
+++
b/components/camel-infinispan/camel-infinispan/src/generated/resources/META-INF/services/org/apache/camel/bean/InfinispanRemoteKeyValueRepository.json
@@ -0,0 +1,16 @@
+{
+ "bean": {
+ "kind": "bean",
+ "name": "InfinispanRemoteKeyValueRepository",
+ "javaType":
"org.apache.camel.component.infinispan.remote.InfinispanRemoteKeyValueRepository",
+ "interfaceType": "org.apache.camel.spi.KeyValueRepository",
+ "title": "Infinispan Remote Key Value Repository",
+ "description": "A KeyValueRepository backed by remote Infinispan (HotRod
client).",
+ "deprecated": false,
+ "groupId": "org.apache.camel",
+ "artifactId": "camel-infinispan",
+ "version": "4.23.0-SNAPSHOT",
+ "properties": { "cacheName": { "index": 0, "kind": "property",
"displayName": "Cache Name", "required": true, "type": "string", "javaType":
"java.lang.String", "deprecated": false, "autowired": false, "secret": false,
"description": "Name of cache" }, "configuration": { "index": 1, "kind":
"property", "displayName": "Configuration", "required": false, "type":
"object", "javaType":
"org.apache.camel.component.infinispan.remote.InfinispanRemoteConfiguration",
"deprecated": false, "auto [...]
+ }
+}
+
diff --git
a/components/camel-infinispan/camel-infinispan/src/generated/resources/META-INF/services/org/apache/camel/configurer/org.apache.camel.component.infinispan.remote.InfinispanRemoteKeyValueRepository
b/components/camel-infinispan/camel-infinispan/src/generated/resources/META-INF/services/org/apache/camel/configurer/org.apache.camel.component.infinispan.remote.InfinispanRemoteKeyValueRepository
new file mode 100644
index 000000000000..f3d4b4df7de4
--- /dev/null
+++
b/components/camel-infinispan/camel-infinispan/src/generated/resources/META-INF/services/org/apache/camel/configurer/org.apache.camel.component.infinispan.remote.InfinispanRemoteKeyValueRepository
@@ -0,0 +1,2 @@
+# Generated by camel build tools - do NOT edit this file!
+class=org.apache.camel.component.infinispan.remote.InfinispanRemoteKeyValueRepositoryConfigurer
diff --git
a/components/camel-infinispan/camel-infinispan/src/main/java/org/apache/camel/component/infinispan/remote/InfinispanRemoteKeyValueRepository.java
b/components/camel-infinispan/camel-infinispan/src/main/java/org/apache/camel/component/infinispan/remote/InfinispanRemoteKeyValueRepository.java
new file mode 100644
index 000000000000..19dafea30149
--- /dev/null
+++
b/components/camel-infinispan/camel-infinispan/src/main/java/org/apache/camel/component/infinispan/remote/InfinispanRemoteKeyValueRepository.java
@@ -0,0 +1,266 @@
+/*
+ * 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.component.infinispan.remote;
+
+import java.time.Duration;
+import java.util.Objects;
+import java.util.Set;
+import java.util.concurrent.TimeUnit;
+import java.util.function.Supplier;
+
+import org.apache.camel.CamelContext;
+import org.apache.camel.CamelContextAware;
+import org.apache.camel.api.management.ManagedAttribute;
+import org.apache.camel.api.management.ManagedOperation;
+import org.apache.camel.api.management.ManagedResource;
+import org.apache.camel.spi.Configurer;
+import org.apache.camel.spi.KeyValueRepository;
+import org.apache.camel.spi.Metadata;
+import org.apache.camel.support.KeyValueRepositoryHelper;
+import org.apache.camel.support.service.ServiceHelper;
+import org.apache.camel.support.service.ServiceSupport;
+import org.apache.camel.util.ObjectHelper;
+import org.apache.camel.util.function.Suppliers;
+import org.infinispan.client.hotrod.Flag;
+import org.infinispan.client.hotrod.MetadataValue;
+import org.infinispan.client.hotrod.RemoteCache;
+import org.infinispan.client.hotrod.RemoteCacheManager;
+import org.jspecify.annotations.Nullable;
+
+import static
org.apache.camel.component.infinispan.remote.InfinispanRemoteUtil.getCacheWithFlags;
+
+/**
+ * A {@link KeyValueRepository} implementation backed by a remote Infinispan
server via the HotRod client protocol.
+ * <p/>
+ * TTL is mapped from {@link Duration} to Infinispan's native lifespan via
+ * {@code BasicCache.put(key, value, lifespan, TimeUnit.MILLISECONDS)}. The
cache is configured with
+ * {@link Flag#FORCE_RETURN_VALUE} to ensure that {@code put} and {@code
remove} operations return previous values as
+ * required by the {@link KeyValueRepository} contract.
+ * <p/>
+ * Values are serialized through {@link KeyValueRepositoryHelper} (plain Java
serialization) and stored as raw
+ * {@code byte[]} in the cache. This ensures a consistent serialization format
across all persistent
+ * {@link KeyValueRepository} implementations, avoids coupling to Infinispan's
marshaller configuration, and eliminates
+ * the need for ProtoStream schema registration for complex value types.
+ *
+ * @since 4.23
+ */
+@Metadata(label = "bean",
+ description = "A KeyValueRepository backed by remote Infinispan
(HotRod client).",
+ annotations = {
"interfaceName=org.apache.camel.spi.KeyValueRepository" })
+@Configurer(metadataOnly = true)
+@ManagedResource(description = "Infinispan Remote based key-value repository")
+public class InfinispanRemoteKeyValueRepository extends ServiceSupport
implements KeyValueRepository, CamelContextAware {
+
+ private CamelContext camelContext;
+ private Supplier<RemoteCache<String, byte[]>> cache;
+ private InfinispanRemoteManager manager;
+
+ @Metadata(description = "Name of cache", required = true)
+ private String cacheName;
+ @Metadata(description = "Configuration for remote Infinispan")
+ private InfinispanRemoteConfiguration configuration;
+
+ public InfinispanRemoteKeyValueRepository() {
+ }
+
+ /**
+ * Creates a new Infinispan remote key-value repository for the given
cache name.
+ *
+ * @param cacheName the name of the Infinispan cache to use
+ */
+ public InfinispanRemoteKeyValueRepository(String cacheName) {
+ this.cacheName = cacheName;
+ }
+
+ @Override
+ @ManagedOperation(description = "Get value by key")
+ public @Nullable Object get(String key) {
+ byte[] bytes = cache.get().get(key);
+ return bytes != null ? KeyValueRepositoryHelper.deserialize(bytes) :
null;
+ }
+
+ @Override
+ @ManagedOperation(description = "Put a key-value pair with optional TTL")
+ public @Nullable Object put(String key, Object value, Duration ttl) {
+ byte[] serialized = KeyValueRepositoryHelper.serialize(value);
+ byte[] previous;
+ if (hasPositiveTtl(ttl)) {
+ previous = cache.get().put(key, serialized, ttl.toMillis(),
TimeUnit.MILLISECONDS);
+ } else {
+ previous = cache.get().put(key, serialized);
+ }
+ return previous != null ?
KeyValueRepositoryHelper.deserialize(previous) : null;
+ }
+
+ @Override
+ @ManagedOperation(description = "Delete a key")
+ public @Nullable Object delete(String key) {
+ byte[] bytes = cache.get().remove(key);
+ return bytes != null ? KeyValueRepositoryHelper.deserialize(bytes) :
null;
+ }
+
+ @Override
+ @ManagedOperation(description = "Check if key exists")
+ public boolean contains(String key) {
+ return cache.get().containsKey(key);
+ }
+
+ @Override
+ public Set<String> keys() {
+ return cache.get().keySet();
+ }
+
+ @Override
+ @ManagedOperation(description = "Clear all entries")
+ public void clear() {
+ cache.get().clear();
+ }
+
+ @Override
+ public @Nullable Object putIfAbsent(String key, Object value, Duration
ttl) {
+ byte[] serialized = KeyValueRepositoryHelper.serialize(value);
+ byte[] existing;
+ if (hasPositiveTtl(ttl)) {
+ existing = cache.get().putIfAbsent(key, serialized,
ttl.toMillis(), TimeUnit.MILLISECONDS);
+ } else {
+ existing = cache.get().putIfAbsent(key, serialized);
+ }
+ return existing != null ?
KeyValueRepositoryHelper.deserialize(existing) : null;
+ }
+
+ /**
+ * Atomically replaces the value for the given key using Infinispan's
version-based optimistic locking.
+ * <p/>
+ * Because values are stored as serialized {@code byte[]}, Infinispan's
value-based {@code replace(K, V, V)} cannot
+ * compare entries reliably through the marshaller. Instead, this method
uses {@code getWithMetadata} to read the
+ * current value and its version, compares the deserialized value at the
Java object level, and then calls
+ * {@code replaceWithVersion} to perform an atomic swap guarded by the
entry version.
+ */
+ @Override
+ public boolean replace(String key, Object expectedOldValue, Object
newValue, Duration ttl) {
+ MetadataValue<byte[]> metadata = cache.get().getWithMetadata(key);
+ if (metadata == null) {
+ return false;
+ }
+ Object currentObj =
KeyValueRepositoryHelper.deserialize(metadata.getValue());
+ if (!Objects.equals(currentObj, expectedOldValue)) {
+ return false;
+ }
+ byte[] newBytes = KeyValueRepositoryHelper.serialize(newValue);
+ if (hasPositiveTtl(ttl)) {
+ return cache.get().replaceWithVersion(key, newBytes,
metadata.getVersion(),
+ ttl.toMillis(), TimeUnit.MILLISECONDS, -1,
TimeUnit.MILLISECONDS);
+ }
+ return cache.get().replaceWithVersion(key, newBytes,
metadata.getVersion());
+ }
+
+ /**
+ * Atomically removes the entry for the given key using Infinispan's
version-based optimistic locking.
+ * <p/>
+ * Similar to {@link #replace}, this uses {@code getWithMetadata} and
{@code removeWithVersion} to avoid relying on
+ * marshaller-level value comparison for serialized {@code byte[]} entries.
+ */
+ @Override
+ public boolean delete(String key, Object expectedValue) {
+ MetadataValue<byte[]> metadata = cache.get().getWithMetadata(key);
+ if (metadata == null) {
+ return false;
+ }
+ Object currentObj =
KeyValueRepositoryHelper.deserialize(metadata.getValue());
+ if (!Objects.equals(currentObj, expectedValue)) {
+ return false;
+ }
+ return cache.get().removeWithVersion(key, metadata.getVersion());
+ }
+
+ @Override
+ @ManagedAttribute(description = "The number of entries in the repository")
+ public int size() {
+ return cache.get().size();
+ }
+
+ // ---- Configuration accessors ----
+
+ @ManagedAttribute(description = "The cache name")
+ public String getCacheName() {
+ return cacheName;
+ }
+
+ public void setCacheName(String cacheName) {
+ this.cacheName = cacheName;
+ }
+
+ public InfinispanRemoteConfiguration getConfiguration() {
+ return configuration;
+ }
+
+ public void setConfiguration(InfinispanRemoteConfiguration configuration) {
+ this.configuration = configuration;
+ }
+
+ public RemoteCacheManager getCacheContainer() {
+ return configuration != null ? configuration.getCacheContainer() :
null;
+ }
+
+ public void setCacheContainer(RemoteCacheManager cacheContainer) {
+ if (this.configuration == null) {
+ this.configuration = new InfinispanRemoteConfiguration();
+ }
+ this.configuration.setCacheContainer(cacheContainer);
+ }
+
+ public InfinispanRemoteManager getManager() {
+ return manager;
+ }
+
+ @Override
+ public CamelContext getCamelContext() {
+ return camelContext;
+ }
+
+ @Override
+ public void setCamelContext(CamelContext camelContext) {
+ this.camelContext = camelContext;
+ }
+
+ // ---- Lifecycle ----
+
+ @Override
+ protected void doStart() throws Exception {
+ ObjectHelper.notNull(cacheName, "cacheName", this);
+
+ if (this.configuration == null) {
+ this.configuration = new InfinispanRemoteConfiguration();
+ }
+
+ this.manager = new InfinispanRemoteManager(camelContext,
configuration);
+ this.cache = Suppliers.memorize(
+ () -> getCacheWithFlags(manager, cacheName,
Flag.FORCE_RETURN_VALUE));
+ ServiceHelper.startService(manager);
+ }
+
+ @Override
+ protected void doStop() throws Exception {
+ ServiceHelper.stopAndShutdownService(manager);
+ }
+
+ // ---- Internal ----
+
+ private static boolean hasPositiveTtl(Duration ttl) {
+ return ttl != null && !ttl.isZero() && !ttl.isNegative();
+ }
+}
diff --git
a/components/camel-infinispan/camel-infinispan/src/test/java/org/apache/camel/component/infinispan/remote/InfinispanRemoteKeyValueRepositoryIT.java
b/components/camel-infinispan/camel-infinispan/src/test/java/org/apache/camel/component/infinispan/remote/InfinispanRemoteKeyValueRepositoryIT.java
new file mode 100644
index 000000000000..64cafe5a80f3
--- /dev/null
+++
b/components/camel-infinispan/camel-infinispan/src/test/java/org/apache/camel/component/infinispan/remote/InfinispanRemoteKeyValueRepositoryIT.java
@@ -0,0 +1,308 @@
+/*
+ * 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.component.infinispan.remote;
+
+import java.time.Duration;
+import java.util.Properties;
+import java.util.Set;
+import java.util.concurrent.TimeUnit;
+
+import org.apache.camel.CamelContext;
+import org.apache.camel.impl.DefaultCamelContext;
+import org.apache.camel.test.infra.infinispan.common.InfinispanProperties;
+import org.apache.camel.test.infra.infinispan.services.InfinispanService;
+import
org.apache.camel.test.infra.infinispan.services.InfinispanServiceFactory;
+import org.infinispan.client.hotrod.RemoteCacheManager;
+import org.infinispan.client.hotrod.configuration.ConfigurationBuilder;
+import org.infinispan.configuration.cache.CacheMode;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.RegisterExtension;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+import static org.awaitility.Awaitility.await;
+
+class InfinispanRemoteKeyValueRepositoryIT {
+
+ private static final String CACHE_NAME = "kvr-test";
+
+ @RegisterExtension
+ static InfinispanService service =
InfinispanServiceFactory.createSingletonInfinispanService();
+
+ private CamelContext camelContext;
+ private RemoteCacheManager cacheManager;
+ private InfinispanRemoteKeyValueRepository repository;
+
+ @BeforeEach
+ void setUp() {
+ camelContext = new DefaultCamelContext();
+
+ cacheManager = new RemoteCacheManager(getConfiguration().build());
+
+ // Create the cache on the server
+ cacheManager.administration()
+ .getOrCreateCache(
+ CACHE_NAME,
+ new
org.infinispan.configuration.cache.ConfigurationBuilder()
+ .clustering()
+ .cacheMode(CacheMode.DIST_SYNC)
+ .build());
+
+ InfinispanRemoteConfiguration config = new
InfinispanRemoteConfiguration();
+ config.setCacheContainer(cacheManager);
+
+ repository = new InfinispanRemoteKeyValueRepository(CACHE_NAME);
+ repository.setCamelContext(camelContext);
+ repository.setConfiguration(config);
+ repository.start();
+ }
+
+ @AfterEach
+ void tearDown() {
+ if (repository != null) {
+ try {
+ repository.clear();
+ } catch (Exception e) {
+ // ignore cleanup errors
+ }
+ repository.stop();
+ }
+ if (cacheManager != null) {
+ cacheManager.stop();
+ }
+ if (camelContext != null) {
+ try {
+ camelContext.close();
+ } catch (Exception e) {
+ // ignore cleanup errors
+ }
+ }
+ }
+
+ @Test
+ void testPutAndGet() {
+ Object previous = repository.put("key1", "value1", null);
+ assertThat(previous).isNull();
+
+ Object result = repository.get("key1");
+ assertThat(result).isEqualTo("value1");
+ }
+
+ @Test
+ void testPutReturnsOldValue() {
+ repository.put("key1", "first", null);
+ Object previous = repository.put("key1", "second", null);
+ assertThat(previous).isEqualTo("first");
+
+ Object result = repository.get("key1");
+ assertThat(result).isEqualTo("second");
+ }
+
+ @Test
+ void testGetNonExistent() {
+ Object result = repository.get("nonexistent");
+ assertThat(result).isNull();
+ }
+
+ @Test
+ void testDelete() {
+ repository.put("key1", "value1", null);
+ Object deleted = repository.delete("key1");
+ assertThat(deleted).isEqualTo("value1");
+ assertThat(repository.get("key1")).isNull();
+ }
+
+ @Test
+ void testDeleteNonExistent() {
+ Object deleted = repository.delete("nonexistent");
+ assertThat(deleted).isNull();
+ }
+
+ @Test
+ void testContains() {
+ repository.put("key1", "value1", null);
+ assertThat(repository.contains("key1")).isTrue();
+ assertThat(repository.contains("nonexistent")).isFalse();
+ }
+
+ @Test
+ void testKeys() {
+ repository.put("a", "1", null);
+ repository.put("b", "2", null);
+ repository.put("c", "3", null);
+
+ Set<String> keys = repository.keys();
+ assertThat(keys).containsExactlyInAnyOrder("a", "b", "c");
+ }
+
+ @Test
+ void testKeysEmpty() {
+ Set<String> keys = repository.keys();
+ assertThat(keys).isEmpty();
+ }
+
+ @Test
+ void testClear() {
+ repository.put("a", "1", null);
+ repository.put("b", "2", null);
+
+ repository.clear();
+
+ assertThat(repository.keys()).isEmpty();
+ assertThat(repository.contains("a")).isFalse();
+ }
+
+ @Test
+ void testPutIfAbsentNewKey() {
+ Object existing = repository.putIfAbsent("key1", "value1", null);
+ assertThat(existing).isNull();
+ assertThat(repository.get("key1")).isEqualTo("value1");
+ }
+
+ @Test
+ void testPutIfAbsentExistingKey() {
+ repository.put("key1", "original", null);
+ Object existing = repository.putIfAbsent("key1", "replacement", null);
+ assertThat(existing).isEqualTo("original");
+ assertThat(repository.get("key1")).isEqualTo("original");
+ }
+
+ @Test
+ void testSize() {
+ assertThat(repository.size()).isZero();
+
+ repository.put("a", "1", null);
+ repository.put("b", "2", null);
+ assertThat(repository.size()).isEqualTo(2);
+
+ repository.delete("a");
+ assertThat(repository.size()).isEqualTo(1);
+ }
+
+ @Test
+ void testTtlExpiry() {
+ // Store with a short TTL (Infinispan lifespan)
+ repository.put("expiring", "value", Duration.ofSeconds(1));
+ assertThat(repository.contains("expiring")).isTrue();
+
+ // Wait for it to expire
+ await().atMost(10, TimeUnit.SECONDS)
+ .untilAsserted(() ->
assertThat(repository.contains("expiring")).isFalse());
+ }
+
+ @Test
+ void testTtlPutIfAbsentExpiry() {
+ repository.putIfAbsent("expiring", "value", Duration.ofSeconds(1));
+ assertThat(repository.contains("expiring")).isTrue();
+
+ await().atMost(10, TimeUnit.SECONDS)
+ .untilAsserted(() ->
assertThat(repository.contains("expiring")).isFalse());
+ }
+
+ @Test
+ void testReplaceMatchingValue() {
+ repository.put("key1", "value1", null);
+ boolean replaced = repository.replace("key1", "value1", "value2",
null);
+ assertThat(replaced).isTrue();
+ assertThat(repository.get("key1")).isEqualTo("value2");
+ }
+
+ @Test
+ void testReplaceNonMatchingValue() {
+ repository.put("key1", "value1", null);
+ boolean replaced = repository.replace("key1", "wrong", "value2", null);
+ assertThat(replaced).isFalse();
+ assertThat(repository.get("key1")).isEqualTo("value1");
+ }
+
+ @Test
+ void testReplaceMissingKey() {
+ boolean replaced = repository.replace("nonexistent", "value1",
"value2", null);
+ assertThat(replaced).isFalse();
+ }
+
+ @Test
+ void testReplaceWithTtl() {
+ repository.put("key1", "value1", null);
+ boolean replaced = repository.replace("key1", "value1", "value2",
Duration.ofSeconds(1));
+ assertThat(replaced).isTrue();
+ assertThat(repository.get("key1")).isEqualTo("value2");
+
+ await().atMost(10, TimeUnit.SECONDS)
+ .untilAsserted(() ->
assertThat(repository.get("key1")).isNull());
+ }
+
+ @Test
+ void testDeleteWithMatchingValue() {
+ repository.put("key1", "value1", null);
+ boolean deleted = repository.delete("key1", "value1");
+ assertThat(deleted).isTrue();
+ assertThat(repository.get("key1")).isNull();
+ }
+
+ @Test
+ void testDeleteWithNonMatchingValue() {
+ repository.put("key1", "value1", null);
+ boolean deleted = repository.delete("key1", "wrong");
+ assertThat(deleted).isFalse();
+ assertThat(repository.get("key1")).isEqualTo("value1");
+ }
+
+ @Test
+ void testDeleteWithMissingKey() {
+ boolean deleted = repository.delete("nonexistent", "value1");
+ assertThat(deleted).isFalse();
+ }
+
+ @Test
+ void testCacheNameRequired() {
+ InfinispanRemoteKeyValueRepository repo = new
InfinispanRemoteKeyValueRepository();
+ repo.setCamelContext(camelContext);
+
assertThatThrownBy(repo::start).isInstanceOf(IllegalArgumentException.class);
+ }
+
+ private ConfigurationBuilder getConfiguration() {
+ ConfigurationBuilder clientBuilder = new ConfigurationBuilder();
+
+ clientBuilder.forceReturnValues(true);
+
+ clientBuilder
+ .addServer()
+ .host(service.host())
+ .port(service.port());
+
+ clientBuilder
+ .socketTimeout(15000)
+ .connectionTimeout(15000)
+ .security()
+ .authentication()
+ .username(service.username())
+ .password(service.password())
+ .serverName("infinispan")
+ .saslMechanism("SCRAM-SHA-512")
+ .realm("default");
+
+ if
(!Boolean.getBoolean(InfinispanProperties.INFINISPAN_CONTAINER_NETWORK_MODE_HOST))
{
+ Properties properties = new Properties();
+ properties.put("infinispan.client.hotrod.client_intelligence",
"BASIC");
+ clientBuilder.withProperties(properties);
+ }
+ return clientBuilder;
+ }
+}
diff --git a/components/camel-redis/pom.xml b/components/camel-redis/pom.xml
index fa1ea1977a31..0d01b321f306 100644
--- a/components/camel-redis/pom.xml
+++ b/components/camel-redis/pom.xml
@@ -56,6 +56,17 @@
<artifactId>camel-test-junit6</artifactId>
<scope>test</scope>
</dependency>
+ <dependency>
+ <groupId>org.assertj</groupId>
+ <artifactId>assertj-core</artifactId>
+ <scope>test</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.awaitility</groupId>
+ <artifactId>awaitility</artifactId>
+ <version>${awaitility-version}</version>
+ <scope>test</scope>
+ </dependency>
<!-- test infra -->
<dependency>
diff --git
a/components/camel-redis/src/generated/java/org/apache/camel/component/redis/RedisKeyValueRepositoryConfigurer.java
b/components/camel-redis/src/generated/java/org/apache/camel/component/redis/RedisKeyValueRepositoryConfigurer.java
new file mode 100644
index 000000000000..63e66174255e
--- /dev/null
+++
b/components/camel-redis/src/generated/java/org/apache/camel/component/redis/RedisKeyValueRepositoryConfigurer.java
@@ -0,0 +1,57 @@
+/* Generated by camel build tools - do NOT edit this file! */
+package org.apache.camel.component.redis;
+
+import javax.annotation.processing.Generated;
+import java.util.Map;
+
+import org.apache.camel.CamelContext;
+import org.apache.camel.spi.ExtendedPropertyConfigurerGetter;
+import org.apache.camel.spi.PropertyConfigurerGetter;
+import org.apache.camel.spi.ConfigurerStrategy;
+import org.apache.camel.spi.GeneratedPropertyConfigurer;
+import org.apache.camel.util.CaseInsensitiveMap;
+import org.apache.camel.component.redis.RedisKeyValueRepository;
+
+/**
+ * Generated by camel build tools - do NOT edit this file!
+ */
+@Generated("org.apache.camel.maven.packaging.GenerateConfigurerMojo")
+@SuppressWarnings("unchecked")
+public class RedisKeyValueRepositoryConfigurer extends
org.apache.camel.support.component.PropertyConfigurerSupport implements
GeneratedPropertyConfigurer, PropertyConfigurerGetter {
+
+ @Override
+ public boolean configure(CamelContext camelContext, Object obj, String
name, Object value, boolean ignoreCase) {
+ org.apache.camel.component.redis.RedisKeyValueRepository target =
(org.apache.camel.component.redis.RedisKeyValueRepository) obj;
+ switch (ignoreCase ? name.toLowerCase() : name) {
+ case "endpoint": target.setEndpoint(property(camelContext,
java.lang.String.class, value)); return true;
+ case "keyprefix":
+ case "keyPrefix": target.setKeyPrefix(property(camelContext,
java.lang.String.class, value)); return true;
+ case "redisson": target.setRedisson(property(camelContext,
org.redisson.api.RedissonClient.class, value)); return true;
+ default: return false;
+ }
+ }
+
+ @Override
+ public Class<?> getOptionType(String name, boolean ignoreCase) {
+ switch (ignoreCase ? name.toLowerCase() : name) {
+ case "endpoint": return java.lang.String.class;
+ case "keyprefix":
+ case "keyPrefix": return java.lang.String.class;
+ case "redisson": return org.redisson.api.RedissonClient.class;
+ default: return null;
+ }
+ }
+
+ @Override
+ public Object getOptionValue(Object obj, String name, boolean ignoreCase) {
+ org.apache.camel.component.redis.RedisKeyValueRepository target =
(org.apache.camel.component.redis.RedisKeyValueRepository) obj;
+ switch (ignoreCase ? name.toLowerCase() : name) {
+ case "endpoint": return target.getEndpoint();
+ case "keyprefix":
+ case "keyPrefix": return target.getKeyPrefix();
+ case "redisson": return target.getRedisson();
+ default: return null;
+ }
+ }
+}
+
diff --git
a/components/camel-redis/src/generated/resources/META-INF/services/org/apache/camel/bean.properties
b/components/camel-redis/src/generated/resources/META-INF/services/org/apache/camel/bean.properties
index 6291ead2f7c0..55e6cfc67360 100644
---
a/components/camel-redis/src/generated/resources/META-INF/services/org/apache/camel/bean.properties
+++
b/components/camel-redis/src/generated/resources/META-INF/services/org/apache/camel/bean.properties
@@ -1,5 +1,5 @@
# Generated by camel build tools - do NOT edit this file!
-bean=RedisAggregationRepository
+bean=RedisAggregationRepository RedisKeyValueRepository
groupId=org.apache.camel
artifactId=camel-redis
version=4.23.0-SNAPSHOT
diff --git
a/components/camel-redis/src/generated/resources/META-INF/services/org/apache/camel/bean/RedisKeyValueRepository.json
b/components/camel-redis/src/generated/resources/META-INF/services/org/apache/camel/bean/RedisKeyValueRepository.json
new file mode 100644
index 000000000000..35fad0e6c77f
--- /dev/null
+++
b/components/camel-redis/src/generated/resources/META-INF/services/org/apache/camel/bean/RedisKeyValueRepository.json
@@ -0,0 +1,16 @@
+{
+ "bean": {
+ "kind": "bean",
+ "name": "RedisKeyValueRepository",
+ "javaType": "org.apache.camel.component.redis.RedisKeyValueRepository",
+ "interfaceType": "org.apache.camel.spi.KeyValueRepository",
+ "title": "Redis Key Value Repository",
+ "description": "A KeyValueRepository backed by Redis (Redisson client).",
+ "deprecated": false,
+ "groupId": "org.apache.camel",
+ "artifactId": "camel-redis",
+ "version": "4.23.0-SNAPSHOT",
+ "properties": { "redisson": { "index": 0, "kind": "property",
"displayName": "Redisson", "label": "advanced", "required": false, "type":
"object", "javaType": "org.redisson.api.RedissonClient", "deprecated": false,
"autowired": false, "secret": false, "description": "To use an existing
Redisson client to connect to Redis server" }, "endpoint": { "index": 1,
"kind": "property", "displayName": "Endpoint", "required": true, "type":
"string", "javaType": "java.lang.String", "deprecated": [...]
+ }
+}
+
diff --git
a/components/camel-redis/src/generated/resources/META-INF/services/org/apache/camel/configurer/org.apache.camel.component.redis.RedisKeyValueRepository
b/components/camel-redis/src/generated/resources/META-INF/services/org/apache/camel/configurer/org.apache.camel.component.redis.RedisKeyValueRepository
new file mode 100644
index 000000000000..acfa00a671c2
--- /dev/null
+++
b/components/camel-redis/src/generated/resources/META-INF/services/org/apache/camel/configurer/org.apache.camel.component.redis.RedisKeyValueRepository
@@ -0,0 +1,2 @@
+# Generated by camel build tools - do NOT edit this file!
+class=org.apache.camel.component.redis.RedisKeyValueRepositoryConfigurer
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
new file mode 100644
index 000000000000..3bedccad69fb
--- /dev/null
+++
b/components/camel-redis/src/main/java/org/apache/camel/component/redis/RedisKeyValueRepository.java
@@ -0,0 +1,262 @@
+/*
+ * 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.component.redis;
+
+import java.time.Duration;
+import java.util.Set;
+import java.util.stream.Collectors;
+
+import org.apache.camel.api.management.ManagedAttribute;
+import org.apache.camel.api.management.ManagedOperation;
+import org.apache.camel.api.management.ManagedResource;
+import org.apache.camel.spi.Configurer;
+import org.apache.camel.spi.KeyValueRepository;
+import org.apache.camel.spi.Metadata;
+import org.apache.camel.support.KeyValueRepositoryHelper;
+import org.apache.camel.support.service.ServiceSupport;
+import org.apache.camel.util.StringHelper;
+import org.jspecify.annotations.Nullable;
+import org.redisson.Redisson;
+import org.redisson.api.RBucket;
+import org.redisson.api.RKeys;
+import org.redisson.api.RedissonClient;
+import org.redisson.api.options.KeysScanOptions;
+import org.redisson.client.codec.ByteArrayCodec;
+import org.redisson.config.Config;
+
+/**
+ * A {@link KeyValueRepository} implementation backed by Redis using the
Redisson client.
+ * <p/>
+ * Keys are namespaced under a configurable prefix ({@link #keyPrefix}) to
avoid collisions with other data in the same
+ * Redis instance. Values are serialized through {@link
KeyValueRepositoryHelper} (plain Java serialization) and stored
+ * as raw {@code byte[]} using Redisson's {@link ByteArrayCodec}. This ensures
a consistent serialization format across
+ * all persistent {@link KeyValueRepository} implementations and avoids
coupling to Redisson's built-in codec.
+ * <p/>
+ * TTL is mapped from milliseconds to Redis native key expiry via
+ * {@code RBucket.set(value, ttl, TimeUnit.MILLISECONDS)}. Atomic {@link
#putIfAbsent} is supported via
+ * {@code RBucket.setIfAbsent}.
+ *
+ * @since 4.23
+ */
+@Metadata(label = "bean",
+ description = "A KeyValueRepository backed by Redis (Redisson
client).",
+ annotations = {
"interfaceName=org.apache.camel.spi.KeyValueRepository" })
+@Configurer(metadataOnly = true)
+@ManagedResource(description = "Redis based key-value repository")
+public class RedisKeyValueRepository extends ServiceSupport implements
KeyValueRepository {
+
+ private boolean shutdownRedisson;
+
+ @Metadata(label = "advanced", description = "To use an existing Redisson
client to connect to Redis server")
+ private RedissonClient redisson;
+ @Metadata(description = "URL to remote Redis server (host:port)", required
= true)
+ private String endpoint;
+ @Metadata(description = "Key prefix used to namespace entries in Redis",
defaultValue = "camel-kvr:")
+ private String keyPrefix = "camel-kvr:";
+
+ public RedisKeyValueRepository() {
+ }
+
+ /**
+ * Creates a new Redis key-value repository connecting to the given
endpoint.
+ *
+ * @param endpoint the Redis server address in {@code host:port} format
+ */
+ public RedisKeyValueRepository(String endpoint) {
+ this.endpoint = endpoint;
+ }
+
+ /**
+ * Creates a new Redis key-value repository connecting to the given
endpoint with a custom key prefix.
+ *
+ * @param endpoint the Redis server address in {@code host:port} format
+ * @param keyPrefix the prefix to prepend to all keys stored in Redis
+ */
+ public RedisKeyValueRepository(String endpoint, String keyPrefix) {
+ this.endpoint = endpoint;
+ this.keyPrefix = keyPrefix;
+ }
+
+ @Override
+ @ManagedOperation(description = "Get value by key")
+ public @Nullable Object get(String key) {
+ RBucket<byte[]> bucket = redisson.getBucket(toRedisKey(key),
ByteArrayCodec.INSTANCE);
+ byte[] bytes = bucket.get();
+ return bytes != null ? KeyValueRepositoryHelper.deserialize(bytes) :
null;
+ }
+
+ @Override
+ @ManagedOperation(description = "Put a key-value pair with optional TTL")
+ 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();
+ if (hasPositiveTtl(ttl)) {
+ bucket.set(serialized, ttl);
+ } else {
+ bucket.set(serialized);
+ }
+ return previous != null ?
KeyValueRepositoryHelper.deserialize(previous) : null;
+ }
+
+ @Override
+ @ManagedOperation(description = "Delete a key")
+ public @Nullable Object delete(String key) {
+ RBucket<byte[]> bucket = redisson.getBucket(toRedisKey(key),
ByteArrayCodec.INSTANCE);
+ byte[] bytes = bucket.getAndDelete();
+ return bytes != null ? KeyValueRepositoryHelper.deserialize(bytes) :
null;
+ }
+
+ @Override
+ @ManagedOperation(description = "Check if key exists")
+ public boolean contains(String key) {
+ RBucket<byte[]> bucket = redisson.getBucket(toRedisKey(key),
ByteArrayCodec.INSTANCE);
+ return bucket.isExists();
+ }
+
+ @Override
+ public Set<String> keys() {
+ RKeys rKeys = redisson.getKeys();
+ int prefixLen = keyPrefix.length();
+ return
rKeys.getKeysStream(KeysScanOptions.defaults().pattern(toRedisKey("*")))
+ .map(k -> k.substring(prefixLen))
+ .collect(Collectors.toUnmodifiableSet());
+ }
+
+ @Override
+ @ManagedOperation(description = "Clear all entries")
+ public void clear() {
+ RKeys rKeys = redisson.getKeys();
+ String pattern = toRedisKey("*");
+ rKeys.deleteByPattern(pattern);
+ }
+
+ @Override
+ public @Nullable Object putIfAbsent(String key, Object value, Duration
ttl) {
+ RBucket<byte[]> bucket = redisson.getBucket(toRedisKey(key),
ByteArrayCodec.INSTANCE);
+ byte[] serialized = KeyValueRepositoryHelper.serialize(value);
+ boolean wasSet;
+ if (hasPositiveTtl(ttl)) {
+ wasSet = bucket.setIfAbsent(serialized, ttl);
+ } else {
+ wasSet = bucket.setIfAbsent(serialized);
+ }
+ if (wasSet) {
+ return null;
+ }
+ // Key already existed; return the current value
+ byte[] existing = bucket.get();
+ return existing != null ?
KeyValueRepositoryHelper.deserialize(existing) : null;
+ }
+
+ /**
+ * Atomically replaces the value for a key if the current value matches
the expected one.
+ * <p/>
+ * <b>Note:</b> Redisson does not expose an atomic CAS-and-set-TTL
primitive. When a TTL is specified, the
+ * replacement is performed in two steps: {@code compareAndSet} followed
by {@code expire}. There is a brief window
+ * between the two calls where the new value exists without its TTL
applied. A crash in that window would leave the
+ * entry without expiry. For callers requiring stricter consistency
guarantees (e.g. idempotent repositories), be
+ * aware of this non-atomic TTL application.
+ */
+ @Override
+ public boolean replace(String key, Object expectedOldValue, Object
newValue, Duration ttl) {
+ RBucket<byte[]> bucket = redisson.getBucket(toRedisKey(key),
ByteArrayCodec.INSTANCE);
+ byte[] expectedBytes =
KeyValueRepositoryHelper.serialize(expectedOldValue);
+ byte[] newBytes = KeyValueRepositoryHelper.serialize(newValue);
+ boolean swapped = bucket.compareAndSet(expectedBytes, newBytes);
+ if (swapped && hasPositiveTtl(ttl)) {
+ bucket.expire(ttl);
+ }
+ return swapped;
+ }
+
+ @Override
+ @ManagedAttribute(description = "The number of entries in the repository")
+ public int size() {
+ RKeys rKeys = redisson.getKeys();
+ String[] matchedKeys =
rKeys.getKeysStream(KeysScanOptions.defaults().pattern(toRedisKey("*")))
+ .toArray(String[]::new);
+ if (matchedKeys.length == 0) {
+ return 0;
+ }
+ return (int) rKeys.countExists(matchedKeys);
+ }
+
+ // ---- Configuration accessors ----
+
+ public String getEndpoint() {
+ return endpoint;
+ }
+
+ public void setEndpoint(String endpoint) {
+ this.endpoint = endpoint;
+ }
+
+ @ManagedAttribute(description = "Key prefix used for namespacing")
+ public String getKeyPrefix() {
+ return keyPrefix;
+ }
+
+ public void setKeyPrefix(String keyPrefix) {
+ this.keyPrefix = keyPrefix;
+ }
+
+ public RedissonClient getRedisson() {
+ return redisson;
+ }
+
+ public void setRedisson(RedissonClient redisson) {
+ this.redisson = redisson;
+ }
+
+ // ---- Lifecycle ----
+
+ @Override
+ protected void doInit() throws Exception {
+ if (redisson == null) {
+ StringHelper.notEmpty(endpoint, "endpoint");
+ }
+ }
+
+ @Override
+ protected void doStart() throws Exception {
+ if (redisson == null) {
+ Config config = new Config();
+ config.useSingleServer().setAddress(String.format("redis://%s",
endpoint));
+ redisson = Redisson.create(config);
+ shutdownRedisson = true;
+ }
+ }
+
+ @Override
+ protected void doStop() throws Exception {
+ if (redisson != null && shutdownRedisson) {
+ redisson.shutdown();
+ redisson = null;
+ }
+ }
+
+ // ---- Internal ----
+
+ private static boolean hasPositiveTtl(Duration ttl) {
+ return ttl != null && !ttl.isZero() && !ttl.isNegative();
+ }
+
+ private String toRedisKey(String key) {
+ return keyPrefix + key;
+ }
+}
diff --git
a/components/camel-redis/src/test/java/org/apache/camel/component/redis/integration/RedisKeyValueRepositoryIT.java
b/components/camel-redis/src/test/java/org/apache/camel/component/redis/integration/RedisKeyValueRepositoryIT.java
new file mode 100644
index 000000000000..d0fd1a8f6f24
--- /dev/null
+++
b/components/camel-redis/src/test/java/org/apache/camel/component/redis/integration/RedisKeyValueRepositoryIT.java
@@ -0,0 +1,299 @@
+/*
+ * 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.component.redis.integration;
+
+import java.time.Duration;
+import java.util.Set;
+import java.util.concurrent.TimeUnit;
+
+import org.apache.camel.component.redis.RedisKeyValueRepository;
+import org.apache.camel.test.infra.redis.services.RedisService;
+import org.apache.camel.test.infra.redis.services.RedisServiceFactory;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.RegisterExtension;
+import org.redisson.Redisson;
+import org.redisson.api.RedissonClient;
+import org.redisson.config.Config;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+import static org.awaitility.Awaitility.await;
+
+class RedisKeyValueRepositoryIT {
+
+ @RegisterExtension
+ static RedisService service = RedisServiceFactory.createSingletonService();
+
+ private RedisKeyValueRepository repository;
+
+ @BeforeEach
+ void setUp() {
+ repository = new RedisKeyValueRepository(service.getServiceAddress(),
"test-kvr:");
+ repository.start();
+ }
+
+ @AfterEach
+ void tearDown() {
+ if (repository != null) {
+ repository.clear();
+ repository.stop();
+ }
+ }
+
+ @Test
+ void testPutAndGet() {
+ Object previous = repository.put("key1", "value1", null);
+ assertThat(previous).isNull();
+
+ Object result = repository.get("key1");
+ assertThat(result).isEqualTo("value1");
+ }
+
+ @Test
+ void testPutReturnsOldValue() {
+ repository.put("key1", "first", null);
+ Object previous = repository.put("key1", "second", null);
+ assertThat(previous).isEqualTo("first");
+
+ Object result = repository.get("key1");
+ assertThat(result).isEqualTo("second");
+ }
+
+ @Test
+ void testGetNonExistent() {
+ Object result = repository.get("nonexistent");
+ assertThat(result).isNull();
+ }
+
+ @Test
+ void testDelete() {
+ repository.put("key1", "value1", null);
+ Object deleted = repository.delete("key1");
+ assertThat(deleted).isEqualTo("value1");
+ assertThat(repository.get("key1")).isNull();
+ }
+
+ @Test
+ void testDeleteNonExistent() {
+ Object deleted = repository.delete("nonexistent");
+ assertThat(deleted).isNull();
+ }
+
+ @Test
+ void testContains() {
+ repository.put("key1", "value1", null);
+ assertThat(repository.contains("key1")).isTrue();
+ assertThat(repository.contains("nonexistent")).isFalse();
+ }
+
+ @Test
+ void testKeys() {
+ repository.put("a", "1", null);
+ repository.put("b", "2", null);
+ repository.put("c", "3", null);
+
+ Set<String> keys = repository.keys();
+ assertThat(keys).containsExactlyInAnyOrder("a", "b", "c");
+ }
+
+ @Test
+ void testKeysEmpty() {
+ Set<String> keys = repository.keys();
+ assertThat(keys).isEmpty();
+ }
+
+ @Test
+ void testClear() {
+ repository.put("a", "1", null);
+ repository.put("b", "2", null);
+
+ repository.clear();
+
+ assertThat(repository.keys()).isEmpty();
+ assertThat(repository.contains("a")).isFalse();
+ }
+
+ @Test
+ void testPutIfAbsentNewKey() {
+ Object existing = repository.putIfAbsent("key1", "value1", null);
+ assertThat(existing).isNull();
+ assertThat(repository.get("key1")).isEqualTo("value1");
+ }
+
+ @Test
+ void testPutIfAbsentExistingKey() {
+ repository.put("key1", "original", null);
+ Object existing = repository.putIfAbsent("key1", "replacement", null);
+ assertThat(existing).isEqualTo("original");
+ assertThat(repository.get("key1")).isEqualTo("original");
+ }
+
+ @Test
+ void testSize() {
+ assertThat(repository.size()).isZero();
+
+ repository.put("a", "1", null);
+ repository.put("b", "2", null);
+ assertThat(repository.size()).isEqualTo(2);
+
+ repository.delete("a");
+ assertThat(repository.size()).isEqualTo(1);
+ }
+
+ @Test
+ void testTtlExpiry() {
+ // Store with a short TTL
+ repository.put("expiring", "value", Duration.ofMillis(500));
+ assertThat(repository.contains("expiring")).isTrue();
+
+ // Wait for it to expire
+ await().atMost(5, TimeUnit.SECONDS)
+ .untilAsserted(() ->
assertThat(repository.contains("expiring")).isFalse());
+ }
+
+ @Test
+ void testTtlPutIfAbsentExpiry() {
+ repository.putIfAbsent("expiring", "value", Duration.ofMillis(500));
+ assertThat(repository.contains("expiring")).isTrue();
+
+ await().atMost(5, TimeUnit.SECONDS)
+ .untilAsserted(() ->
assertThat(repository.contains("expiring")).isFalse());
+ }
+
+ @Test
+ void testSerializableValues() {
+ // Test that non-String serializable objects work
+ repository.put("int-key", 42, null);
+ assertThat(repository.get("int-key")).isEqualTo(42);
+
+ repository.put("bool-key", Boolean.TRUE, null);
+ assertThat(repository.get("bool-key")).isEqualTo(Boolean.TRUE);
+ }
+
+ @Test
+ void testKeyPrefixIsolation() {
+ // Create a second repository with a different prefix
+ RedisKeyValueRepository other = new
RedisKeyValueRepository(service.getServiceAddress(), "other-kvr:");
+ other.start();
+ try {
+ repository.put("shared-key", "from-repo1", null);
+ other.put("shared-key", "from-repo2", null);
+
+ assertThat(repository.get("shared-key")).isEqualTo("from-repo1");
+ assertThat(other.get("shared-key")).isEqualTo("from-repo2");
+
+ assertThat(repository.keys()).containsExactly("shared-key");
+ assertThat(other.keys()).containsExactly("shared-key");
+
+ // Clearing one should not affect the other
+ repository.clear();
+ assertThat(repository.keys()).isEmpty();
+ assertThat(other.get("shared-key")).isEqualTo("from-repo2");
+ } finally {
+ other.clear();
+ other.stop();
+ }
+ }
+
+ @Test
+ void testCustomRedissonClient() {
+ Config config = new Config();
+ config.useSingleServer().setAddress(String.format("redis://%s",
service.getServiceAddress()));
+ RedissonClient customClient = Redisson.create(config);
+
+ try {
+ RedisKeyValueRepository customRepo = new RedisKeyValueRepository();
+ customRepo.setKeyPrefix("custom-kvr:");
+ customRepo.setRedisson(customClient);
+ customRepo.start();
+
+ customRepo.put("key1", "value1", null);
+ assertThat(customRepo.get("key1")).isEqualTo("value1");
+
+ customRepo.clear();
+ customRepo.stop();
+
+ // Custom client should not be shut down by the repository
+ assertThat(customClient.isShutdown()).isFalse();
+ } finally {
+ customClient.shutdown();
+ }
+ }
+
+ @Test
+ void testReplaceMatchingValue() {
+ repository.put("key1", "value1", null);
+ boolean replaced = repository.replace("key1", "value1", "value2",
null);
+ assertThat(replaced).isTrue();
+ assertThat(repository.get("key1")).isEqualTo("value2");
+ }
+
+ @Test
+ void testReplaceNonMatchingValue() {
+ repository.put("key1", "value1", null);
+ boolean replaced = repository.replace("key1", "wrong", "value2", null);
+ assertThat(replaced).isFalse();
+ assertThat(repository.get("key1")).isEqualTo("value1");
+ }
+
+ @Test
+ void testReplaceMissingKey() {
+ boolean replaced = repository.replace("nonexistent", "value1",
"value2", null);
+ assertThat(replaced).isFalse();
+ }
+
+ @Test
+ void testReplaceWithTtl() {
+ repository.put("key1", "value1", null);
+ boolean replaced = repository.replace("key1", "value1", "value2",
Duration.ofMillis(500));
+ assertThat(replaced).isTrue();
+ assertThat(repository.get("key1")).isEqualTo("value2");
+
+ await().atMost(5, TimeUnit.SECONDS)
+ .untilAsserted(() ->
assertThat(repository.get("key1")).isNull());
+ }
+
+ @Test
+ void testDeleteWithMatchingValue() {
+ repository.put("key1", "value1", null);
+ boolean deleted = repository.delete("key1", "value1");
+ assertThat(deleted).isTrue();
+ assertThat(repository.get("key1")).isNull();
+ }
+
+ @Test
+ void testDeleteWithNonMatchingValue() {
+ repository.put("key1", "value1", null);
+ boolean deleted = repository.delete("key1", "wrong");
+ assertThat(deleted).isFalse();
+ assertThat(repository.get("key1")).isEqualTo("value1");
+ }
+
+ @Test
+ void testDeleteWithMissingKey() {
+ boolean deleted = repository.delete("nonexistent", "value1");
+ assertThat(deleted).isFalse();
+ }
+
+ @Test
+ void testEndpointValidation() {
+ RedisKeyValueRepository repo = new RedisKeyValueRepository();
+
assertThatThrownBy(repo::init).isInstanceOf(IllegalArgumentException.class);
+ }
+}
