Croway commented on PR #2046:
URL:
https://github.com/apache/camel-spring-boot/pull/2046#issuecomment-6262539043
Thanks for the work on this, @ramu11. I reviewed the PR against CAMEL-24757
and reproduced the main concerns with a test that starts a real Spring Boot
context (`CamelAutoConfiguration` + `CacheAutoConfiguration` + this
auto-configuration, real Caffeine cache, compared against a control with the
property off). All 6 scenarios below reproduce on the current head (98be59f).
### Findings
**1. Enabling the property breaks existing `optimisticLocking()`
aggregators**
The global `KeyValueRepository` bean is picked up by
`AggregateProcessor.doStart()` through `findSingleByType`, so an aggregator
with no explicit repository gets `KeyValueAggregationRepository` instead of
`MemoryAggregationRepository(optimisticLocking)`.
`KeyValueAggregationRepository` does not implement
`OptimisticLockingAggregationRepository`, so startup fails with
`IllegalArgumentException: Optimistic locking cannot be enabled ...`. With the
property off, the same route starts.
**2. `keys()` is effectively unsupported on every production provider, and
that breaks `completionTimeout` at startup**
`keys()`/`size()` work only when `getNativeCache()` is a `ConcurrentMap`,
which in practice means `spring.cache.type=simple`. With a real
`spring.cache.type=caffeine`, any `aggregate(...).completionTimeout(...)` route
without its own repository fails CamelContext startup
(`FailedToStartRouteException` caused by `UnsupportedOperationException`). The
existing tests only use a fake `NonConcurrentNativeCache`, not a real provider.
The JIRA issue says `keys()`/`size()` always throw
`UnsupportedOperationException`, and the review comment there pointed out that
casting to the native cache breaks the abstraction. Either way, please state
clearly in the docs that enumeration is not supported on Caffeine, Redis or
JCache, instead of only in the detailed "Enumeration" section.
Correction to the docs: the "Aggregator Impact" paragraph says `RecoverTask`
logs a WARN every 5s when `keys()` throws. That doesn't happen. Recovery is
never scheduled for an auto-discovered repository, because `doStart()` assigns
the field directly instead of calling `setAggregationRepository()`, so
`isRecoverableRepository` stays `false`. That's a Camel core bug outside this
PR, but the docs shouldn't describe a recovery behaviour that doesn't exist.
**3. Two `CacheManager` beans make the application fail to start**
`@ConditionalOnBean(CacheManager.class)` matches when there are several
candidates, but the injected parameter needs exactly one, so creating
`springCacheKeyValueRepository` fails with `NoUniqueBeanDefinitionException`.
`@ConditionalOnSingleCandidate(CacheManager.class)` would fix it.
**4. Cache EIP TTL is silently dropped**
With `.cache(header("id")).ttl(200)`, the default `MemoryKeyValueRepository`
expires the entry after 200ms. With this adapter, the entry is still served
after 1s and never expires (one WARN per instance). With the default `simple`
cache, that also means unbounded growth.
**5. Store names collapse into one cache, and the JIRA example doesn't work**
CAMEL-24757 proposes `storeName` → `CacheManager.getCache(storeName)`. The
PR binds everything to `camel-kvr`, so:
- `state-store:orders` key `1` is returned by `state-store:users` key `1`
(without the PR, `StateStoreComponent` keeps one repository per store name)
- the JIRA's own example (`spring.cache.cache-names: myStore`) fails every
operation with `IllegalStateException: Spring Cache 'camel-kvr' not found`
### Design question
Issues 1, 2, 4 and 5 all come from registering a single global
`KeyValueRepository` bean, which every stateful EIP auto-discovers. The JIRA
scope was the State Store component only. An alternative would be a small Camel
core change that lets `StateStoreComponent` take a per-store backend factory
(for example `Function<String, KeyValueRepository>`), which camel-spring-boot
could then set with `name -> new SpringCacheKeyValueRepository(cacheManager,
name)`. That would give the 1:1 cache mapping from the JIRA and isolate stores,
and Aggregate, Idempotent Consumer and Cache EIP would no longer be affected.
Since `camel.spring.cache.state-store.enabled` and the global bean become
public behaviour once released, it seems worth settling this before merge. What
do you think?
### Smaller points
- Since the feature targets Spring Boot 4.x / Spring Framework 7: Spring 7's
`Cache` still has no key-enumeration API, but Caffeine's native cache exposes
`asMap()` (a `ConcurrentMap`) and JCache's native cache is `Iterable`, so both
could support `keys()` behind `ClassUtils.isPresent` checks. Redis would remain
unsupported.
- A configurable cache name (for example
`camel.spring.cache.state-store.cache-name`) would help users whose
`spring.cache.cache-names` is fixed.
- `put()` and `delete()` read and then write in two separate calls, so the
returned old value isn't atomic.
- The class Javadoc uses a fully qualified `{@link
org.springframework.cache.concurrent.ConcurrentMapCache}`. The project
conventions ask for imports instead.
### Request: an integration test with Redis
The main motivation in CAMEL-24757 is reusing Spring Data Redis (Lettuce)
through `spring.cache.type=redis`, but nothing exercises a non-in-memory
provider. Could you add an IT with a real Redis (`camel-test-infra-redis` from
apache/camel provides the container) that covers at least:
- state-store `put`/`get`/`putIfAbsent`/`delete`/`contains` through
`RedisCacheManager`
- `keys`/`size` throwing `UnsupportedOperationException`
- value serialization, for example an Idempotent Consumer and an Aggregator
round-trip, since `RedisCache` uses JDK serialization by default and
`DefaultExchangeHolder`/message bodies must be serializable
- TTL behaviour with a cache-level `spring.cache.redis.time-to-live`
<details>
<summary>Reproducer:
<code>SpringCacheKeyValueRepositoryReviewReproducerTest.java</code> (6 tests,
all pass on 98be59f)</summary>
Extra test dependencies in `core/camel-spring-boot/pom.xml`:
```xml
<dependency>
<groupId>com.github.ben-manes.caffeine</groupId>
<artifactId>caffeine</artifactId>
<version>3.2.4</version>
<scope>test</scope>
</dependency>
<dependency>
<groupId>org.apache.camel</groupId>
<artifactId>camel-state-store</artifactId>
<version>${camel-version}</version>
<scope>test</scope>
</dependency>
```
```java
package org.apache.camel.spring.boot.statestore;
import java.time.Duration;
import java.util.List;
import java.util.Map;
import java.util.concurrent.CopyOnWriteArrayList;
import java.util.concurrent.TimeUnit;
import ch.qos.logback.classic.Logger;
import ch.qos.logback.classic.spi.ILoggingEvent;
import ch.qos.logback.core.AppenderBase;
import org.apache.camel.CamelContext;
import org.apache.camel.ProducerTemplate;
import org.apache.camel.builder.RouteBuilder;
import org.apache.camel.processor.aggregate.AggregateProcessor;
import org.apache.camel.processor.aggregate.GroupedBodyAggregationStrategy;
import org.apache.camel.spi.KeyValueRepository;
import org.apache.camel.spring.boot.CamelAutoConfiguration;
import org.junit.jupiter.api.Test;
import org.slf4j.LoggerFactory;
import org.springframework.boot.autoconfigure.AutoConfigurations;
import org.springframework.boot.cache.autoconfigure.CacheAutoConfiguration;
import org.springframework.boot.test.context.runner.ApplicationContextRunner;
import org.springframework.cache.CacheManager;
import org.springframework.cache.annotation.ProxyCachingConfiguration;
import org.springframework.cache.concurrent.ConcurrentMapCacheManager;
import static org.assertj.core.api.Assertions.assertThat;
import static org.assertj.core.api.Assertions.catchThrowable;
import static org.awaitility.Awaitility.await;
/**
* Review reproducers for PR #2046. Each scenario runs a real Spring Boot
context with CamelAutoConfiguration,
* CacheAutoConfiguration and the new auto-configuration. Every test also
runs a control with the property disabled.
*/
class SpringCacheKeyValueRepositoryReviewReproducerTest {
private final ApplicationContextRunner runner = new
ApplicationContextRunner()
.withConfiguration(AutoConfigurations.of(
CamelAutoConfiguration.class,
CacheAutoConfiguration.class,
SpringCacheKeyValueRepositoryAutoConfiguration.class))
.withUserConfiguration(ProxyCachingConfiguration.class);
// ---- 1. optimisticLocking aggregator breaks once the property is
enabled ----
static RouteBuilder optimisticLockingRoute() {
return new RouteBuilder() {
@Override
public void configure() {
from("direct:agg").routeId("agg")
.aggregate(header("id"), new
GroupedBodyAggregationStrategy())
.completionSize(2).optimisticLocking()
.to("mock:out");
}
};
}
@Test
void issue1OptimisticLockingAggregatorFailsToStart() {
runner.withBean(RouteBuilder.class,
SpringCacheKeyValueRepositoryReviewReproducerTest::optimisticLockingRoute)
.withPropertyValues("spring.cache.type=simple",
"spring.cache.cache-names=camel-kvr")
.withPropertyValues("camel.spring.cache.state-store.enabled=false")
.run(ctx -> assertThat(ctx).hasNotFailed());
runner.withBean(RouteBuilder.class,
SpringCacheKeyValueRepositoryReviewReproducerTest::optimisticLockingRoute)
.withPropertyValues("spring.cache.type=simple",
"spring.cache.cache-names=camel-kvr")
.withPropertyValues("camel.spring.cache.state-store.enabled=true")
.run(ctx -> {
assertThat(ctx).hasFailed();
assertThat(ctx.getStartupFailure()).rootCause()
.isInstanceOf(IllegalArgumentException.class)
.hasMessageContaining("Optimistic locking cannot
be enabled");
});
}
// ---- 2. keys() unsupported on a real Caffeine cache ----
static RouteBuilder completionTimeoutRoute() {
return new RouteBuilder() {
@Override
public void configure() {
from("direct:agg").routeId("agg")
.aggregate(header("id"), new
GroupedBodyAggregationStrategy())
.completionTimeout(1000)
.to("mock:out");
}
};
}
static RouteBuilder completionSizeRoute() {
return new RouteBuilder() {
@Override
public void configure() {
from("direct:agg").routeId("agg")
.aggregate(header("id"), new
GroupedBodyAggregationStrategy())
.completionSize(2)
.to("mock:out");
}
};
}
@Test
void issue2CaffeineKeysUnsupportedBreaksCompletionTimeoutStartup() {
runner.withBean(RouteBuilder.class,
SpringCacheKeyValueRepositoryReviewReproducerTest::completionTimeoutRoute)
.withPropertyValues("spring.cache.type=caffeine",
"spring.cache.cache-names=camel-kvr",
"camel.spring.cache.state-store.enabled=true")
.run(ctx -> {
assertThat(ctx).hasFailed();
assertThat(ctx.getStartupFailure()).rootCause()
.isInstanceOf(UnsupportedOperationException.class);
});
}
@Test
void issue2RecoveryIsNeverScheduledForAutoDiscoveredRepository() {
Logger logger = (Logger)
LoggerFactory.getLogger(AggregateProcessor.class);
CapturingAppender appender = new CapturingAppender();
appender.start();
logger.addAppender(appender);
try {
runner.withBean(RouteBuilder.class,
SpringCacheKeyValueRepositoryReviewReproducerTest::completionSizeRoute)
.withPropertyValues("spring.cache.type=caffeine",
"spring.cache.cache-names=camel-kvr",
"camel.spring.cache.state-store.enabled=true")
.run(ctx -> {
assertThat(ctx).hasNotFailed();
assertThat(appender.events).extracting(ILoggingEvent::getFormattedMessage)
.anyMatch(m -> m.startsWith("Auto-discovered
KeyValueRepository"))
.noneMatch(m -> m.contains("scheduling
recover checker"));
});
} finally {
logger.detachAppender(appender);
}
}
// ---- 3. Two CacheManager beans: condition matches, injection fails
----
@Test
void issue3TwoCacheManagersFailApplicationStartup() {
new ApplicationContextRunner()
.withConfiguration(AutoConfigurations.of(SpringCacheKeyValueRepositoryAutoConfiguration.class))
.withBean("local", CacheManager.class, () -> new
ConcurrentMapCacheManager("camel-kvr"))
.withBean("remote", CacheManager.class, () -> new
ConcurrentMapCacheManager("camel-kvr"))
.withPropertyValues("camel.spring.cache.state-store.enabled=true")
.run(ctx -> {
assertThat(ctx).hasFailed();
assertThat(ctx.getStartupFailure())
.hasStackTraceContaining("NoUniqueBeanDefinitionException")
.hasStackTraceContaining("springCacheKeyValueRepository");
});
}
// ---- 4. Cache EIP TTL silently ignored ----
static RouteBuilder cacheEipRoute() {
return new RouteBuilder() {
@Override
public void configure() {
from("direct:cached").routeId("cached")
.cache(header("id")).ttl(200)
.setBody(simple("computed-${header.n}"))
.end();
}
};
}
@Test
void issue4CacheEipTtlIgnored() {
// control: default MemoryKeyValueRepository honours the 200ms TTL
runner.withBean(RouteBuilder.class,
SpringCacheKeyValueRepositoryReviewReproducerTest::cacheEipRoute)
.withPropertyValues("spring.cache.type=simple",
"spring.cache.cache-names=camel-kvr",
"camel.spring.cache.state-store.enabled=false")
.run(ctx -> {
ProducerTemplate pt =
ctx.getBean(CamelContext.class).createProducerTemplate();
assertThat(call(pt, 1)).isEqualTo("computed-1");
await().atMost(5,
TimeUnit.SECONDS).pollDelay(Duration.ofMillis(250))
.untilAsserted(() -> assertThat(call(pt,
2)).isEqualTo("computed-2"));
});
// with the adapter the entry never expires
runner.withBean(RouteBuilder.class,
SpringCacheKeyValueRepositoryReviewReproducerTest::cacheEipRoute)
.withPropertyValues("spring.cache.type=simple",
"spring.cache.cache-names=camel-kvr",
"camel.spring.cache.state-store.enabled=true")
.run(ctx -> {
ProducerTemplate pt =
ctx.getBean(CamelContext.class).createProducerTemplate();
assertThat(call(pt, 1)).isEqualTo("computed-1");
await().during(Duration.ofSeconds(1)).atMost(3,
TimeUnit.SECONDS)
.until(() -> "computed-1".equals(call(pt, 2)));
});
}
// ---- 5. state-store store names collapse into one shared cache ----
@Test
void issue5StoreNamesShareOneCache() {
runner.withPropertyValues("spring.cache.type=simple",
"spring.cache.cache-names=camel-kvr",
"camel.spring.cache.state-store.enabled=true")
.run(ctx -> {
ProducerTemplate pt =
ctx.getBean(CamelContext.class).createProducerTemplate();
pt.sendBodyAndHeader("state-store:orders?operation=put",
"order-1", "CamelStateStoreKey", "1");
Object fromUsers =
pt.requestBodyAndHeader("state-store:users?operation=get", null,
"CamelStateStoreKey", "1");
assertThat(fromUsers).isEqualTo("order-1");
});
// JIRA's documented example: cache-names=myStore -> camel-kvr
missing -> every operation fails
runner.withPropertyValues("spring.cache.type=simple",
"spring.cache.cache-names=myStore",
"camel.spring.cache.state-store.enabled=true")
.run(ctx -> {
ProducerTemplate pt =
ctx.getBean(CamelContext.class).createProducerTemplate();
assertThat(catchThrowable(() -> pt.sendBodyAndHeader(
"state-store:myStore?operation=put", "v",
"CamelStateStoreKey", "k")))
.hasRootCauseInstanceOf(IllegalStateException.class)
.hasStackTraceContaining("'camel-kvr' not
found");
});
}
private static Object call(ProducerTemplate pt, int n) {
return pt.requestBodyAndHeaders("direct:cached", null, Map.of("id",
"k", "n", n));
}
static final class CapturingAppender extends AppenderBase<ILoggingEvent>
{
final List<ILoggingEvent> events = new CopyOnWriteArrayList<>();
@Override
protected void append(ILoggingEvent event) {
events.add(event);
}
}
}
```
</details>
_Claude Code on behalf of Federico Mariani_
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
