[
https://issues.apache.org/jira/browse/CAMEL-24954?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=18118641#comment-18118641
]
shashank commented on CAMEL-24954:
----------------------------------
Update: the PR (https://github.com/apache/camel/pull/26815) also includes the
Cache EIP change. {{CacheProcessor.doStop()}} no longer stops its
{{KeyValueRepository}}, and it is still stopped in {{doShutdown}}. The
regression test now has 8 tests, covering the Cache EIP both when its route is
stopped and when it is removed. The Cache EIP is new in 4.23, so it needs no
upgrade-guide entry.
Note on history: for the Idempotent Consumer, this restores the {{doStop()}}
from before CAMEL-9569. CAMEL-9569 added the repository stop there together
with the actual leak fix in {{doShutdown()}}, and the leak fix is kept.
_Claude Code on behalf of allthingssecurity_
> Stopping a route with an Idempotent Consumer stops and clears a repository
> that other running routes still use
> --------------------------------------------------------------------------------------------------------------
>
> Key: CAMEL-24954
> URL: https://issues.apache.org/jira/browse/CAMEL-24954
> Project: Camel
> Issue Type: Bug
> Components: camel-core
> Reporter: shashank
> Priority: Minor
>
> {{IdempotentConsumer.doStop()}} stops the idempotent repository
> ({{ServiceHelper.stopService(processor, idempotentRepository)}}). The
> in-memory repositories clear their data when stopped:
> * {{MemoryIdempotentRepository.doStop()}} runs {{cache.clear()}}.
> * {{KeyValueIdempotentRepository.doStop()}} stops the underlying
> {{KeyValueRepository}}, and {{MemoryKeyValueRepository.doStop()}} runs
> {{store.clear()}}. That clears every entry in the store, not only the ones
> with the {{idempotent:}} prefix.
> A repository is often shared: the same bean is referenced from several
> routes. Since 4.23 it is also shared by default. When no repository is
> configured, {{IdempotentConsumerReifier}} wraps the single
> {{KeyValueRepository}} bean from the registry, and {{AggregateProcessor}} and
> the Cache EIP do the same. So stopping one route silently turns off duplicate
> detection for all the other routes, which are still running. With a shared
> {{KeyValueRepository}}, it also throws away the aggregator's in-progress
> groups.
> Reproducer:
> {code:java}
> from("direct:a").routeId("a").idempotentConsumer(header("id")).idempotentRepository(repo).process(e
> -> count.incrementAndGet());
> from("direct:b").routeId("b").idempotentConsumer(header("id")).idempotentRepository(repo).log("b");
> {code}
> Send id=1 to {{direct:a}} twice (count=1), then call
> {{context.getRouteController().stopRoute("b")}}, then send id=1 to
> {{direct:a}} again. Route a is still started, but {{repo.contains("1")}} is
> now false and the message is processed again (count=2). Results:
> * {{MemoryIdempotentRepository}}: count=2.
> * {{new KeyValueIdempotentRepository()}}: count=2.
> * No repository configured and a {{MemoryKeyValueRepository}} bean in the
> registry (auto-discovered by both routes): count=2.
> * {{FileIdempotentRepository}}: count=1, because {{add}} falls back to the
> file store.
> With a {{KeyValueIdempotentRepository}} and a
> {{KeyValueAggregationRepository}} over the same {{MemoryKeyValueRepository}},
> the store contains {{[aggregate:g1, idempotent:1]}} before {{stopRoute}} on
> the idempotent route and {{[]}} after it. The next message for group g1 then
> starts a new group, so the first message is lost from the aggregate.
> The Cache EIP (new in 4.23) has the same pattern: {{CacheProcessor.doStop()}}
> stops its {{KeyValueRepository}}. With one {{MemoryKeyValueRepository}} bean
> auto-discovered by an idempotent consumer and a Cache EIP, {{stopRoute}} on
> the cache route clears the store ({{[idempotent:1, K]}} before, {{[]}}
> after), and the idempotent route processes id 1 again.
> A single route that is stopped and started also forgets all ids it has seen,
> because its in-memory repository was cleared by the stop.
> Other code already handles this:
> * {{GenericFileEndpoint.doStart}} adds its idempotent repository as a context
> service ("idempotent repository may be used by others, so add it as a service
> so its stopped when CamelContext stops"), and its {{doStop}} does not stop
> the repository.
> * {{AggregateProcessor}} stops its aggregation repository only in
> {{doShutdown}}.
> Proposed fix: {{IdempotentConsumer.doStop()}} and {{CacheProcessor.doStop()}}
> no longer stop the repository. {{doStart}} and {{doShutdown}} are unchanged,
> so the repository is still stopped and shut down when the route is removed or
> the CamelContext stops (CAMEL-9569), like {{AggregateProcessor}} does.
> Behaviour change for the 4.23 upgrade guide: a route that is stopped and
> started again keeps the ids in its in-memory repository, and a repository
> backed by a remote store stays started while the route is stopped
> ({{IdempotentRepository.clear()}} or the {{clear}} JMX operation forgets the
> ids). A PR with the fix follows, with regression tests.
> Not covered by the fix (possible follow-ups):
> * Removing a route (rather than stopping it) still shuts down the repository,
> so removing one of several routes that share an in-memory repository still
> clears it for the others. This is the existing CAMEL-9569 trade-off; fixing
> it needs reference counting or an ownership flag.
> * {{KeyValueIdempotentRepository.doStop()}} (and
> {{KeyValueAggregationRepository}}) stops the {{KeyValueRepository}} that was
> passed in, although the adapter does not own it.
> Found with a TLA+ model of the Idempotent Consumer EIP, then reproduced
> against 4.23.0-SNAPSHOT.
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
