This is an automated email from the ASF dual-hosted git repository.
davsclaus 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 d224bceef2db CAMEL-26663: camel-hazelcast -
HazelcastAggregationRepository optimistic remove() no longer loses a completed
group (#29278)
d224bceef2db is described below
commit d224bceef2dbc6813ddd1fe4f2461e3ed84a7127
Author: Claus Ibsen <[email protected]>
AuthorDate: Thu Nov 5 21:56:56 2026 +0100
CAMEL-26663: camel-hazelcast - HazelcastAggregationRepository optimistic
remove() no longer loses a completed group (#29278)
Co-Authored-By: Claude Opus 5.5 <[email protected]>
Signed-off-by: Claus Ibsen <[email protected]>
---
.../hazelcast/HazelcastAggregationRepository.java | 32 ++++++--
...zelcastAggregationRepositoryOperationsTest.java | 86 ++++++++++++++++++++++
2 files changed, 111 insertions(+), 7 deletions(-)
diff --git
a/components/camel-hazelcast/src/main/java/org/apache/camel/processor/aggregate/hazelcast/HazelcastAggregationRepository.java
b/components/camel-hazelcast/src/main/java/org/apache/camel/processor/aggregate/hazelcast/HazelcastAggregationRepository.java
index 5a836a754a4b..b4ee97a5d167 100644
---
a/components/camel-hazelcast/src/main/java/org/apache/camel/processor/aggregate/hazelcast/HazelcastAggregationRepository.java
+++
b/components/camel-hazelcast/src/main/java/org/apache/camel/processor/aggregate/hazelcast/HazelcastAggregationRepository.java
@@ -358,21 +358,27 @@ public class HazelcastAggregationRepository extends
ServiceSupport
public void remove(CamelContext camelContext, String key, Exchange
exchange) {
DefaultExchangeHolder holder = DefaultExchangeHolder.marshal(exchange,
true, allowSerializedHeaders);
if (optimistic) {
+ // put the exchange into the recoverable storage before removing
it from the operational storage: the two
+ // maps cannot be updated atomically, and if the process stops in
between, the group is sent twice
+ // instead of being lost
+ if (useRecovery) {
+ LOG.trace("Putting an exchange with ID {} for key {} into a
recoverable storage in an optimistic manner.",
+ exchange.getExchangeId(), key);
+ persistedCache.put(exchange.getExchangeId(), holder);
+ LOG.trace("Put an exchange with ID {} for key {} into a
recoverable storage in an optimistic manner.",
+ exchange.getExchangeId(), key);
+ }
LOG.trace("Removing an exchange with ID {} for key {} in an
optimistic manner.", exchange.getExchangeId(), key);
if (!cache.remove(key, holder)) {
LOG.error(
"Optimistic locking failed for exchange with key {}:
IMap#remove removed no Exchanges, while it's expected to remove one.",
key);
+ if (useRecovery) {
+ removeFromRecoverableStorage(exchange.getExchangeId(),
key);
+ }
throw new OptimisticLockingException();
}
LOG.trace("Removed an exchange with ID {} for key {} in an
optimistic manner.", exchange.getExchangeId(), key);
- if (useRecovery) {
- LOG.trace("Putting an exchange with ID {} for key {} into a
recoverable storage in an optimistic manner.",
- exchange.getExchangeId(), key);
- persistedCache.put(exchange.getExchangeId(), holder);
- LOG.trace("Put an exchange with ID {} for key {} into a
recoverable storage in an optimistic manner.",
- exchange.getExchangeId(), key);
- }
} else {
if (useRecovery) {
LOG.trace("Removing an exchange with ID {} for key {} in a
thread-safe manner.", exchange.getExchangeId(), key);
@@ -417,6 +423,18 @@ public class HazelcastAggregationRepository extends
ServiceSupport
}
}
+ private void removeFromRecoverableStorage(String exchangeId, String key) {
+ try {
+ persistedCache.remove(exchangeId);
+ } catch (RuntimeException e) {
+ // still throw the optimistic locking failure: the recovery task
may send the stale entry, which is a
+ // duplicate and not a loss
+ LOG.warn("Could not remove the exchange with ID {} for key {} from
the recoverable storage after an"
+ + " optimistic locking failure. The recovery task may
send it.",
+ exchangeId, key, e);
+ }
+ }
+
@Override
public void confirm(CamelContext camelContext, String exchangeId) {
LOG.trace("Confirming an exchange with ID {}.", exchangeId);
diff --git
a/components/camel-hazelcast/src/test/java/org/apache/camel/processor/aggregate/hazelcast/HazelcastAggregationRepositoryOperationsTest.java
b/components/camel-hazelcast/src/test/java/org/apache/camel/processor/aggregate/hazelcast/HazelcastAggregationRepositoryOperationsTest.java
index d142205d9d94..ded3b5c26948 100644
---
a/components/camel-hazelcast/src/test/java/org/apache/camel/processor/aggregate/hazelcast/HazelcastAggregationRepositoryOperationsTest.java
+++
b/components/camel-hazelcast/src/test/java/org/apache/camel/processor/aggregate/hazelcast/HazelcastAggregationRepositoryOperationsTest.java
@@ -18,13 +18,23 @@ package org.apache.camel.processor.aggregate.hazelcast;
import java.util.Set;
+import com.hazelcast.core.HazelcastException;
+import com.hazelcast.map.IMap;
import org.apache.camel.Exchange;
+import
org.apache.camel.spi.OptimisticLockingAggregationRepository.OptimisticLockingException;
+import org.apache.camel.support.DefaultExchangeHolder;
import org.junit.jupiter.api.Test;
import static org.apache.camel.test.junit6.TestSupport.assertCollectionSize;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertNotNull;
import static org.junit.jupiter.api.Assertions.assertNull;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.anyString;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
public class HazelcastAggregationRepositoryOperationsTest extends
HazelcastAggregationRepositoryCamelTestSupport {
@@ -181,4 +191,80 @@ public class HazelcastAggregationRepositoryOperationsTest
extends HazelcastAggre
repoTwo.doStop();
}
}
+
+ @Test
+ public void
checkOptimisticRemoveKeepsTheGroupWhenTheRecoverableStorageFails() throws
Exception {
+ HazelcastAggregationRepository repo
+ = new
HazelcastAggregationRepository("optimisticRemovePutFailsRepo", true,
getFirstInstance());
+
+ try {
+ repo.doStart();
+
+ final String key = "optimisticRemovePutFailsKey";
+ Exchange ex = createExchangeWithBody("This group must not be
lost.");
+ repo.add(context(), key, null, ex);
+
+ IMap<String, DefaultExchangeHolder> failingPersistedCache =
mockMap();
+ when(failingPersistedCache.put(anyString(), any())).thenThrow(new
HazelcastException("Simulated failure"));
+ repo.persistedCache = failingPersistedCache;
+
+ assertThrows(HazelcastException.class, () ->
repo.remove(context(), key, ex));
+
+ Exchange getBackEx = repo.get(context(), key);
+ assertNotNull(getBackEx, "the group must still be in the
repository after a failed remove");
+ assertEquals(ex.getIn().getBody(), getBackEx.getIn().getBody());
+ } finally {
+ repo.doStop();
+ }
+ }
+
+ @Test
+ public void checkOptimisticRemoveOfAStaleExchangeLeavesNothingToRecover()
throws Exception {
+ HazelcastAggregationRepository repo
+ = new
HazelcastAggregationRepository("optimisticRemoveStaleRepo", true,
getFirstInstance());
+
+ try {
+ repo.doStart();
+
+ final String key = "optimisticRemoveStaleKey";
+ Exchange ex = createExchangeWithBody("The current group.");
+ repo.add(context(), key, null, ex);
+ Exchange staleEx = createExchangeWithBody("A stale group.");
+
+ assertThrows(OptimisticLockingException.class, () ->
repo.remove(context(), key, staleEx));
+
+ assertTrue(repo.scan(context()).isEmpty(), "a failed remove must
not leave an exchange to recover");
+ assertEquals(ex.getIn().getBody(), repo.get(context(),
key).getIn().getBody());
+ } finally {
+ repo.doStop();
+ }
+ }
+
+ @Test
+ public void
checkOptimisticRemoveOfAStaleExchangeFailsWhenTheCleanupFails() throws
Exception {
+ HazelcastAggregationRepository repo
+ = new
HazelcastAggregationRepository("optimisticRemoveCleanupFailsRepo", true,
getFirstInstance());
+
+ try {
+ repo.doStart();
+
+ final String key = "optimisticRemoveCleanupFailsKey";
+ Exchange ex = createExchangeWithBody("The current group.");
+ repo.add(context(), key, null, ex);
+ Exchange staleEx = createExchangeWithBody("A stale group.");
+
+ IMap<String, DefaultExchangeHolder> failingPersistedCache =
mockMap();
+ when(failingPersistedCache.remove(anyString())).thenThrow(new
HazelcastException("Simulated failure"));
+ repo.persistedCache = failingPersistedCache;
+
+ assertThrows(OptimisticLockingException.class, () ->
repo.remove(context(), key, staleEx));
+ } finally {
+ repo.doStop();
+ }
+ }
+
+ @SuppressWarnings("unchecked")
+ private static IMap<String, DefaultExchangeHolder> mockMap() {
+ return mock(IMap.class);
+ }
}
