[
https://issues.apache.org/jira/browse/CAMEL-26315?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=18132368#comment-18132368
]
Guillaume Nodet commented on CAMEL-26315:
-----------------------------------------
This issue is being investigated by a coding agent (on behalf of gnodet).
*Bug confirmed in code:*
1. TransientResumeStrategy.createSimpleCache().forEach() removes an entry when
the action returns false, but ResumeAction.evalEntry() Javadoc specifies that
false means "keep the entry" (only true should trigger invalidation). The logic
is exactly inverted compared to the documented contract.
2. CaffeineCache.forEach() correctly removes on true -- confirming
createSimpleCache() has the boolean sense backwards.
3. Iterating over cache.entrySet() while calling cache.remove() inside the loop
throws ConcurrentModificationException with 2 or more entries.
4. KinesisResumeAction.evalEntry() always returns false (semantic: "keep the
entry"), but with createSimpleCache() this removes the entry -- breaking
multi-shard resume.
5. KafkaResumeAdapter.resume() returns true after consumer.seek(). With
CaffeineCache (correct contract) this means every Kafka offset entry is
invalidated after the first resume -- also broken.
*Proposed fix:*
- Invert the boolean in createSimpleCache().forEach(): remove on true (not
false)
- Use entrySet().removeIf(...) or an Iterator to avoid
ConcurrentModificationException
- Fix KafkaResumeAdapter.resume() to return false (offset applied, keep the
entry)
- Add a unit test for createSimpleCache().forEach() with 2+ entries
_Note: This comment was generated by an AI coding agent and requires manual
verification._
> 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-aws2-kinesis, camel-core, camel-kafka
> Reporter: Claus Ibsen
> Assignee: Guillaume Nodet
> Priority: Major
> 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)
