[
https://issues.apache.org/jira/browse/CAMEL-26315?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=18132376#comment-18132376
]
Guillaume Nodet commented on CAMEL-26315:
-----------------------------------------
A fix has been opened in PR https://github.com/apache/camel/pull/28521
Two bugs fixed in `TransientResumeStrategy.createSimpleCache().forEach()`:
1. Inverted boolean: entries were removed when the action returned `false`, but
the `ResumeCache.forEach` contract (and `CaffeineCache` reference
implementation) specify *true = invalidate, false = keep*. Fixed by removing
the negation.
2. `ConcurrentModificationException` with 2+ entries: `cache.remove(key)` was
called during a for-each iteration. Fixed by using an `Iterator` and calling
`it.remove()`.
Also fixed `KafkaResumeAdapter.resume()` which returned `true` after
`consumer.seek()`, causing `CaffeineCache` to evict all Kafka partition offsets
immediately after each resume call. Fixed by returning `false`.
New tests added:
- `TransientResumeStrategyCacheTest` — 5 unit tests for the fixed
`createSimpleCache().forEach()` behaviour including the multi-entry
ConcurrentModificationException scenario.
- `KafkaResumeAdapterTest.testResumeReturnsFalseToKeepCacheEntries` — verifies
`consumer.seek()` is called and cache entries are retained after `resume()`.
_Hermes Agent (forgebot) on behalf of gnodet_
> 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)
