This is an automated email from the ASF dual-hosted git repository.
davsclaus pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/camel-spring-boot.git
The following commit(s) were added to refs/heads/main by this push:
new 4b978c40359 CAMEL-24757: Improve Spring Cache state store backend
support (#2046)
4b978c40359 is described below
commit 4b978c40359b5015edce6ef4f7f0579d87162c32
Author: Ramu <[email protected]>
AuthorDate: Wed Oct 21 10:04:43 2026 +0530
CAMEL-24757: Improve Spring Cache state store backend support (#2046)
* CAMEL-24757: Add Spring Cache-backed KeyValueRepository for
camel-spring-boot
* CAMEL-24757: Finalize SpringCacheKeyValueRepository and address review
feedback
- Document @EnableCaching prerequisite and spring.cache.cache-names
requirement for camel-kvr
- Explicitly detail cross-EIP auto-discovery side effects
(AggregateProcessor, IdempotentConsumer, and Cache EIP)
- Resolve test context runner bean resolution and resource lifecycle
handling
- Add configuration metadata for camel.spring.cache.state-store.enabled
* CAMEL-24757: Add Spring Cache-backed KeyValueRepository adapter
* CAMEL-24757: Move Spring Cache state store to starter
* CAMEL-24757: Fix Spring Cache state store Redis tests
* CAMEL-24757: Improve state store backend tests
* CAMEL-24757: Document state store backend choices
* CAMEL-24757: Clarify Spring Cache setup requirements
---
.../camel-state-store-starter/pom.xml | 108 ++++++
.../src/main/doc/limitations.adoc | 11 +
.../src/main/doc/usage.adoc | 43 +++
.../src/main/docs/state-store.json | 6 +
.../cache/SpringCacheKeyValueRepository.java | 203 ++++++++++
.../SpringCacheStateStoreAutoConfiguration.java | 49 +++
.../additional-spring-configuration-metadata.json | 10 +
...rk.boot.autoconfigure.AutoConfiguration.imports | 1 +
.../cache/SpringCacheKeyValueRepositoryTest.java | 407 +++++++++++++++++++++
...SpringCacheStateStoreAutoConfigurationTest.java | 337 +++++++++++++++++
.../cache/SpringCacheStateStoreRedisIT.java | 185 ++++++++++
.../src/main/docs/spring-boot.adoc | 2 +-
.../modules/ROOT/pages/starters/state-store.adoc | 63 +++-
13 files changed, 1423 insertions(+), 2 deletions(-)
diff --git a/components-starter/camel-state-store-starter/pom.xml
b/components-starter/camel-state-store-starter/pom.xml
index b8992727974..5fa393787cb 100644
--- a/components-starter/camel-state-store-starter/pom.xml
+++ b/components-starter/camel-state-store-starter/pom.xml
@@ -38,6 +38,39 @@
<artifactId>camel-state-store</artifactId>
<version>${camel-version}</version>
</dependency>
+ <dependency>
+ <groupId>com.github.ben-manes.caffeine</groupId>
+ <artifactId>caffeine</artifactId>
+ <scope>test</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.springframework</groupId>
+ <artifactId>spring-context-support</artifactId>
+ <scope>test</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.springframework.boot</groupId>
+ <artifactId>spring-boot-starter-cache</artifactId>
+ <version>${spring-boot-version}</version>
+ <scope>test</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.springframework.boot</groupId>
+ <artifactId>spring-boot-starter-data-redis</artifactId>
+ <version>${spring-boot-version}</version>
+ <scope>test</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.camel</groupId>
+ <artifactId>camel-test-infra-redis</artifactId>
+ <version>${camel-version}</version>
+ <scope>test</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.awaitility</groupId>
+ <artifactId>awaitility</artifactId>
+ <scope>test</scope>
+ </dependency>
<!--START OF GENERATED CODE-->
<dependency>
<groupId>org.apache.camel.springboot</groupId>
@@ -45,4 +78,79 @@
</dependency>
<!--END OF GENERATED CODE-->
</dependencies>
+ <profiles>
+ <profile>
+ <id>state-store-integration-tests-docker-file</id>
+ <activation>
+ <file>
+ <exists>/var/run/docker.sock</exists>
+ </file>
+ </activation>
+ <build>
+ <plugins>
+ <plugin>
+ <groupId>org.apache.maven.plugins</groupId>
+ <artifactId>maven-failsafe-plugin</artifactId>
+ <executions>
+ <execution>
+ <goals>
+ <goal>integration-test</goal>
+ <goal>verify</goal>
+ </goals>
+ </execution>
+ </executions>
+ </plugin>
+ </plugins>
+ </build>
+ </profile>
+ <profile>
+ <id>state-store-integration-tests-docker-env</id>
+ <activation>
+ <property>
+ <name>env.DOCKER_HOST</name>
+ </property>
+ </activation>
+ <build>
+ <plugins>
+ <plugin>
+ <groupId>org.apache.maven.plugins</groupId>
+ <artifactId>maven-failsafe-plugin</artifactId>
+ <executions>
+ <execution>
+ <goals>
+ <goal>integration-test</goal>
+ <goal>verify</goal>
+ </goals>
+ </execution>
+ </executions>
+ </plugin>
+ </plugins>
+ </build>
+ </profile>
+ <profile>
+ <id>state-store-integration-tests-ci</id>
+ <activation>
+ <property>
+ <name>ci.env.name</name>
+ <value>apache.org</value>
+ </property>
+ </activation>
+ <build>
+ <plugins>
+ <plugin>
+ <groupId>org.apache.maven.plugins</groupId>
+ <artifactId>maven-failsafe-plugin</artifactId>
+ <executions>
+ <execution>
+ <goals>
+ <goal>integration-test</goal>
+ <goal>verify</goal>
+ </goals>
+ </execution>
+ </executions>
+ </plugin>
+ </plugins>
+ </build>
+ </profile>
+ </profiles>
</project>
diff --git
a/components-starter/camel-state-store-starter/src/main/doc/limitations.adoc
b/components-starter/camel-state-store-starter/src/main/doc/limitations.adoc
new file mode 100644
index 00000000000..880f77dcb5e
--- /dev/null
+++ b/components-starter/camel-state-store-starter/src/main/doc/limitations.adoc
@@ -0,0 +1,11 @@
+=== Spring Cache
+
+Spring Cache provides a common API, but some operations depend on the
underlying cache provider.
+
+* **Per-entry TTL:** Spring Cache has no portable per-entry TTL API. A
positive, non-zero TTL requested through the repository is not applied by the
Spring adapter. Configure expiration using the underlying cache provider, for
example with `spring.cache.redis.time-to-live` or a Caffeine specification.
+* **Atomic operations:** Spring Cache does not provide a portable atomic
get-and-put operation. The adapter's `put` and `delete` read the previous value
and then write or evict separately, so the returned previous value is not
atomic. `putIfAbsent` delegates to the provider's Spring Cache implementation;
its atomicity depends on that provider. Conditional `replace` and conditional
`delete` may use non-atomic check-then-act operations.
+* **Key enumeration and size:** `keys()` and `size()` are supported only when
`Cache.getNativeCache()` returns a `ConcurrentMap`, such as with the simple
cache provider. They operate on a snapshot of current string keys. Other
providers, including Caffeine, Redis, and JCache, do not support these
operations through the Spring adapter.
+
+=== Camel KeyValueRepository
+
+Choose a Camel repository when you need capabilities beyond the portable
Spring Cache API, such as per-entry TTL, atomic `putIfAbsent` or `replace`,
conditional deletion, and key enumeration or size. Actual support and
guarantees depend on the selected repository implementation.
diff --git
a/components-starter/camel-state-store-starter/src/main/doc/usage.adoc
b/components-starter/camel-state-store-starter/src/main/doc/usage.adoc
new file mode 100644
index 00000000000..b86d9ce7474
--- /dev/null
+++ b/components-starter/camel-state-store-starter/src/main/doc/usage.adoc
@@ -0,0 +1,43 @@
+=== Choosing a backend
+
+The State Store component supports Spring Cache and Camel `KeyValueRepository`
backends.
+
+==== Spring Cache
+
+Set `camel.state-store.spring-cache.enabled=true` to register a
`StateStoreBackendFactory` backed by Spring's `CacheManager`. Each state-store
name maps to a Spring cache with the same name, reusing the application's
`spring.cache.*` configuration.
+
+The factory is registered when a single `CacheManager` candidate is available
and no other `StateStoreBackendFactory` bean is defined. It does not register a
global `KeyValueRepository` bean.
+
+==== Camel KeyValueRepository
+
+Alternatively, use a Camel `KeyValueRepository` implementation, such as
`camel-redis`, `camel-caffeine`, `camel-jcache`, `camel-sql`, or
`camel-infinispan`. Define the repository as a bean and leave
`camel.state-store.spring-cache.enabled` disabled. The State Store component
automatically discovers the repository.
+
+These repositories can also be discovered by Camel features that support
`KeyValueRepository`, including the Aggregate EIP, Idempotent Consumer, and
Cache EIP.
+
+==== Configuration examples
+
+To use Spring Cache, configure a cache provider and enable the factory:
+
+[source,properties]
+----
+spring.cache.type=caffeine
+spring.cache.cache-names=orders,users
+camel.state-store.spring-cache.enabled=true
+----
+
+Spring Boot creates the `CacheManager` from `spring.cache.*` only when the
application enables caching with `@EnableCaching`; otherwise, provide your own
`CacheManager` bean. Without a single eligible `CacheManager`, the factory is
not registered, and stores may fall back to the default in-memory repository.
If `spring.cache.cache-names` specifies a fixed set of names, include every
state-store name you use. A dynamic cache manager can create caches on demand.
+
+To use a specific Camel repository, register it as a bean, for example under
the name `redisRepo`, and select it on the endpoint:
+
+[source,java]
+----
+from("direct:order")
+ .setHeader("CamelStateStoreKey", simple("${header.orderId}"))
+ .to("state-store:orders?backend=#redisRepo");
+----
+
+==== Using both backends
+
+When the Spring Cache factory is enabled, it is the default backend for
state-store endpoints unless an endpoint explicitly selects another backend or
a component-level backend factory takes precedence.
+
+You can therefore use Spring Cache for ordinary stores and explicitly select a
Camel repository for stores that need its additional capabilities.
diff --git
a/components-starter/camel-state-store-starter/src/main/docs/state-store.json
b/components-starter/camel-state-store-starter/src/main/docs/state-store.json
index 20f3e7d3ea4..3b4ff143f73 100644
---
a/components-starter/camel-state-store-starter/src/main/docs/state-store.json
+++
b/components-starter/camel-state-store-starter/src/main/docs/state-store.json
@@ -43,6 +43,12 @@
"description": "Whether the producer should be started lazy (on the
first message). By starting lazy you can use this to allow CamelContext and
routes to startup in situations where a producer may otherwise fail during
starting and cause the route to fail being started. By deferring this startup
to be lazy then the startup failure can be handled during routing messages via
Camel's routing error handlers. Beware that when the first message is processed
then creating and starting the [...]
"sourceType":
"org.apache.camel.component.statestore.springboot.StateStoreComponentConfiguration",
"defaultValue": false
+ },
+ {
+ "name": "camel.state-store.spring-cache.enabled",
+ "type": "java.lang.Boolean",
+ "description": "Whether to register a StateStoreBackendFactory that
creates a Spring Cache-backed KeyValueRepository for each state-store name.
Disabled by default. Requires a single CacheManager bean and no user-defined
StateStoreBackendFactory. Does not register a global KeyValueRepository.",
+ "defaultValue": false
}
],
"hints": [],
diff --git
a/components-starter/camel-state-store-starter/src/main/java/org/apache/camel/component/statestore/springboot/cache/SpringCacheKeyValueRepository.java
b/components-starter/camel-state-store-starter/src/main/java/org/apache/camel/component/statestore/springboot/cache/SpringCacheKeyValueRepository.java
new file mode 100644
index 00000000000..a9f9a5d4429
--- /dev/null
+++
b/components-starter/camel-state-store-starter/src/main/java/org/apache/camel/component/statestore/springboot/cache/SpringCacheKeyValueRepository.java
@@ -0,0 +1,203 @@
+/*
+ * 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.statestore.springboot.cache;
+
+import java.time.Duration;
+import java.util.Collections;
+import java.util.HashSet;
+import java.util.Set;
+import java.util.concurrent.ConcurrentMap;
+import java.util.concurrent.atomic.AtomicBoolean;
+
+import org.apache.camel.spi.KeyValueRepository;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.cache.Cache;
+import org.springframework.cache.CacheManager;
+import org.springframework.cache.concurrent.ConcurrentMapCache;
+import org.springframework.util.Assert;
+
+/**
+ * An Apache Camel {@link KeyValueRepository} adapter backed by Spring's
+ * {@link CacheManager}. Each instance is bound to one Spring cache whose name
+ * matches the Camel state-store name.
+ * <h3>Design Constraints and Backend Behavior</h3>
+ * <ul>
+ * <li><strong>TTL support:</strong> Spring Cache does not define a standard
+ * per-entry TTL API. A positive, non-zero TTL triggers a single
+ * {@code WARN} log per repository instance. The entry is stored without
+ * applying the requested TTL. {@code null}, zero, and negative TTL
values
+ * do not trigger this warning or consume the warning-once guard. Users
+ * requiring expiration must configure it through the underlying cache
+ * provider or {@link CacheManager}, for example using a Caffeine cache
+ * specification.</li>
+ * <li><strong>Atomicity:</strong> Spring Cache has no atomic get-and-put
+ * API. {@code put} and {@code delete} read the previous value and then
+ * write or evict in a second call, so the returned old value is not
+ * atomic. {@code putIfAbsent} delegates to
+ * {@link Cache#putIfAbsent(Object, Object)}, whose atomicity depends on
+ * the provider. Conditional operations such as {@code replace} and
+ * conditional {@code delete} use the {@link KeyValueRepository} SPI
+ * default implementations and may perform non-atomic check-then-act
+ * operations.</li>
+ * <li><strong>Enumeration:</strong> {@link #keys()} and {@link #size()} are
+ * supported only when {@link Cache#getNativeCache()} is a
+ * {@link ConcurrentMap}, for example {@link ConcurrentMapCache}. They
+ * operate on a snapshot of current {@link String} keys; {@link #size()}
+ * is the size of that snapshot, and non-{@link String} native keys are
+ * ignored. Other native cache types throw
+ * {@link UnsupportedOperationException} from {@link #keys()} and
+ * {@link #size()} only; {@code get}, {@code put}, {@code delete}, and
+ * related operations still delegate to Spring Cache. Spring's Caffeine,
+ * Redis, and JCache adapters do not return a {@code ConcurrentMap} from
+ * {@code getNativeCache()}, so enumeration is unsupported for those
+ * backends. When this adapter is bound explicitly, Camel operations that
+ * enumerate keys (Aggregate recovery, {@code completionTimeout} restore
+ * via {@code getKeys()}, and Idempotent Consumer {@code clear()})
+ * therefore require a {@code ConcurrentMap} native cache.</li>
+ * <li><strong>Scope:</strong> The repository is bound to the cache named by
+ * the constructor. {@link #clear()} invalidates that entire cache. Use a
+ * distinct store name (and therefore cache name) for each isolated
+ * store.</li>
+ * </ul>
+ */
+public class SpringCacheKeyValueRepository implements KeyValueRepository {
+
+ private static final Logger LOG =
LoggerFactory.getLogger(SpringCacheKeyValueRepository.class);
+ private static final String ENUMERATION_UNSUPPORTED
+ = "Spring Cache does not provide a portable key enumeration API;
keys() and size() are supported only when Cache.getNativeCache() is a
ConcurrentMap.";
+
+ private final CacheManager cacheManager;
+ private final String cacheName;
+ private final AtomicBoolean ttlWarningLogged = new AtomicBoolean(false);
+
+ public SpringCacheKeyValueRepository(CacheManager cacheManager, String
cacheName) {
+ Assert.notNull(cacheManager, "CacheManager must not be null");
+ Assert.hasText(cacheName, "CacheName must not be empty");
+ this.cacheManager = cacheManager;
+ this.cacheName = cacheName;
+ }
+
+ public String getCacheName() {
+ return cacheName;
+ }
+
+ private Cache getCache() {
+ Cache cache = cacheManager.getCache(cacheName);
+ if (cache == null) {
+ throw new IllegalStateException(String.format("Spring Cache '%s'
not found in CacheManager of type %s",
+ cacheName, cacheManager.getClass().getName()));
+ }
+ return cache;
+ }
+
+ @Override
+ public void start() {
+ // Lifecycle start hook (managed externally by Spring CacheManager
container)
+ }
+
+ @Override
+ public void stop() {
+ // Lifecycle stop hook (managed externally by Spring CacheManager
container)
+ }
+
+ @Override
+ public Object get(String key) {
+ Cache.ValueWrapper wrapper = getCache().get(key);
+ return wrapper != null ? wrapper.get() : null;
+ }
+
+ @Override
+ public Object put(String key, Object value) {
+ return put(key, value, null);
+ }
+
+ @Override
+ public Object put(String key, Object value, Duration ttl) {
+ checkTtl(ttl);
+ Cache cache = getCache();
+ Cache.ValueWrapper oldWrapper = cache.get(key);
+ Object oldValue = oldWrapper != null ? oldWrapper.get() : null;
+ cache.put(key, value);
+ return oldValue;
+ }
+
+ @Override
+ public Object putIfAbsent(String key, Object value) {
+ return putIfAbsent(key, value, null);
+ }
+
+ @Override
+ public Object putIfAbsent(String key, Object value, Duration ttl) {
+ checkTtl(ttl);
+ Cache.ValueWrapper existing = getCache().putIfAbsent(key, value);
+ return existing != null ? existing.get() : null;
+ }
+
+ @Override
+ public Object delete(String key) {
+ Cache cache = getCache();
+ Cache.ValueWrapper oldWrapper = cache.get(key);
+ Object oldValue = oldWrapper != null ? oldWrapper.get() : null;
+ cache.evictIfPresent(key);
+ return oldValue;
+ }
+
+ @Override
+ public boolean contains(String key) {
+ return getCache().get(key) != null;
+ }
+
+ @Override
+ public void clear() {
+ Cache cache = getCache();
+ cache.invalidate();
+ }
+
+ @Override
+ public Set<String> keys() {
+ return Collections.unmodifiableSet(snapshotStringKeys());
+ }
+
+ @Override
+ public int size() {
+ return snapshotStringKeys().size();
+ }
+
+ private Set<String> snapshotStringKeys() {
+ Object nativeCache = getCache().getNativeCache();
+ if (!(nativeCache instanceof ConcurrentMap<?, ?> concurrentMap)) {
+ throw new UnsupportedOperationException(ENUMERATION_UNSUPPORTED);
+ }
+ Set<String> snapshot = new HashSet<>();
+ for (Object key : concurrentMap.keySet()) {
+ if (key instanceof String stringKey) {
+ snapshot.add(stringKey);
+ }
+ }
+ return snapshot;
+ }
+
+ private void checkTtl(Duration ttl) {
+ if (ttl != null && !ttl.isNegative() && !ttl.isZero()) {
+ if (ttlWarningLogged.compareAndSet(false, true)) {
+ LOG.warn("Per-entry TTL of {} is not supported by the Spring
Cache backend for store '{}'; storing without expiration.",
+ ttl, cacheName);
+ }
+ }
+ }
+}
diff --git
a/components-starter/camel-state-store-starter/src/main/java/org/apache/camel/component/statestore/springboot/cache/SpringCacheStateStoreAutoConfiguration.java
b/components-starter/camel-state-store-starter/src/main/java/org/apache/camel/component/statestore/springboot/cache/SpringCacheStateStoreAutoConfiguration.java
new file mode 100644
index 00000000000..a8720f3d4f6
--- /dev/null
+++
b/components-starter/camel-state-store-starter/src/main/java/org/apache/camel/component/statestore/springboot/cache/SpringCacheStateStoreAutoConfiguration.java
@@ -0,0 +1,49 @@
+/*
+ * 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.statestore.springboot.cache;
+
+import org.apache.camel.component.statestore.StateStoreBackendFactory;
+import org.apache.camel.spi.KeyValueRepository;
+import org.apache.camel.spring.boot.CamelAutoConfiguration;
+import org.springframework.boot.autoconfigure.AutoConfiguration;
+import org.springframework.boot.autoconfigure.condition.ConditionalOnClass;
+import
org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean;
+import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
+import
org.springframework.boot.autoconfigure.condition.ConditionalOnSingleCandidate;
+import org.springframework.cache.CacheManager;
+import org.springframework.context.annotation.Bean;
+
+/**
+ * Opt-in auto-configuration that registers a {@link StateStoreBackendFactory}
+ * backed by Spring Cache. Disabled unless
+ * {@code camel.state-store.spring-cache.enabled=true}. The factory creates one
+ * {@link SpringCacheKeyValueRepository} per state-store name and does not
+ * register a global {@link KeyValueRepository} bean.
+ */
+@AutoConfiguration(after = CamelAutoConfiguration.class,
+ afterName =
"org.springframework.boot.cache.autoconfigure.CacheAutoConfiguration")
+@ConditionalOnClass({ CacheManager.class, StateStoreBackendFactory.class })
+@ConditionalOnProperty(prefix = "camel.state-store.spring-cache", name =
"enabled", havingValue = "true")
+public class SpringCacheStateStoreAutoConfiguration {
+
+ @Bean
+ @ConditionalOnSingleCandidate(CacheManager.class)
+ @ConditionalOnMissingBean(StateStoreBackendFactory.class)
+ public StateStoreBackendFactory
springCacheStateStoreBackendFactory(CacheManager cacheManager) {
+ return storeName -> new SpringCacheKeyValueRepository(cacheManager,
storeName);
+ }
+}
diff --git
a/components-starter/camel-state-store-starter/src/main/resources/META-INF/additional-spring-configuration-metadata.json
b/components-starter/camel-state-store-starter/src/main/resources/META-INF/additional-spring-configuration-metadata.json
new file mode 100644
index 00000000000..2e72f1239d6
--- /dev/null
+++
b/components-starter/camel-state-store-starter/src/main/resources/META-INF/additional-spring-configuration-metadata.json
@@ -0,0 +1,10 @@
+{
+ "properties": [
+ {
+ "name": "camel.state-store.spring-cache.enabled",
+ "type": "java.lang.Boolean",
+ "description": "Whether to register a StateStoreBackendFactory that
creates a Spring Cache-backed KeyValueRepository for each state-store name.
Disabled by default. Requires a single CacheManager bean and no user-defined
StateStoreBackendFactory. Does not register a global KeyValueRepository.",
+ "defaultValue": false
+ }
+ ]
+}
diff --git
a/components-starter/camel-state-store-starter/src/main/resources/META-INF/spring/org.springframework.boot.autoconfigure.AutoConfiguration.imports
b/components-starter/camel-state-store-starter/src/main/resources/META-INF/spring/org.springframework.boot.autoconfigure.AutoConfiguration.imports
index b4d44e5a8ff..aad76b7df12 100644
---
a/components-starter/camel-state-store-starter/src/main/resources/META-INF/spring/org.springframework.boot.autoconfigure.AutoConfiguration.imports
+++
b/components-starter/camel-state-store-starter/src/main/resources/META-INF/spring/org.springframework.boot.autoconfigure.AutoConfiguration.imports
@@ -16,3 +16,4 @@
## ---------------------------------------------------------------------------
org.apache.camel.component.statestore.springboot.StateStoreComponentAutoConfiguration
org.apache.camel.component.statestore.springboot.StateStoreComponentConverter
+org.apache.camel.component.statestore.springboot.cache.SpringCacheStateStoreAutoConfiguration
diff --git
a/components-starter/camel-state-store-starter/src/test/java/org/apache/camel/component/statestore/springboot/cache/SpringCacheKeyValueRepositoryTest.java
b/components-starter/camel-state-store-starter/src/test/java/org/apache/camel/component/statestore/springboot/cache/SpringCacheKeyValueRepositoryTest.java
new file mode 100644
index 00000000000..aa86a491b90
--- /dev/null
+++
b/components-starter/camel-state-store-starter/src/test/java/org/apache/camel/component/statestore/springboot/cache/SpringCacheKeyValueRepositoryTest.java
@@ -0,0 +1,407 @@
+/*
+ * 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.statestore.springboot.cache;
+
+import java.time.Duration;
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.UUID;
+import java.util.concurrent.Callable;
+import java.util.concurrent.ConcurrentMap;
+import java.util.concurrent.CopyOnWriteArrayList;
+
+import ch.qos.logback.classic.Level;
+import ch.qos.logback.classic.Logger;
+import ch.qos.logback.classic.spi.ILoggingEvent;
+import ch.qos.logback.core.AppenderBase;
+
+import org.apache.camel.spi.KeyValueRepository;
+import org.junit.jupiter.api.Test;
+import org.slf4j.LoggerFactory;
+import org.springframework.cache.Cache;
+import org.springframework.cache.CacheManager;
+import org.springframework.cache.caffeine.CaffeineCacheManager;
+import org.springframework.cache.concurrent.ConcurrentMapCache;
+import org.springframework.cache.concurrent.ConcurrentMapCacheManager;
+import org.springframework.cache.support.SimpleCacheManager;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+class SpringCacheKeyValueRepositoryTest {
+
+ @Test
+ void shouldHandleCrudLifecycleAndReturnValues() {
+ CacheManager cacheManager = new ConcurrentMapCacheManager("orders");
+ SpringCacheKeyValueRepository repo = new
SpringCacheKeyValueRepository(cacheManager, "orders");
+ assertThat(repo.contains("key1")).isFalse();
+ assertThat(repo.get("key1")).isNull();
+
+ Object previous = repo.put("key1", "val1");
+ assertThat(previous).isNull();
+ assertThat(repo.contains("key1")).isTrue();
+ assertThat(repo.get("key1")).isEqualTo("val1");
+
+ Object updatedPrevious = repo.put("key1", "val2");
+ assertThat(updatedPrevious).isEqualTo("val1");
+ assertThat(repo.get("key1")).isEqualTo("val2");
+
+ Object deletedValue = repo.delete("key1");
+ assertThat(deletedValue).isEqualTo("val2");
+ assertThat(repo.contains("key1")).isFalse();
+
+ assertThat(repo.delete("non-existent")).isNull();
+ }
+
+ @Test
+ void shouldHandlePutIfAbsent() {
+ CacheManager cacheManager = new ConcurrentMapCacheManager("orders");
+ SpringCacheKeyValueRepository repo = new
SpringCacheKeyValueRepository(cacheManager, "orders");
+ Object existing1 = repo.putIfAbsent("key1", "val1");
+ assertThat(existing1).isNull();
+ assertThat(repo.get("key1")).isEqualTo("val1");
+
+ Object existing2 = repo.putIfAbsent("key1", "newVal");
+ assertThat(existing2).isEqualTo("val1");
+ assertThat(repo.get("key1")).isEqualTo("val1");
+ }
+
+ @Test
+ void shouldClearOnlyTheNamedCache() {
+ CacheManager cacheManager = new ConcurrentMapCacheManager("orders");
+ SpringCacheKeyValueRepository repo = new
SpringCacheKeyValueRepository(cacheManager, "orders");
+ repo.put("key1", "val1");
+ repo.put("key2", "val2");
+ repo.clear();
+
+ assertThat(repo.contains("key1")).isFalse();
+ assertThat(repo.contains("key2")).isFalse();
+ }
+
+ @Test
+ void shouldSupportInheritedConditionalOperations() {
+ CacheManager cacheManager = new ConcurrentMapCacheManager("orders");
+ SpringCacheKeyValueRepository repo = new
SpringCacheKeyValueRepository(cacheManager, "orders");
+ repo.put("key1", "val1");
+
+ boolean replaced = repo.replace("key1", "val1", "val2", null);
+ assertThat(replaced).isTrue();
+ assertThat(repo.get("key1")).isEqualTo("val2");
+
+ boolean deletedMismatch = repo.delete("key1", "wrongVal");
+ assertThat(deletedMismatch).isFalse();
+ assertThat(repo.contains("key1")).isTrue();
+
+ boolean deletedMatch = repo.delete("key1", "val2");
+ assertThat(deletedMatch).isTrue();
+ assertThat(repo.contains("key1")).isFalse();
+ }
+
+ @Test
+ void shouldThrowWhenNamedCacheIsMissing() {
+ SimpleCacheManager strictCacheManager = new SimpleCacheManager();
+ strictCacheManager.setCaches(Collections.singletonList(new
ConcurrentMapCache("other-cache")));
+ strictCacheManager.initializeCaches();
+
+ SpringCacheKeyValueRepository repo = new
SpringCacheKeyValueRepository(strictCacheManager, "orders");
+ assertThatThrownBy(() -> repo.get("key1"))
+ .isInstanceOf(IllegalStateException.class)
+ .hasMessageContaining("Spring Cache 'orders' not found in
CacheManager");
+ }
+
+ @Test
+ void shouldEnumerateKeysAndSizeForConcurrentMapNativeCache() {
+ CacheManager cacheManager = new ConcurrentMapCacheManager("orders");
+ SpringCacheKeyValueRepository repo = new
SpringCacheKeyValueRepository(cacheManager, "orders");
+ repo.put("key1", "val1");
+ repo.put("key2", "val2");
+
+ Cache cache = cacheManager.getCache("orders");
+ assertThat(cache).isNotNull();
+ assertThat(cache.getNativeCache()).isInstanceOf(ConcurrentMap.class);
+ cache.put(42, "non-string-key");
+
+ Set<String> keys = repo.keys();
+ assertThat(keys).containsExactlyInAnyOrder("key1", "key2");
+ assertThat(repo.size()).isEqualTo(2);
+ assertThat(repo.size()).isEqualTo(keys.size());
+
+ repo.put("key3", "val3");
+ assertThat(keys).doesNotContain("key3");
+ assertThat(keys).isUnmodifiable();
+ assertThat(repo.keys()).containsExactlyInAnyOrder("key1", "key2",
"key3");
+ assertThat(repo.size()).isEqualTo(3);
+ }
+
+ @Test
+ void shouldThrowUnsupportedForKeysAndSizeWithRealCaffeineCache() {
+ CaffeineCacheManager cacheManager = new CaffeineCacheManager("orders");
+ KeyValueRepository repo = new
SpringCacheKeyValueRepository(cacheManager, "orders");
+ repo.put("key1", "val1");
+
+ Cache cache = cacheManager.getCache("orders");
+ assertThat(cache).isNotNull();
+
assertThat(cache.getNativeCache()).isNotInstanceOf(ConcurrentMap.class);
+ assertThat(repo.get("key1")).isEqualTo("val1");
+
+ assertThatThrownBy(repo::keys)
+ .isInstanceOf(UnsupportedOperationException.class)
+ .hasMessageContaining("ConcurrentMap");
+ assertThatThrownBy(repo::size)
+ .isInstanceOf(UnsupportedOperationException.class)
+ .hasMessageContaining("ConcurrentMap");
+ }
+
+ @Test
+ void
shouldThrowUnsupportedForKeysAndSizeWhenNativeCacheIsNotConcurrentMap() {
+ CacheManager cacheManager = new SingleCacheManager(new
NonConcurrentNativeCache("orders"));
+ KeyValueRepository repo = new
SpringCacheKeyValueRepository(cacheManager, "orders");
+ repo.put("key1", "val1");
+
+ assertThatThrownBy(repo::keys)
+ .isInstanceOf(UnsupportedOperationException.class)
+ .hasMessageContaining("portable key enumeration API")
+ .hasMessageContaining("ConcurrentMap");
+
+ assertThatThrownBy(repo::size)
+ .isInstanceOf(UnsupportedOperationException.class)
+ .hasMessageContaining("portable key enumeration API")
+ .hasMessageContaining("ConcurrentMap");
+
+ assertThat(repo.get("key1")).isEqualTo("val1");
+ }
+
+ @Test
+ void shouldStoreEntriesWhenTtlIsPositiveWithoutApplyingExpiration() {
+ CacheManager cacheManager = new ConcurrentMapCacheManager("orders");
+ SpringCacheKeyValueRepository repo = new
SpringCacheKeyValueRepository(cacheManager, "orders");
+ repo.put("k1", "v1", Duration.ofMinutes(5));
+ repo.putIfAbsent("k2", "v2", Duration.ofMinutes(10));
+ repo.put("k1", "v3", Duration.ofMinutes(2));
+
+ assertThat(repo.get("k1")).isEqualTo("v3");
+ assertThat(repo.get("k2")).isEqualTo("v2");
+ }
+
+ @Test
+ void shouldMaintainIsolationBetweenStoreNames() {
+ CacheManager cacheManager = new ConcurrentMapCacheManager("orders",
"users");
+ SpringCacheKeyValueRepository orders = new
SpringCacheKeyValueRepository(cacheManager, "orders");
+ SpringCacheKeyValueRepository users = new
SpringCacheKeyValueRepository(cacheManager, "users");
+
+ orders.put("id", "ORDER_PLACED");
+ users.put("id", "IN_STOCK");
+
+ assertThat(orders.get("id")).isEqualTo("ORDER_PLACED");
+ assertThat(users.get("id")).isEqualTo("IN_STOCK");
+
+ orders.clear();
+
+ assertThat(orders.contains("id")).isFalse();
+ assertThat(users.contains("id")).isTrue();
+ }
+
+ @Test
+ void ttlWarningLoggedOnlyOncePerRepositoryInstance() {
+ String cacheName = uniqueCacheName();
+ Logger logger = (Logger)
LoggerFactory.getLogger(SpringCacheKeyValueRepository.class);
+ ThreadSafeTestAppender appender = attachAppender(logger);
+
+ try {
+ CacheManager cacheManager = new
ConcurrentMapCacheManager(cacheName);
+ KeyValueRepository first = new
SpringCacheKeyValueRepository(cacheManager, cacheName);
+ KeyValueRepository second = new
SpringCacheKeyValueRepository(cacheManager, cacheName);
+
+ first.put("key1", "value1", Duration.ofSeconds(10));
+ first.put("key2", "value2", Duration.ofSeconds(20));
+ second.put("key3", "value3", Duration.ofSeconds(10));
+
+ assertThat(ttlWarningCount(appender.getEvents(),
cacheName)).isEqualTo(2);
+ } finally {
+ detachAppender(logger, appender);
+ }
+ }
+
+ @Test
+ void nonPositiveTtlShouldNotTriggerWarningOrConsumeGuard() {
+ String cacheName = uniqueCacheName();
+ Logger logger = (Logger)
LoggerFactory.getLogger(SpringCacheKeyValueRepository.class);
+ ThreadSafeTestAppender appender = attachAppender(logger);
+
+ try {
+ CacheManager cacheManager = new
ConcurrentMapCacheManager(cacheName);
+ KeyValueRepository repository = new
SpringCacheKeyValueRepository(cacheManager, cacheName);
+
+ repository.put("key-null", "value", null);
+ repository.put("key-zero", "value", Duration.ZERO);
+ repository.put("key-negative", "value", Duration.ofSeconds(-5));
+
+ assertThat(ttlWarningCount(appender.getEvents(),
cacheName)).isZero();
+
+ repository.put("key-positive", "value", Duration.ofSeconds(10));
+
+ assertThat(ttlWarningCount(appender.getEvents(),
cacheName)).isEqualTo(1);
+ } finally {
+ detachAppender(logger, appender);
+ }
+ }
+
+ private String uniqueCacheName() {
+ return "ttl-warning-test-" + UUID.randomUUID();
+ }
+
+ private ThreadSafeTestAppender attachAppender(Logger logger) {
+ ThreadSafeTestAppender appender = new ThreadSafeTestAppender();
+ appender.setContext(logger.getLoggerContext());
+ appender.start();
+ logger.addAppender(appender);
+ return appender;
+ }
+
+ private void detachAppender(Logger logger, ThreadSafeTestAppender
appender) {
+ logger.detachAppender(appender);
+ appender.stop();
+ }
+
+ private long ttlWarningCount(List<ILoggingEvent> events, String cacheName)
{
+ List<ILoggingEvent> snapshot = new ArrayList<>(events);
+ String expectedStore = "store '" + cacheName + "'";
+
+ return snapshot.stream()
+ .filter(event -> event.getLevel() == Level.WARN)
+ .filter(event ->
SpringCacheKeyValueRepository.class.getName().equals(event.getLoggerName()))
+ .filter(event ->
event.getFormattedMessage().contains("Per-entry TTL"))
+ .filter(event ->
event.getFormattedMessage().contains(expectedStore))
+ .count();
+ }
+
+ static final class ThreadSafeTestAppender extends
AppenderBase<ILoggingEvent> {
+ private final List<ILoggingEvent> events = new
CopyOnWriteArrayList<>();
+
+ @Override
+ protected void append(ILoggingEvent eventObject) {
+ events.add(eventObject);
+ }
+
+ public List<ILoggingEvent> getEvents() {
+ return events;
+ }
+ }
+
+ static final class NonConcurrentNativeCache implements Cache {
+ private final String name;
+ private final Map<Object, Object> store = new HashMap<>();
+
+ NonConcurrentNativeCache(String name) {
+ this.name = name;
+ }
+
+ @Override
+ public String getName() {
+ return name;
+ }
+
+ @Override
+ public Object getNativeCache() {
+ return this;
+ }
+
+ @Override
+ public ValueWrapper get(Object key) {
+ if (!store.containsKey(key)) {
+ return null;
+ }
+ Object value = store.get(key);
+ return () -> value;
+ }
+
+ @Override
+ @SuppressWarnings("unchecked")
+ public <T> T get(Object key, Class<T> type) {
+ Object value = store.get(key);
+ if (value != null && type != null && !type.isInstance(value)) {
+ throw new IllegalStateException("Cached value is not of
required type [" + type.getName() + "]: " + value);
+ }
+ return (T) value;
+ }
+
+ @Override
+ @SuppressWarnings("unchecked")
+ public <T> T get(Object key, Callable<T> valueLoader) {
+ if (store.containsKey(key)) {
+ return (T) store.get(key);
+ }
+ try {
+ T value = valueLoader.call();
+ store.put(key, value);
+ return value;
+ } catch (Exception ex) {
+ throw new ValueRetrievalException(key, valueLoader, ex);
+ }
+ }
+
+ @Override
+ public void put(Object key, Object value) {
+ store.put(key, value);
+ }
+
+ @Override
+ public void evict(Object key) {
+ store.remove(key);
+ }
+
+ @Override
+ public boolean evictIfPresent(Object key) {
+ return store.remove(key) != null;
+ }
+
+ @Override
+ public void clear() {
+ store.clear();
+ }
+
+ @Override
+ public boolean invalidate() {
+ boolean hadEntries = !store.isEmpty();
+ store.clear();
+ return hadEntries;
+ }
+ }
+
+ static final class SingleCacheManager implements CacheManager {
+ private final Cache cache;
+
+ SingleCacheManager(Cache cache) {
+ this.cache = cache;
+ }
+
+ @Override
+ public Cache getCache(String name) {
+ return cache.getName().equals(name) ? cache : null;
+ }
+
+ @Override
+ public Collection<String> getCacheNames() {
+ return Collections.singleton(cache.getName());
+ }
+ }
+}
diff --git
a/components-starter/camel-state-store-starter/src/test/java/org/apache/camel/component/statestore/springboot/cache/SpringCacheStateStoreAutoConfigurationTest.java
b/components-starter/camel-state-store-starter/src/test/java/org/apache/camel/component/statestore/springboot/cache/SpringCacheStateStoreAutoConfigurationTest.java
new file mode 100644
index 00000000000..15b86585c46
--- /dev/null
+++
b/components-starter/camel-state-store-starter/src/test/java/org/apache/camel/component/statestore/springboot/cache/SpringCacheStateStoreAutoConfigurationTest.java
@@ -0,0 +1,337 @@
+/*
+ * 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.statestore.springboot.cache;
+
+import java.time.Duration;
+import java.util.Map;
+import java.util.concurrent.TimeUnit;
+
+import org.apache.camel.CamelContext;
+import org.apache.camel.ProducerTemplate;
+import org.apache.camel.builder.RouteBuilder;
+import org.apache.camel.component.statestore.StateStoreBackendFactory;
+import org.apache.camel.component.statestore.StateStoreComponent;
+import org.apache.camel.component.statestore.StateStoreConstants;
+import
org.apache.camel.component.statestore.springboot.StateStoreComponentAutoConfiguration;
+import
org.apache.camel.component.statestore.springboot.StateStoreComponentConverter;
+import org.apache.camel.processor.aggregate.UseLatestAggregationStrategy;
+import org.apache.camel.spi.KeyValueRepository;
+import org.apache.camel.spring.boot.CamelAutoConfiguration;
+import org.apache.camel.support.MemoryKeyValueRepository;
+import org.junit.jupiter.api.Test;
+import org.springframework.boot.autoconfigure.AutoConfigurations;
+import org.springframework.boot.test.context.runner.ApplicationContextRunner;
+import org.springframework.cache.CacheManager;
+import org.springframework.cache.concurrent.ConcurrentMapCacheManager;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.awaitility.Awaitility.await;
+
+class SpringCacheStateStoreAutoConfigurationTest {
+
+ private final ApplicationContextRunner contextRunner = new
ApplicationContextRunner()
+
.withConfiguration(AutoConfigurations.of(SpringCacheStateStoreAutoConfiguration.class))
+ .withPropertyValues("spring.main.banner-mode=off");
+
+ private final ApplicationContextRunner camelRunner = new
ApplicationContextRunner()
+ .withConfiguration(AutoConfigurations.of(
+ CamelAutoConfiguration.class,
+ StateStoreComponentAutoConfiguration.class,
+ SpringCacheStateStoreAutoConfiguration.class))
+ .withUserConfiguration(StateStoreComponentConverter.class)
+ .withPropertyValues("spring.main.banner-mode=off");
+
+ private final ApplicationContextRunner mixedBackendRunner = new
ApplicationContextRunner()
+ .withConfiguration(AutoConfigurations.of(
+ CamelAutoConfiguration.class,
+ StateStoreComponentAutoConfiguration.class,
+ SpringCacheStateStoreAutoConfiguration.class))
+ .withUserConfiguration(StateStoreComponentConverter.class)
+ .withBean(CacheManager.class, ConcurrentMapCacheManager::new)
+ .withBean("camelRepo", MemoryKeyValueRepository.class,
MemoryKeyValueRepository::new)
+ .withPropertyValues("spring.main.banner-mode=off");
+
+ @Test
+ void doesNotRegisterFactoryWhenPropertyIsAbsent() {
+ contextRunner
+ .withBean("cacheManager", CacheManager.class,
ConcurrentMapCacheManager::new)
+ .run(context -> {
+
assertThat(context).doesNotHaveBean(SpringCacheStateStoreAutoConfiguration.class);
+
assertThat(context).doesNotHaveBean(StateStoreBackendFactory.class);
+
assertThat(context).doesNotHaveBean(KeyValueRepository.class);
+ });
+ }
+
+ @Test
+ void doesNotRegisterFactoryWhenPropertyIsFalse() {
+ contextRunner
+ .withBean("cacheManager", CacheManager.class,
ConcurrentMapCacheManager::new)
+
.withPropertyValues("camel.state-store.spring-cache.enabled=false")
+ .run(context -> {
+
assertThat(context).doesNotHaveBean(StateStoreBackendFactory.class);
+
assertThat(context).doesNotHaveBean(KeyValueRepository.class);
+ });
+ }
+
+ @Test
+ void doesNotRegisterFactoryWhenCacheManagerIsAbsent() {
+ contextRunner
+
.withPropertyValues("camel.state-store.spring-cache.enabled=true")
+ .run(context -> {
+
assertThat(context).doesNotHaveBean(StateStoreBackendFactory.class);
+
assertThat(context).doesNotHaveBean(KeyValueRepository.class);
+ });
+ }
+
+ @Test
+ void registersFactoryWhenEnabledWithSingleCacheManager() {
+ contextRunner
+ .withBean("cacheManager", CacheManager.class,
ConcurrentMapCacheManager::new)
+
.withPropertyValues("camel.state-store.spring-cache.enabled=true")
+ .run(context -> {
+
assertThat(context).hasSingleBean(StateStoreBackendFactory.class);
+
assertThat(context).doesNotHaveBean(KeyValueRepository.class);
+
+ StateStoreBackendFactory factory =
context.getBean(StateStoreBackendFactory.class);
+ KeyValueRepository repository =
factory.createBackend("orders");
+
assertThat(repository).isInstanceOf(SpringCacheKeyValueRepository.class);
+ assertThat(((SpringCacheKeyValueRepository)
repository).getCacheName()).isEqualTo("orders");
+ });
+ }
+
+ @Test
+ void doesNotRegisterFactoryWhenMultipleCacheManagersPresent() {
+ contextRunner
+ .withBean("cacheManagerOne", CacheManager.class,
ConcurrentMapCacheManager::new)
+ .withBean("cacheManagerTwo", CacheManager.class,
ConcurrentMapCacheManager::new)
+
.withPropertyValues("camel.state-store.spring-cache.enabled=true")
+ .run(context -> {
+
assertThat(context.getBeansOfType(CacheManager.class)).hasSize(2);
+
assertThat(context).doesNotHaveBean(StateStoreBackendFactory.class);
+
assertThat(context).doesNotHaveBean(KeyValueRepository.class);
+ });
+ }
+
+ @Test
+ void backsOffWhenCustomFactoryExists() {
+ StateStoreBackendFactory customFactory = storeName -> new
MemoryKeyValueRepository();
+ contextRunner
+ .withBean("cacheManager", CacheManager.class,
ConcurrentMapCacheManager::new)
+ .withBean("customFactory", StateStoreBackendFactory.class, ()
-> customFactory)
+
.withPropertyValues("camel.state-store.spring-cache.enabled=true")
+ .run(context -> {
+
assertThat(context).hasSingleBean(StateStoreBackendFactory.class);
+
assertThat(context.getBean(StateStoreBackendFactory.class)).isSameAs(customFactory);
+
assertThat(context.getBean(StateStoreBackendFactory.class).createBackend("orders"))
+ .isInstanceOf(MemoryKeyValueRepository.class);
+
assertThat(context).doesNotHaveBean(KeyValueRepository.class);
+ });
+ }
+
+ @Test
+ void createsIndependentRepositoriesPerStoreName() {
+ contextRunner
+ .withBean("cacheManager", CacheManager.class,
ConcurrentMapCacheManager::new)
+
.withPropertyValues("camel.state-store.spring-cache.enabled=true")
+ .run(context -> {
+ StateStoreBackendFactory factory =
context.getBean(StateStoreBackendFactory.class);
+ KeyValueRepository orders =
factory.createBackend("orders");
+ KeyValueRepository users = factory.createBackend("users");
+
+ assertThat(((SpringCacheKeyValueRepository)
orders).getCacheName()).isEqualTo("orders");
+ assertThat(((SpringCacheKeyValueRepository)
users).getCacheName()).isEqualTo("users");
+
+ orders.put("id", "order-1");
+ users.put("id", "user-1");
+
+ assertThat(orders.get("id")).isEqualTo("order-1");
+ assertThat(users.get("id")).isEqualTo("user-1");
+
+ orders.clear();
+ assertThat(orders.contains("id")).isFalse();
+ assertThat(users.get("id")).isEqualTo("user-1");
+ });
+ }
+
+ @Test
+ void doesNotExposeGlobalRepositoryToUnrelatedEips() {
+ camelRunner
+ .withBean("cacheManager", CacheManager.class,
ConcurrentMapCacheManager::new)
+
.withPropertyValues("camel.state-store.spring-cache.enabled=true")
+ .run(context -> {
+
assertThat(context).hasSingleBean(StateStoreBackendFactory.class);
+
assertThat(context).doesNotHaveBean(KeyValueRepository.class);
+
+ CamelContext camelContext =
context.getBean(CamelContext.class);
+
assertThat(camelContext.getRegistry().findSingleByType(KeyValueRepository.class)).isNull();
+
assertThat(camelContext.getRegistry().findByType(KeyValueRepository.class)).isEmpty();
+
+ StateStoreComponent component =
camelContext.getComponent("state-store", StateStoreComponent.class);
+ assertThat(component.getBackendFactory())
+
.isSameAs(context.getBean(StateStoreBackendFactory.class));
+
+ ProducerTemplate template =
camelContext.createProducerTemplate();
+ try {
+
template.requestBodyAndHeader("state-store:orders?operation=put", "order-1",
+ StateStoreConstants.KEY, "id");
+
template.requestBodyAndHeader("state-store:users?operation=put", "user-1",
+ StateStoreConstants.KEY, "id");
+
+
assertThat(template.requestBodyAndHeader("state-store:orders?operation=get",
null,
+ StateStoreConstants.KEY,
"id")).isEqualTo("order-1");
+
assertThat(template.requestBodyAndHeader("state-store:users?operation=get",
null,
+ StateStoreConstants.KEY,
"id")).isEqualTo("user-1");
+ } finally {
+ template.stop();
+ }
+
+ CacheManager cacheManager =
context.getBean(CacheManager.class);
+
assertThat(cacheManager.getCache("orders").get("id").get()).isEqualTo("order-1");
+
assertThat(cacheManager.getCache("users").get("id").get()).isEqualTo("user-1");
+ });
+ }
+
+ @Test
+ void springFactoryAndCamelRepositoryCoexist() {
+ mixedBackendRunner
+
.withPropertyValues("camel.state-store.spring-cache.enabled=true")
+ .run(ctx -> {
+ ProducerTemplate template =
ctx.getBean(CamelContext.class).createProducerTemplate();
+ try {
+
template.sendBodyAndHeader("state-store:orders?operation=put", "o1",
+ StateStoreConstants.KEY, "1");
+
template.sendBodyAndHeader("state-store:sessions?operation=put&backend=#camelRepo",
"s1",
+ StateStoreConstants.KEY, "1");
+
+
assertThat(ctx.getBean(CacheManager.class).getCache("orders").get("1").get())
+ .isEqualTo("o1");
+ assertThat(ctx.getBean("camelRepo",
MemoryKeyValueRepository.class).get("1"))
+ .isEqualTo("s1");
+
assertThat(ctx.getBean(CacheManager.class).getCache("sessions").get("1")).isNull();
+ } finally {
+ template.stop();
+ }
+ });
+ }
+
+ @Test
+ void camelRepositoryIsUsedWhenSpringCacheIsDisabled() {
+ mixedBackendRunner.run(ctx -> {
+ ProducerTemplate template =
ctx.getBean(CamelContext.class).createProducerTemplate();
+ try {
+ template.sendBodyAndHeader("state-store:orders?operation=put",
"o1",
+ StateStoreConstants.KEY, "1");
+
+ assertThat(ctx.getBean("camelRepo",
MemoryKeyValueRepository.class).get("1")).isEqualTo("o1");
+ } finally {
+ template.stop();
+ }
+ });
+ }
+
+ @Test
+ void optimisticLockingAggregatorStartsWhenFactoryEnabled() {
+ camelRunner
+ .withBean("cacheManager", CacheManager.class,
ConcurrentMapCacheManager::new)
+ .withBean(RouteBuilder.class, this::optimisticLockingRoute)
+
.withPropertyValues("camel.state-store.spring-cache.enabled=true")
+ .run(context -> {
+ assertThat(context).hasNotFailed();
+
assertThat(context).doesNotHaveBean(KeyValueRepository.class);
+ CamelContext camelContext =
context.getBean(CamelContext.class);
+ assertThat(camelContext.getStatus().isStarted()).isTrue();
+ assertThat(camelContext.getRoute("agg")).isNotNull();
+ });
+ }
+
+ @Test
+ void completionTimeoutAggregatorStartsWhenFactoryEnabled() {
+ camelRunner
+ .withBean("cacheManager", CacheManager.class,
ConcurrentMapCacheManager::new)
+ .withBean(RouteBuilder.class, this::completionTimeoutRoute)
+
.withPropertyValues("camel.state-store.spring-cache.enabled=true")
+ .run(context -> {
+ assertThat(context).hasNotFailed();
+ CamelContext camelContext =
context.getBean(CamelContext.class);
+ assertThat(camelContext.getStatus().isStarted()).isTrue();
+ assertThat(camelContext.getRoute("timeout")).isNotNull();
+ });
+ }
+
+ @Test
+ void cacheEipTtlStillExpiresWhenFactoryEnabled() {
+ camelRunner
+ .withBean("cacheManager", CacheManager.class,
ConcurrentMapCacheManager::new)
+ .withBean(RouteBuilder.class, this::cacheEipRoute)
+
.withPropertyValues("camel.state-store.spring-cache.enabled=true")
+ .run(context -> {
+
assertThat(context).doesNotHaveBean(KeyValueRepository.class);
+ CamelContext camelContext =
context.getBean(CamelContext.class);
+ ProducerTemplate template =
camelContext.createProducerTemplate();
+ try {
+ assertThat(callCache(template,
1)).isEqualTo("computed-1");
+ assertThat(callCache(template,
2)).isEqualTo("computed-1");
+ await().atMost(5, TimeUnit.SECONDS)
+ .pollDelay(Duration.ofMillis(250))
+ .untilAsserted(() ->
assertThat(callCache(template, 3)).isEqualTo("computed-3"));
+ } finally {
+ template.stop();
+ }
+ });
+ }
+
+ private RouteBuilder optimisticLockingRoute() {
+ return new RouteBuilder() {
+ @Override
+ public void configure() {
+ from("direct:agg").routeId("agg")
+ .aggregate(header("id"), new
UseLatestAggregationStrategy())
+ .completionSize(2).optimisticLocking()
+ .to("mock:agg");
+ }
+ };
+ }
+
+ private RouteBuilder completionTimeoutRoute() {
+ return new RouteBuilder() {
+ @Override
+ public void configure() {
+ from("direct:timeout").routeId("timeout")
+ .aggregate(header("id"), new
UseLatestAggregationStrategy())
+ .completionTimeout(1000)
+ .to("mock:timeout");
+ }
+ };
+ }
+
+ private RouteBuilder cacheEipRoute() {
+ return new RouteBuilder() {
+ @Override
+ public void configure() {
+ from("direct:cached")
+ .cache(header("id"))
+ .ttl(200)
+ .setBody(simple("computed-${header.n}"))
+ .end();
+ }
+ };
+ }
+
+ private static Object callCache(ProducerTemplate template, int n) {
+ return template.requestBodyAndHeaders("direct:cached", null,
Map.of("id", "k", "n", n));
+ }
+}
diff --git
a/components-starter/camel-state-store-starter/src/test/java/org/apache/camel/component/statestore/springboot/cache/SpringCacheStateStoreRedisIT.java
b/components-starter/camel-state-store-starter/src/test/java/org/apache/camel/component/statestore/springboot/cache/SpringCacheStateStoreRedisIT.java
new file mode 100644
index 00000000000..c5c6978f0d3
--- /dev/null
+++
b/components-starter/camel-state-store-starter/src/test/java/org/apache/camel/component/statestore/springboot/cache/SpringCacheStateStoreRedisIT.java
@@ -0,0 +1,185 @@
+/*
+ * 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.statestore.springboot.cache;
+
+import java.time.Duration;
+import java.util.concurrent.TimeUnit;
+
+import org.apache.camel.CamelContext;
+import org.apache.camel.Exchange;
+import org.apache.camel.ProducerTemplate;
+import org.apache.camel.component.statestore.StateStoreBackendFactory;
+import org.apache.camel.component.statestore.StateStoreConstants;
+import
org.apache.camel.component.statestore.springboot.StateStoreComponentAutoConfiguration;
+import
org.apache.camel.component.statestore.springboot.StateStoreComponentConverter;
+import org.apache.camel.spi.KeyValueRepository;
+import org.apache.camel.spring.boot.CamelAutoConfiguration;
+import org.apache.camel.support.DefaultExchange;
+import org.apache.camel.support.DefaultExchangeHolder;
+import org.apache.camel.support.KeyValueAggregationRepository;
+import org.apache.camel.support.KeyValueIdempotentRepository;
+import org.apache.camel.test.infra.redis.services.RedisService;
+import org.apache.camel.test.infra.redis.services.RedisServiceFactory;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.condition.DisabledIfSystemProperty;
+import org.junit.jupiter.api.extension.RegisterExtension;
+import org.springframework.boot.autoconfigure.AutoConfigurations;
+import org.springframework.boot.cache.autoconfigure.CacheAutoConfiguration;
+import
org.springframework.boot.data.redis.autoconfigure.DataRedisAutoConfiguration;
+import org.springframework.boot.test.context.runner.ApplicationContextRunner;
+import org.springframework.cache.CacheManager;
+import org.springframework.cache.annotation.ProxyCachingConfiguration;
+import org.springframework.data.redis.cache.RedisCacheManager;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+import static org.awaitility.Awaitility.await;
+
+@DisabledIfSystemProperty(named = "ci.env.name", matches = "github.com",
+ disabledReason = "Disabled on GH Action due to Docker limit")
+class SpringCacheStateStoreRedisIT {
+
+ @RegisterExtension
+ static RedisService redis = RedisServiceFactory.createService();
+
+ private ApplicationContextRunner runner() {
+ return new ApplicationContextRunner()
+ .withConfiguration(AutoConfigurations.of(
+ CamelAutoConfiguration.class,
+ DataRedisAutoConfiguration.class,
+ CacheAutoConfiguration.class,
+ StateStoreComponentAutoConfiguration.class,
+ SpringCacheStateStoreAutoConfiguration.class))
+ .withUserConfiguration(ProxyCachingConfiguration.class,
StateStoreComponentConverter.class)
+ .withPropertyValues(
+ "spring.main.banner-mode=off",
+ "camel.state-store.spring-cache.enabled=true",
+ "spring.cache.type=redis",
+
"spring.cache.cache-names=orders,users,idempotent,aggregate",
+ "spring.data.redis.host=" + redis.host(),
+ "spring.data.redis.port=" + redis.port());
+ }
+
+ @Test
+ void stateStoreOperationsUseRedisCacheManager() {
+ runner().run(context -> {
+ assertThat(context).hasNotFailed();
+
assertThat(context.getBean(CacheManager.class)).isInstanceOf(RedisCacheManager.class);
+ assertThat(context).hasSingleBean(StateStoreBackendFactory.class);
+ assertThat(context).doesNotHaveBean(KeyValueRepository.class);
+
+ CamelContext camelContext = context.getBean(CamelContext.class);
+ ProducerTemplate template = camelContext.createProducerTemplate();
+ try {
+ Object previous =
template.requestBodyAndHeader("state-store:orders?operation=put", "order-1",
+ StateStoreConstants.KEY, "1");
+ assertThat(previous).isNull();
+
assertThat(template.requestBodyAndHeader("state-store:orders?operation=get",
null,
+ StateStoreConstants.KEY, "1")).isEqualTo("order-1");
+
assertThat(template.requestBodyAndHeader("state-store:orders?operation=contains",
null,
+ StateStoreConstants.KEY, "1")).isEqualTo(true);
+
+ Object existing =
template.requestBodyAndHeader("state-store:orders?operation=putIfAbsent",
"order-2",
+ StateStoreConstants.KEY, "1");
+ assertThat(existing).isEqualTo("order-1");
+
assertThat(template.requestBodyAndHeader("state-store:orders?operation=get",
null,
+ StateStoreConstants.KEY, "1")).isEqualTo("order-1");
+
+ Object deleted =
template.requestBodyAndHeader("state-store:orders?operation=delete", null,
+ StateStoreConstants.KEY, "1");
+ assertThat(deleted).isEqualTo("order-1");
+
assertThat(template.requestBodyAndHeader("state-store:orders?operation=contains",
null,
+ StateStoreConstants.KEY, "1")).isEqualTo(false);
+
+
template.requestBodyAndHeader("state-store:users?operation=put", "user-1",
+ StateStoreConstants.KEY, "1");
+
assertThat(template.requestBodyAndHeader("state-store:users?operation=get",
null,
+ StateStoreConstants.KEY, "1")).isEqualTo("user-1");
+ } finally {
+ template.stop();
+ }
+ });
+ }
+
+ @Test
+ void keysAndSizeThrowUnsupportedOperationException() {
+ runner().run(context -> {
+ StateStoreBackendFactory factory =
context.getBean(StateStoreBackendFactory.class);
+ KeyValueRepository repository = factory.createBackend("orders");
+ repository.put("k", "v");
+
+
assertThatThrownBy(repository::keys).isInstanceOf(UnsupportedOperationException.class);
+
assertThatThrownBy(repository::size).isInstanceOf(UnsupportedOperationException.class);
+ assertThat(repository.get("k")).isEqualTo("v");
+ });
+ }
+
+ @Test
+ void explicitEipAdaptersRoundTripSerializableValues() {
+ runner().run(context -> {
+ CacheManager cacheManager = context.getBean(CacheManager.class);
+ CamelContext camelContext = context.getBean(CamelContext.class);
+
+ KeyValueIdempotentRepository idempotent
+ = new KeyValueIdempotentRepository(new
SpringCacheKeyValueRepository(cacheManager, "idempotent"));
+ idempotent.start();
+ try {
+ assertThat(idempotent.add("msg-1")).isTrue();
+ assertThat(idempotent.add("msg-1")).isFalse();
+ assertThat(idempotent.contains("msg-1")).isTrue();
+ } finally {
+ idempotent.stop();
+ }
+
+ SpringCacheKeyValueRepository aggregateStore
+ = new SpringCacheKeyValueRepository(cacheManager,
"aggregate");
+ KeyValueAggregationRepository aggregationRepository = new
KeyValueAggregationRepository(aggregateStore);
+ aggregationRepository.setCamelContext(camelContext);
+ aggregationRepository.start();
+ try {
+ Exchange exchange = new DefaultExchange(camelContext);
+ exchange.getMessage().setBody("payload");
+ aggregationRepository.add(camelContext, "corr-1", exchange);
+
+ assertThat(aggregateStore.get("aggregate:corr-1"))
+ .isInstanceOf(DefaultExchangeHolder.class);
+
+ Exchange restored = aggregationRepository.get(camelContext,
"corr-1");
+ assertThat(restored).isNotNull();
+
assertThat(restored.getMessage().getBody()).isEqualTo("payload");
+ } finally {
+ aggregationRepository.stop();
+ }
+ });
+ }
+
+ @Test
+ void providerLevelTimeToLiveExpiresEntries() {
+ runner()
+ .withPropertyValues("spring.cache.redis.time-to-live=2s")
+ .run(context -> {
+ StateStoreBackendFactory factory =
context.getBean(StateStoreBackendFactory.class);
+ KeyValueRepository repository =
factory.createBackend("orders");
+ repository.put("expiring", "value");
+ assertThat(repository.get("expiring")).isEqualTo("value");
+
+ await().atMost(5, TimeUnit.SECONDS)
+ .pollInterval(Duration.ofMillis(50))
+ .untilAsserted(() ->
assertThat(repository.get("expiring")).isNull());
+ });
+ }
+}
diff --git a/core/camel-spring-boot/src/main/docs/spring-boot.adoc
b/core/camel-spring-boot/src/main/docs/spring-boot.adoc
index b52750d95f3..3a83148fca1 100644
--- a/core/camel-spring-boot/src/main/docs/spring-boot.adoc
+++ b/core/camel-spring-boot/src/main/docs/spring-boot.adoc
@@ -755,4 +755,4 @@ If you encounter issues with virtual threads:
1. **Check JDK Version**: Ensure you're running JDK 21 or later
2. **Verify Property Setting**: Confirm `spring.threads.virtual.enabled=true`
is set
3. **Review Logs**: Check for virtual thread configuration messages in startup
logs
-4. **Monitor Thread Usage**: Use JVM monitoring tools to observe virtual
thread behavior
\ No newline at end of file
+4. **Monitor Thread Usage**: Use JVM monitoring tools to observe virtual
thread behavior
diff --git a/docs/spring-boot/modules/ROOT/pages/starters/state-store.adoc
b/docs/spring-boot/modules/ROOT/pages/starters/state-store.adoc
index 53ddafa63ec..a9eeaba5253 100644
--- a/docs/spring-boot/modules/ROOT/pages/starters/state-store.adoc
+++ b/docs/spring-boot/modules/ROOT/pages/starters/state-store.adoc
@@ -21,9 +21,69 @@ Please refer to the above links for usage and configuration
details.
</dependency>
----
+== Usage
+
+=== Choosing a backend
+
+The State Store component supports Spring Cache and Camel `KeyValueRepository`
backends.
+
+==== Spring Cache
+
+Set `camel.state-store.spring-cache.enabled=true` to register a
`StateStoreBackendFactory` backed by Spring's `CacheManager`. Each state-store
name maps to a Spring cache with the same name, reusing the application's
`spring.cache.*` configuration.
+
+The factory is registered when a single `CacheManager` candidate is available
and no other `StateStoreBackendFactory` bean is defined. It does not register a
global `KeyValueRepository` bean.
+
+==== Camel KeyValueRepository
+
+Alternatively, use a Camel `KeyValueRepository` implementation, such as
`camel-redis`, `camel-caffeine`, `camel-jcache`, `camel-sql`, or
`camel-infinispan`. Define the repository as a bean and leave
`camel.state-store.spring-cache.enabled` disabled. The State Store component
automatically discovers the repository.
+
+These repositories can also be discovered by Camel features that support
`KeyValueRepository`, including the Aggregate EIP, Idempotent Consumer, and
Cache EIP.
+
+==== Configuration examples
+
+To use Spring Cache, configure a cache provider and enable the factory:
+
+[source,properties]
+----
+spring.cache.type=caffeine
+spring.cache.cache-names=orders,users
+camel.state-store.spring-cache.enabled=true
+----
+
+Spring Boot creates the `CacheManager` from `spring.cache.*` only when the
application enables caching with `@EnableCaching`; otherwise, provide your own
`CacheManager` bean. Without a single eligible `CacheManager`, the factory is
not registered, and stores may fall back to the default in-memory repository.
If `spring.cache.cache-names` specifies a fixed set of names, include every
state-store name you use. A dynamic cache manager can create caches on demand.
+
+To use a specific Camel repository, register it as a bean, for example under
the name `redisRepo`, and select it on the endpoint:
+
+[source,java]
+----
+from("direct:order")
+ .setHeader("CamelStateStoreKey", simple("${header.orderId}"))
+ .to("state-store:orders?backend=#redisRepo");
+----
+
+==== Using both backends
+
+When the Spring Cache factory is enabled, it is the default backend for
state-store endpoints unless an endpoint explicitly selects another backend or
a component-level backend factory takes precedence.
+
+You can therefore use Spring Cache for ordinary stores and explicitly select a
Camel repository for stores that need its additional capabilities.
+
+== Limitations
+
+=== Spring Cache
+
+Spring Cache provides a common API, but some operations depend on the
underlying cache provider.
+
+* **Per-entry TTL:** Spring Cache has no portable per-entry TTL API. A
positive, non-zero TTL requested through the repository is not applied by the
Spring adapter. Configure expiration using the underlying cache provider, for
example with `spring.cache.redis.time-to-live` or a Caffeine specification.
+* **Atomic operations:** Spring Cache does not provide a portable atomic
get-and-put operation. The adapter's `put` and `delete` read the previous value
and then write or evict separately, so the returned previous value is not
atomic. `putIfAbsent` delegates to the provider's Spring Cache implementation;
its atomicity depends on that provider. Conditional `replace` and conditional
`delete` may use non-atomic check-then-act operations.
+* **Key enumeration and size:** `keys()` and `size()` are supported only when
`Cache.getNativeCache()` returns a `ConcurrentMap`, such as with the simple
cache provider. They operate on a snapshot of current string keys. Other
providers, including Caffeine, Redis, and JCache, do not support these
operations through the Spring adapter.
+
+=== Camel KeyValueRepository
+
+Choose a Camel repository when you need capabilities beyond the portable
Spring Cache API, such as per-entry TTL, atomic `putIfAbsent` or `replace`,
conditional deletion, and key enumeration or size. Actual support and
guarantees depend on the selected repository implementation.
+
== Spring Boot Auto-Configuration
-The starter supports 4 options, which are listed below.
+The starter supports 5 options, which are listed below.
[width="100%",cols="2,5,^1,2",options="header"]
|===
@@ -32,4 +92,5 @@ The starter supports 4 options, which are listed below.
| camel.component.state-store.backend-factory | Factory that creates a
dedicated backend for each store name. When set, it takes precedence over
registry auto-discovery for endpoints without an explicit backend. The option
is a org.apache.camel.component.statestore.StateStoreBackendFactory type. | |
StateStoreBackendFactory
| camel.component.state-store.enabled | Whether to enable auto configuration
of the state-store component. This is enabled by default. | | Boolean
| camel.component.state-store.lazy-start-producer | Whether the producer
should be started lazy (on the first message). By starting lazy you can use
this to allow CamelContext and routes to startup in situations where a producer
may otherwise fail during starting and cause the route to fail being started.
By deferring this startup to be lazy then the startup failure can be handled
during routing messages via Camel's routing error handlers. Beware that when
the first message is processed [...]
+| camel.state-store.spring-cache.enabled | Whether to register a
StateStoreBackendFactory that creates a Spring Cache-backed KeyValueRepository
for each state-store name. Disabled by default. Requires a single CacheManager
bean and no user-defined StateStoreBackendFactory. Does not register a global
KeyValueRepository. | false | Boolean
|===
