Claus Ibsen created CAMEL-26315:
-----------------------------------
Summary: camel-core - TransientResumeStrategy.createSimpleCache()
removes an entry when the resume action returns false, the opposite of the
ResumeAction contract, and fails with ConcurrentModificationException
Key: CAMEL-26315
URL: https://issues.apache.org/jira/browse/CAMEL-26315
Project: Camel
Issue Type: Bug
Components: camel-kafka, camel-core, camel-aws2-kinesis
Reporter: Claus Ibsen
Fix For: 4.24.0
{{ResumeAction.evalEntry}} documents its return value as "true if the entry
addressed should be invalidated or false otherwise", and
{{ResumeCache.forEach}} applies the action to every entry. The two
{{ResumeCache}} implementations do opposite things with that return value:
* camel-caffeine {{CaffeineCache.forEach}} follows the contract: it invalidates
the entry when the action returns {{true}}.
* {{TransientResumeStrategy.createSimpleCache()}} (camel-core-processor)
removes the entry when the action returns {{false}}, and it removes it from a
plain {{HashMap}} while it is iterating over {{entrySet()}}:
{code:java}
public void forEach(BiFunction<? super Object, ? super Object, Boolean> action)
{
for (Map.Entry<Object, Object> e : cache.entrySet()) {
if (!action.apply(e.getKey(), e.getValue())) {
cache.remove(e.getKey());
}
}
}
{code}
The resume actions each work with only one of the two caches:
* camel-aws2-kinesis {{KinesisResumeAction.evalEntry}} always returns {{false}}
("keep the entry"). With the simple cache, the first shard that resumes removes
the entries of every other shard. With two or more entries, a
{{ConcurrentModificationException}} is more likely, because the entry is
removed during the iteration. {{KinesisConsumerResumeIT}}, the only resume
example for the component, uses
{{TransientResumeStrategy.createSimpleCache()}}. CAMEL-25942 (PR #28461) made
each shard resume from its own entry, but it relies on the cache keeping the
other entries.
* camel-kafka {{KafkaResumeAdapter.resume}} returns {{true}} after
{{consumer.seek(...)}}. With the simple cache the entry is kept. With
{{CaffeineCache}}, which follows the contract, every entry is invalidated after
the first resume.
The simple cache is also used by the resume ITs of camel-cassandraql,
camel-couchbase and camel-kafka, and by the file consumer resume tests in
camel-core.
h4. Proposed change
# Make {{createSimpleCache().forEach}} follow the documented contract (remove
on {{true}}). Remove through the iterator ({{Iterator.remove()}}), or use
{{entrySet().removeIf(...)}}, so the iteration cannot fail with a
{{ConcurrentModificationException}}. A {{ConcurrentHashMap}} may also be worth
it, since consumers of parallel shards or partitions use one cache.
# Check every {{evalEntry}} / {{forEach}} callback against the contract.
{{KafkaResumeAdapter.resume}} should most likely return {{false}} (keep the
offset), and {{KinesisResumeAction}} already returns {{false}}.
# Add a unit test for {{createSimpleCache().forEach}} with two or more entries,
for both return values.
# Changing the simple cache's behaviour affects custom {{ResumeAction}}s
written against it, so it needs an upgrade-guide note.
--
This message was sent by Atlassian Jira
(v10.3.26#10030026)
