This is an automated email from the ASF dual-hosted git repository. davsclaus pushed a commit to branch CAMEL-26663-hazelcast-optimistic-remove in repository https://gitbox.apache.org/repos/asf/camel.git
commit 156c9a6d8dcd5aa920dbea62dd4f1c672fa97a46 Author: Claus Ibsen <[email protected]> AuthorDate: Thu Nov 5 20:54:37 2026 +0100 CAMEL-26663: camel-hazelcast - HazelcastAggregationRepository optimistic remove() no longer loses a completed group The optimistic remove() removed the group from the active map before it put the exchange into the recovery map, so a crash or a failed put in between lost the group. Put it into the recovery map first, and take it out again when the optimistic remove fails. 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); + } }
