This is an automated email from the ASF dual-hosted git repository.
gnodet pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/camel.git
The following commit(s) were added to refs/heads/main by this push:
new ccbbcde2ea47 CAMEL-26315: Fix
TransientResumeStrategy.createSimpleCache().forEach() inverted boolean and
ConcurrentModificationException; fix KafkaResumeAdapter.resume() return value
(#28521)
ccbbcde2ea47 is described below
commit ccbbcde2ea474a159aa06cccc770d7faa43ca658
Author: Guillaume Nodet <[email protected]>
AuthorDate: Mon Nov 2 15:20:56 2026 +0100
CAMEL-26315: Fix TransientResumeStrategy.createSimpleCache().forEach()
inverted boolean and ConcurrentModificationException; fix
KafkaResumeAdapter.resume() return value (#28521)
* CAMEL-26315: Fix TransientResumeStrategy.createSimpleCache() forEach()
inverted boolean and ConcurrentModificationException, and
KafkaResumeAdapter.resume() wrong return value
Two bugs in TransientResumeStrategy.createSimpleCache().forEach():
1. The boolean sense was inverted: the method removed entries when the
action returned false, but the ResumeCache contract (and CaffeineCache
reference impl) specify true=invalidate/remove, false=keep. Fix by
removing the negation.
2. cache.remove() was called during an entrySet() iteration, causing a
ConcurrentModificationException with 2+ entries. Fix by using an
Iterator and calling it.remove() instead.
KafkaResumeAdapter.resume() returned true after calling consumer.seek(),
which caused CaffeineCache (whose forEach correctly implements the
true=invalidate contract) to evict all Kafka partition offsets immediately
after the first resume call. Fix by returning false (keep the entry).
Also fix the TestResumeCache.forEach() stub in KafkaResumeAdapterTest,
which had the same inverted-boolean defect, and add:
- TransientResumeStrategyCacheTest: 5 unit tests covering the fixed
forEach() contract with 1 and 3+ entries (no
ConcurrentModificationException,
correct remove-on-true / keep-on-false semantics, selective removal).
- KafkaResumeAdapterTest.testResumeReturnsFalseToKeepCacheEntries: verifies
consumer.seek() is called and the cache entry is retained after resume.
* CAMEL-26315: Fix Splitter.readFromStrategyCache() invoking invalid
forEach contract, update SplitterTestResumeStrategy, document
ResumeCache.forEach contract, use ConcurrentHashMap, add upgrade guide
* CAMEL-26315: Guard KafkaResumeAdapter.seek() against unassigned
partitions; fix InMemoryStringResumeCache and KinesisConsumerPositionTest
forEach contract
* CAMEL-26315: Add Splitter regression test for forEach contract; fix
upgrade guide wording; improve ResumeCache.forEach Javadoc
* CAMEL-26315: Fix Javadoc line wrapping in ResumeCache.forEach (formatter
check)
---
.../aws2/kinesis/KinesisConsumerPositionTest.java | 7 +-
.../support/resume/KafkaResumeAdapter.java | 8 +-
.../support/resume/KafkaResumeAdapterTest.java | 61 +++++++++++-
.../mongodb/support/InMemoryStringResumeCache.java | 9 +-
.../org/apache/camel/resume/cache/ResumeCache.java | 11 ++-
core/camel-core-processor/pom.xml | 7 ++
.../java/org/apache/camel/processor/Splitter.java | 13 +--
.../processor/resume/TransientResumeStrategy.java | 13 ++-
.../resume/TransientResumeStrategyCacheTest.java | 104 +++++++++++++++++++++
.../processor/SplitterTestResumeStrategy.java | 9 +-
.../camel/processor/SplitterWatermarkTest.java | 19 ++++
.../ROOT/pages/camel-4x-upgrade-guide-4_24.adoc | 17 ++++
12 files changed, 252 insertions(+), 26 deletions(-)
diff --git
a/components/camel-aws/camel-aws2-kinesis/src/test/java/org/apache/camel/component/aws2/kinesis/KinesisConsumerPositionTest.java
b/components/camel-aws/camel-aws2-kinesis/src/test/java/org/apache/camel/component/aws2/kinesis/KinesisConsumerPositionTest.java
index ab5f299e61f8..1052c7c52a0b 100644
---
a/components/camel-aws/camel-aws2-kinesis/src/test/java/org/apache/camel/component/aws2/kinesis/KinesisConsumerPositionTest.java
+++
b/components/camel-aws/camel-aws2-kinesis/src/test/java/org/apache/camel/component/aws2/kinesis/KinesisConsumerPositionTest.java
@@ -21,6 +21,7 @@ import java.util.ArrayList;
import java.util.Collections;
import java.util.HashMap;
import java.util.HashSet;
+import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.Set;
@@ -400,9 +401,11 @@ class KinesisConsumerPositionTest {
@Override
public void forEach(BiFunction<? super Object, ? super Object,
Boolean> action) {
- for (Map.Entry<Object, Object> entry : entries.entrySet()) {
+ Iterator<Map.Entry<Object, Object>> it =
entries.entrySet().iterator();
+ while (it.hasNext()) {
+ Map.Entry<Object, Object> entry = it.next();
if (action.apply(entry.getKey(), entry.getValue())) {
- entries.remove(entry.getKey());
+ it.remove();
}
}
}
diff --git
a/components/camel-kafka/src/main/java/org/apache/camel/component/kafka/consumer/support/resume/KafkaResumeAdapter.java
b/components/camel-kafka/src/main/java/org/apache/camel/component/kafka/consumer/support/resume/KafkaResumeAdapter.java
index 7825ecd45d37..0757c63c634d 100644
---
a/components/camel-kafka/src/main/java/org/apache/camel/component/kafka/consumer/support/resume/KafkaResumeAdapter.java
+++
b/components/camel-kafka/src/main/java/org/apache/camel/component/kafka/consumer/support/resume/KafkaResumeAdapter.java
@@ -39,9 +39,13 @@ public class KafkaResumeAdapter implements ResumeAdapter,
Deserializable, Cachea
private ResumeCache<TopicPartition> resumeCache;
private boolean resume(TopicPartition topicPartition, Object value) {
- consumer.seek(topicPartition, (Long) value);
+ if (consumer.assignment().contains(topicPartition)) {
+ consumer.seek(topicPartition, (Long) value);
+ } else {
+ LOG.debug("Skipping seek for {} as it is not in the current
consumer assignment", topicPartition);
+ }
- return true;
+ return false;
}
@Override
diff --git
a/components/camel-kafka/src/test/java/org/apache/camel/component/kafka/consumer/support/resume/KafkaResumeAdapterTest.java
b/components/camel-kafka/src/test/java/org/apache/camel/component/kafka/consumer/support/resume/KafkaResumeAdapterTest.java
index d74ab5ed5be8..40732025b470 100644
---
a/components/camel-kafka/src/test/java/org/apache/camel/component/kafka/consumer/support/resume/KafkaResumeAdapterTest.java
+++
b/components/camel-kafka/src/test/java/org/apache/camel/component/kafka/consumer/support/resume/KafkaResumeAdapterTest.java
@@ -18,20 +18,28 @@ package
org.apache.camel.component.kafka.consumer.support.resume;
import java.nio.ByteBuffer;
import java.util.HashMap;
+import java.util.Iterator;
import java.util.Map;
+import java.util.Set;
import java.util.function.BiFunction;
import java.util.function.Function;
import org.apache.camel.resume.Serializable;
import org.apache.camel.resume.cache.ResumeCache;
+import org.apache.kafka.clients.consumer.Consumer;
import org.apache.kafka.common.TopicPartition;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertNotNull;
import static org.junit.jupiter.api.Assertions.assertNull;
import static org.junit.jupiter.api.Assertions.assertTrue;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.never;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
class KafkaResumeAdapterTest {
@@ -99,6 +107,51 @@ class KafkaResumeAdapterTest {
assertNull(cache.get(new TopicPartition("myTopic", 0)));
}
+ @Test
+ @SuppressWarnings("unchecked")
+ void testResumeReturnsFalseToKeepCacheEntries() {
+ // KafkaResumeAdapter.resume() must return false so that the cache
entry is retained
+ // (true = invalidate according to the ResumeCache.forEach contract).
+ Consumer<?, ?> consumer = mock(Consumer.class);
+ adapter.setConsumer(consumer);
+
+ TopicPartition tp = new TopicPartition("myTopic", 0);
+ cache.add(tp, 42L);
+
+ // Stub assignment so the seek guard passes
+ when(consumer.assignment()).thenReturn(Set.of(tp));
+
+ adapter.resume();
+
+ // consumer.seek must have been called with the stored offset
+ verify(consumer).seek(tp, 42L);
+ // entry must remain in the cache after resume (action returned false
→ keep)
+ assertNotNull(cache.get(tp), "Cache entry should be retained after
resume");
+ assertEquals(42L, cache.get(tp));
+ }
+
+ @Test
+ @SuppressWarnings("unchecked")
+ void testResumeSkipsSeekForUnassignedPartition() {
+ // KafkaResumeAdapter.resume() must skip seek() if the partition is
not in the current assignment
+ // to avoid IllegalStateException from the Kafka consumer.
+ Consumer<?, ?> consumer = mock(Consumer.class);
+ adapter.setConsumer(consumer);
+
+ TopicPartition tp = new TopicPartition("myTopic", 0);
+ cache.add(tp, 42L);
+
+ // Partition not in assignment (e.g. it was revoked)
+ when(consumer.assignment()).thenReturn(Set.of());
+
+ adapter.resume();
+
+ // consumer.seek must NOT have been called
+ verify(consumer, never()).seek(tp, 42L);
+ // entry must remain in the cache
+ assertNotNull(cache.get(tp), "Cache entry should be retained even when
seek is skipped");
+ }
+
private ByteBuffer serializeString(String value) {
byte[] data = value.getBytes();
ByteBuffer buffer = ByteBuffer.allocate(Integer.BYTES + data.length);
@@ -162,9 +215,11 @@ class KafkaResumeAdapterTest {
@Override
public void forEach(BiFunction<? super TopicPartition, ? super Object,
Boolean> action) {
- for (Map.Entry<TopicPartition, Object> entry : store.entrySet()) {
- if (!action.apply(entry.getKey(), entry.getValue())) {
- break;
+ Iterator<Map.Entry<TopicPartition, Object>> it =
store.entrySet().iterator();
+ while (it.hasNext()) {
+ Map.Entry<TopicPartition, Object> entry = it.next();
+ if (action.apply(entry.getKey(), entry.getValue())) {
+ it.remove();
}
}
}
diff --git
a/components/camel-mongodb/src/test/java/org/apache/camel/component/mongodb/support/InMemoryStringResumeCache.java
b/components/camel-mongodb/src/test/java/org/apache/camel/component/mongodb/support/InMemoryStringResumeCache.java
index c113c26bf6e9..06b4ca69ca9d 100644
---
a/components/camel-mongodb/src/test/java/org/apache/camel/component/mongodb/support/InMemoryStringResumeCache.java
+++
b/components/camel-mongodb/src/test/java/org/apache/camel/component/mongodb/support/InMemoryStringResumeCache.java
@@ -17,6 +17,7 @@
package org.apache.camel.component.mongodb.support;
import java.util.HashMap;
+import java.util.Iterator;
import java.util.Map;
import java.util.Objects;
import java.util.function.BiFunction;
@@ -69,9 +70,11 @@ public class InMemoryStringResumeCache implements
ResumeCache<String> {
@Override
public void forEach(BiFunction<? super String, ? super Object, Boolean>
action) {
- for (Map.Entry<String, Object> entry : data.entrySet()) {
- if (!action.apply(entry.getKey(), entry.getValue())) {
- break;
+ Iterator<Map.Entry<String, Object>> it = data.entrySet().iterator();
+ while (it.hasNext()) {
+ Map.Entry<String, Object> entry = it.next();
+ if (action.apply(entry.getKey(), entry.getValue())) {
+ it.remove();
}
}
}
diff --git
a/core/camel-api/src/main/java/org/apache/camel/resume/cache/ResumeCache.java
b/core/camel-api/src/main/java/org/apache/camel/resume/cache/ResumeCache.java
index 8df0be0330e6..076dd3b04a03 100644
---
a/core/camel-api/src/main/java/org/apache/camel/resume/cache/ResumeCache.java
+++
b/core/camel-api/src/main/java/org/apache/camel/resume/cache/ResumeCache.java
@@ -98,9 +98,16 @@ public interface ResumeCache<K> {
Object get(K key);
/**
- * Performs the given action for each member of the cache
+ * Performs the given action for each member of the cache.
+ * <p>
+ * The {@code action} function controls whether an entry is invalidated
(removed) after being visited: returning
+ * {@code true} causes the entry to be removed from the cache; returning
{@code false} keeps it. Implementations
+ * must support removing the visited entry during the iteration (for
example with
+ * {@link java.util.Iterator#remove()} or a concurrent map), and must
visit every entry regardless of the action's
+ * return value.
*
- * @param action the action to execute
+ * @param action the action to execute; return {@code true} to
invalidate/remove the entry, {@code false} to keep
+ * it. See also {@link
org.apache.camel.resume.ResumeAction#evalEntry}.
*/
void forEach(BiFunction<? super K, ? super Object, Boolean> action);
}
diff --git a/core/camel-core-processor/pom.xml
b/core/camel-core-processor/pom.xml
index 33256098d684..a2dac6e1dad0 100644
--- a/core/camel-core-processor/pom.xml
+++ b/core/camel-core-processor/pom.xml
@@ -61,5 +61,12 @@
<groupId>org.slf4j</groupId>
<artifactId>slf4j-api</artifactId>
</dependency>
+
+ <!-- test dependencies -->
+ <dependency>
+ <groupId>org.junit.jupiter</groupId>
+ <artifactId>junit-jupiter</artifactId>
+ <scope>test</scope>
+ </dependency>
</dependencies>
</project>
diff --git
a/core/camel-core-processor/src/main/java/org/apache/camel/processor/Splitter.java
b/core/camel-core-processor/src/main/java/org/apache/camel/processor/Splitter.java
index b13ffd8aed62..d3be7d02d801 100644
---
a/core/camel-core-processor/src/main/java/org/apache/camel/processor/Splitter.java
+++
b/core/camel-core-processor/src/main/java/org/apache/camel/processor/Splitter.java
@@ -161,13 +161,14 @@ public class Splitter extends MulticastProcessor {
if (cache != null) {
String[] result = new String[1];
cache.forEach((key, value) -> {
- String keyStr = key instanceof OffsetKey<?> ok
- ? String.valueOf(ok.getValue()) :
String.valueOf(key);
- if (watermarkKey.equals(keyStr) && value != null) {
- result[0] = toWatermarkString(value);
- return false;
+ if (result[0] == null && value != null) {
+ String keyStr = key instanceof OffsetKey<?> ok
+ ? String.valueOf(ok.getValue()) :
String.valueOf(key);
+ if (watermarkKey.equals(keyStr)) {
+ result[0] = toWatermarkString(value);
+ }
}
- return true;
+ return false; // never invalidate: true means remove
the entry
});
return result[0];
}
diff --git
a/core/camel-core-processor/src/main/java/org/apache/camel/processor/resume/TransientResumeStrategy.java
b/core/camel-core-processor/src/main/java/org/apache/camel/processor/resume/TransientResumeStrategy.java
index 87cb7ed80509..3645ae358f9f 100644
---
a/core/camel-core-processor/src/main/java/org/apache/camel/processor/resume/TransientResumeStrategy.java
+++
b/core/camel-core-processor/src/main/java/org/apache/camel/processor/resume/TransientResumeStrategy.java
@@ -17,9 +17,10 @@
package org.apache.camel.processor.resume;
-import java.util.HashMap;
+import java.util.Iterator;
import java.util.Map;
import java.util.Objects;
+import java.util.concurrent.ConcurrentHashMap;
import java.util.function.BiFunction;
import java.util.function.Function;
@@ -133,7 +134,7 @@ public class TransientResumeStrategy implements
ResumeStrategy {
public static ResumeCache<Object> createSimpleCache() {
return new ResumeCache<>() {
- private final Map<Object, Object> cache = new HashMap<>();
+ private final Map<Object, Object> cache = new
ConcurrentHashMap<>();
@Override
public Object computeIfAbsent(Object key, Function<? super Object,
? super Object> mapping) {
@@ -179,9 +180,11 @@ public class TransientResumeStrategy implements
ResumeStrategy {
@Override
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());
+ Iterator<Map.Entry<Object, Object>> it =
cache.entrySet().iterator();
+ while (it.hasNext()) {
+ Map.Entry<Object, Object> e = it.next();
+ if (action.apply(e.getKey(), e.getValue())) {
+ it.remove();
}
}
}
diff --git
a/core/camel-core-processor/src/test/java/org/apache/camel/processor/resume/TransientResumeStrategyCacheTest.java
b/core/camel-core-processor/src/test/java/org/apache/camel/processor/resume/TransientResumeStrategyCacheTest.java
new file mode 100644
index 000000000000..e9f38ef7402f
--- /dev/null
+++
b/core/camel-core-processor/src/test/java/org/apache/camel/processor/resume/TransientResumeStrategyCacheTest.java
@@ -0,0 +1,104 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.camel.processor.resume;
+
+import java.util.ArrayList;
+import java.util.List;
+
+import org.apache.camel.resume.cache.ResumeCache;
+import org.junit.jupiter.api.Test;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.junit.jupiter.api.Assertions.assertNull;
+
+/**
+ * Unit tests for {@link TransientResumeStrategy#createSimpleCache()}.
+ */
+class TransientResumeStrategyCacheTest {
+
+ @Test
+ void testForEachKeepsEntryWhenActionReturnsFalse() {
+ ResumeCache<Object> cache =
TransientResumeStrategy.createSimpleCache();
+ cache.add("key1", "value1");
+
+ // action returns false → keep entry
+ cache.forEach((k, v) -> false);
+
+ assertNotNull(cache.get("key1"), "Entry should be retained when action
returns false");
+ assertEquals("value1", cache.get("key1"));
+ }
+
+ @Test
+ void testForEachRemovesEntryWhenActionReturnsTrue() {
+ ResumeCache<Object> cache =
TransientResumeStrategy.createSimpleCache();
+ cache.add("key1", "value1");
+
+ // action returns true → invalidate/remove entry
+ cache.forEach((k, v) -> true);
+
+ assertNull(cache.get("key1"), "Entry should be removed when action
returns true");
+ }
+
+ @Test
+ void testForEachWithMultipleEntriesNoException() {
+ ResumeCache<Object> cache =
TransientResumeStrategy.createSimpleCache();
+ cache.add("key1", "value1");
+ cache.add("key2", "value2");
+ cache.add("key3", "value3");
+
+ // This must not throw ConcurrentModificationException
+ List<Object> visited = new ArrayList<>();
+ cache.forEach((k, v) -> {
+ visited.add(k);
+ return false; // keep all entries
+ });
+
+ assertEquals(3, visited.size(), "All entries should be visited");
+ assertNotNull(cache.get("key1"), "key1 should be retained");
+ assertNotNull(cache.get("key2"), "key2 should be retained");
+ assertNotNull(cache.get("key3"), "key3 should be retained");
+ }
+
+ @Test
+ void testForEachWithMultipleEntriesRemovesAllOnTrue() {
+ ResumeCache<Object> cache =
TransientResumeStrategy.createSimpleCache();
+ cache.add("key1", "value1");
+ cache.add("key2", "value2");
+ cache.add("key3", "value3");
+
+ // action returns true → all entries invalidated; must not throw
ConcurrentModificationException
+ cache.forEach((k, v) -> true);
+
+ assertNull(cache.get("key1"), "key1 should be removed");
+ assertNull(cache.get("key2"), "key2 should be removed");
+ assertNull(cache.get("key3"), "key3 should be removed");
+ }
+
+ @Test
+ void testForEachSelectiveRemoval() {
+ ResumeCache<Object> cache =
TransientResumeStrategy.createSimpleCache();
+ cache.add("keep", "v1");
+ cache.add("remove", "v2");
+
+ // invalidate only the "remove" key
+ cache.forEach((k, v) -> "remove".equals(k));
+
+ assertNotNull(cache.get("keep"), "keep entry should be retained");
+ assertNull(cache.get("remove"), "remove entry should be deleted");
+ }
+}
diff --git
a/core/camel-core/src/test/java/org/apache/camel/processor/SplitterTestResumeStrategy.java
b/core/camel-core/src/test/java/org/apache/camel/processor/SplitterTestResumeStrategy.java
index 5771d8ca9185..c9a94a17f518 100644
---
a/core/camel-core/src/test/java/org/apache/camel/processor/SplitterTestResumeStrategy.java
+++
b/core/camel-core/src/test/java/org/apache/camel/processor/SplitterTestResumeStrategy.java
@@ -16,6 +16,7 @@
*/
package org.apache.camel.processor;
+import java.util.Iterator;
import java.util.Map;
import java.util.function.BiFunction;
import java.util.function.Function;
@@ -163,9 +164,11 @@ class SplitterTestResumeStrategy implements ResumeStrategy
{
@Override
public void forEach(BiFunction<? super Object, ? super Object,
Boolean> action) {
- for (Map.Entry<String, String> e : store.entrySet()) {
- if (!action.apply(e.getKey(), e.getValue())) {
- break;
+ Iterator<Map.Entry<String, String>> it =
store.entrySet().iterator();
+ while (it.hasNext()) {
+ Map.Entry<String, String> e = it.next();
+ if (action.apply(e.getKey(), e.getValue())) {
+ it.remove(); // true = invalidate/remove the entry
}
}
}
diff --git
a/core/camel-core/src/test/java/org/apache/camel/processor/SplitterWatermarkTest.java
b/core/camel-core/src/test/java/org/apache/camel/processor/SplitterWatermarkTest.java
index d7f7947229bc..7ccc454f3cf2 100644
---
a/core/camel-core/src/test/java/org/apache/camel/processor/SplitterWatermarkTest.java
+++
b/core/camel-core/src/test/java/org/apache/camel/processor/SplitterWatermarkTest.java
@@ -286,6 +286,25 @@ class SplitterWatermarkTest extends ContextTestSupport {
"Watermark should not be updated when processing is aborted");
}
+ @Test
+ void testWatermarkReadKeepsOtherKeys() throws Exception {
+ // CAMEL-26315: reading the watermark for "testJob" via
readFromStrategyCache must NOT
+ // evict watermarks stored under other keys (regression guard for the
Splitter.readFromStrategyCache fix)
+ store.put("otherJob", "7");
+ store.put("testJob", "2");
+
+ MockEndpoint mock = getMockEndpoint("mock:split");
+ mock.expectedMessageCount(2); // only items 3 and 4 (watermark was at
"2")
+
+ template.sendBody("direct:index", Arrays.asList("a", "b", "c", "d",
"e"));
+
+ mock.assertIsSatisfied();
+
+ assertEquals("4", store.get("testJob"), "testJob watermark should be
updated to last index");
+ // reading the testJob watermark must not evict the watermark of
another key
+ assertEquals("7", store.get("otherJob"), "otherJob watermark must
survive a readFromStrategyCache call for testJob");
+ }
+
@Test
void testIndexBasedWatermarkNoUpdateOnHandledAbort() throws Exception {
// CAMEL-24139: streaming split + stopOnException + handled(true)
should NOT
diff --git
a/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_24.adoc
b/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_24.adoc
index 08b16636c546..3fc0178f4c68 100644
--- a/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_24.adoc
+++ b/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_24.adoc
@@ -4379,6 +4379,23 @@ Use `0.0.0.0` as the host to accept connections on all
network interfaces.
`AvroRpcHttpServerFactory` has a new method `create(SpecificResponder, String,
int)` that receives the host. A custom factory should implement it.
+=== camel-core - the simple resume cache follows the ResumeCache.forEach
contract
+
+The action passed to `ResumeCache.forEach(BiFunction<K, V, Boolean>)` returns
`true` to invalidate (remove) the entry
+and `false` to keep it, as documented on `ResumeAction.evalEntry` and as the
Caffeine resume cache has always done.
+The cache created by `TransientResumeStrategy.createSimpleCache()` previously
did it the other way round: it removed
+the entries for which the action returned `false`. It now removes the entries
for which the action returns `true`.
+
+* A custom `ResumeAction`, or any other action lambda passed to
`ResumeCache.forEach`, that was written for the old
+ behaviour of the simple cache must swap its return values: `false` keeps the
entry, `true` removes it.
+* A custom `ResumeCache` implementation must remove an entry when the action
returns `true`, and must visit every
+ entry (do not stop iterating when the action returns `false`).
+
+*Impact on `KafkaResumeAdapter`:* `KafkaResumeAdapter.resume()` now returns
`false` (keep the cache entry), so that
+the topic-partition offset remains cached after the first seek and is
re-applied on subsequent resumes.
+Additionally, `KafkaResumeAdapter` now guards against seeking partitions that
are no longer in the current consumer
+assignment, avoiding `IllegalStateException` on revoked partitions.
+
=== camel-plc4x - PLC4X 1.0.0 and the triggered consumer
`camel-plc4x` now uses PLC4X 1.0.0, which requires Java 21, as Camel 4.24 does.
