shashank created CAMEL-24954:
--------------------------------
Summary: 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
{{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)
