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 fff4c415f5c4 CAMEL-26796: camel-hazelcast - fix optimistic remove()
race that deletes winner's recovery entry (#29564)
fff4c415f5c4 is described below
commit fff4c415f5c45ca274a8d915d67e3e2026573fa3
Author: Guillaume Nodet <[email protected]>
AuthorDate: Thu Nov 12 21:46:40 2026 +0100
CAMEL-26796: camel-hazelcast - fix optimistic remove() race that deletes
winner's recovery entry (#29564)
In the optimistic remove() path, HazelcastAggregationRepository wrote the
recovery entry to persistedCache before the compare-and-remove on cache. When
two threads raced on the same group (same exchangeId), the loser's
removeFromRecoverableStorage() deleted the winner's recovery entry, so a later
failure of the winner lost the group. Introduced by CAMEL-26663 / #29278.
With recovery enabled, the optimistic path now does the conditional remove
and the recovery write in one ONE_PHASE Hazelcast transaction, as the
non-optimistic path already does. The loser rolls back without touching
persistedCache and throws OptimisticLockingException, and a crash
mid-transaction rolls back atomically. Without recovery, the plain
cache.remove(key, holder) path is unchanged. Adds a deterministic regression
test for the race.
Created by an AI coding agent (Hermes/forgebot) on behalf of gnodet.
---
.../hazelcast/HazelcastAggregationRepository.java | 96 ++++++++-----
...zelcastAggregationRepositoryOperationsTest.java | 151 +++++++++++++++------
2 files changed, 169 insertions(+), 78 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 b4ee97a5d167..8791ab3dbbf2 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
@@ -346,9 +346,13 @@ public class HazelcastAggregationRepository extends
ServiceSupport
}
/**
- * This method performs transactional operation on removing the {@code
exchange} from the operational storage and
- * moving it into the persistent one if the {@link
HazelcastAggregationRepository} runs in recoverable mode and
- * {@code optimistic} is false. It will act at <u>your own</u> risk
otherwise.
+ * Removes the {@code exchange} from the operational storage and moves it
into the persistent one if the
+ * {@link HazelcastAggregationRepository} runs in recoverable mode. Both
operations are applied atomically inside a
+ * Hazelcast transaction (ONE_PHASE): when {@code optimistic} is {@code
true} and {@code useRecovery} is
+ * {@code true}, a {@link
com.hazelcast.transaction.TransactionalMap#remove(Object, Object)} is used as
the
+ * conditional compare-and-delete, so the losing thread's transaction sees
the entry already gone, rolls back, and
+ * never touches the persistent map. When {@code optimistic} is {@code
false} and {@code useRecovery} is
+ * {@code true}, the same transactional pattern is used without the
conditional check.
*
* @param camelContext the current CamelContext
* @param key the correlation key
@@ -358,27 +362,65 @@ 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);
+ if (useRecovery) {
+ // Use a Hazelcast transaction to atomically apply the
conditional remove and the recovery write.
+ // A plain two-step approach (in either order) has two failure
modes:
+ // 1. put-then-remove (the pre-fix order): the two steps
cannot be fenced: a concurrent thread
+ // holds the same exchangeId (it read the same stored
entry), so it would delete the winning
+ // thread's recovery entry during its own cleanup on the
compare-and-remove failure (CAMEL-26796).
+ // 2. remove-then-put: if the JVM stops or
persistedCache.put() throws after a successful
+ // cache.remove(), the group is in neither map
(CAMEL-26663).
+ // TransactionalMap.remove(key, value) is a compare-and-delete
that enrolls the entry in the
+ // transaction: the entry is gone or holds a different value
from the loser's perspective, so
+ // its remove returns false, and it never reaches the recovery
write.
+ TransactionOptions tOpts = new TransactionOptions();
+
tOpts.setTransactionType(TransactionOptions.TransactionType.ONE_PHASE);
+ TransactionContext tCtx =
hazelcastInstance.newTransactionContext(tOpts);
+ try {
+ tCtx.beginTransaction();
+
+ TransactionalMap<String, DefaultExchangeHolder> tCache =
tCtx.getMap(cache.getName());
+ TransactionalMap<String, DefaultExchangeHolder>
tPersistentCache = tCtx.getMap(persistedCache.getName());
+
+ if (!tCache.remove(key, holder)) {
+ tCtx.rollbackTransaction();
+ LOG.error(
+ "Optimistic locking failed for exchange with
key {}: TransactionalMap#remove returned false (compare-and-delete found no
matching entry).",
+ key);
+ throw new OptimisticLockingException();
+ }
+ LOG.trace("Putting an exchange with ID {} for key {} into
a recoverable storage in an optimistic manner.",
+ exchange.getExchangeId(), key);
+ // store the given exchange and not the removed entry:
when a group is completed by an incoming
+ // exchange, the last aggregated exchange is not added to
the repository before it is removed
+ tPersistentCache.put(exchange.getExchangeId(), holder);
+
+ tCtx.commitTransaction();
+ LOG.trace("Removed an exchange with ID {} for key {} in an
optimistic manner.", exchange.getExchangeId(),
+ key);
+ LOG.trace("Put an exchange with ID {} for key {} into a
recoverable storage in an optimistic manner.",
+ exchange.getExchangeId(), key);
+ } catch (OptimisticLockingException e) {
+ throw e;
+ } catch (Exception exception) {
+ tCtx.rollbackTransaction();
+
+ final String msg = String.format(
+ "Transaction with ID %s was rolled back for
optimistic remove operation with a key %s and an Exchange ID %s.",
+ tCtx.getTxnId(), key, exchange.getExchangeId());
+ LOG.warn(msg, exception);
+ throw new RuntimeCamelException(msg, exception);
}
- throw new OptimisticLockingException();
+ } else {
+ if (!cache.remove(key, holder)) {
+ LOG.error(
+ "Optimistic locking failed for exchange with key
{}: IMap#remove returned false (compare-and-delete found no matching entry).",
+ key);
+ throw new OptimisticLockingException();
+ }
+ LOG.trace("Removed an exchange with ID {} for key {} in an
optimistic manner.", exchange.getExchangeId(), key);
}
- LOG.trace("Removed an exchange with ID {} for key {} 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);
@@ -423,18 +465,6 @@ 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 ded3b5c26948..d6e9c326b8be 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,12 +18,15 @@ package org.apache.camel.processor.aggregate.hazelcast;
import java.util.Set;
-import com.hazelcast.core.HazelcastException;
-import com.hazelcast.map.IMap;
+import com.hazelcast.core.HazelcastInstance;
+import com.hazelcast.transaction.TransactionContext;
+import com.hazelcast.transaction.TransactionException;
+import com.hazelcast.transaction.TransactionOptions;
import org.apache.camel.Exchange;
+import org.apache.camel.RuntimeCamelException;
import
org.apache.camel.spi.OptimisticLockingAggregationRepository.OptimisticLockingException;
-import org.apache.camel.support.DefaultExchangeHolder;
import org.junit.jupiter.api.Test;
+import org.mockito.AdditionalAnswers;
import static org.apache.camel.test.junit6.TestSupport.assertCollectionSize;
import static org.junit.jupiter.api.Assertions.assertEquals;
@@ -32,9 +35,9 @@ 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.doAnswer;
+import static org.mockito.Mockito.doThrow;
import static org.mockito.Mockito.mock;
-import static org.mockito.Mockito.when;
public class HazelcastAggregationRepositoryOperationsTest extends
HazelcastAggregationRepositoryCamelTestSupport {
@@ -193,78 +196,136 @@ public class
HazelcastAggregationRepositoryOperationsTest extends HazelcastAggre
}
@Test
- public void
checkOptimisticRemoveKeepsTheGroupWhenTheRecoverableStorageFails() throws
Exception {
+ public void checkOptimisticRemoveOfAStaleExchangeLeavesNothingToRecover()
throws Exception {
HazelcastAggregationRepository repo
- = new
HazelcastAggregationRepository("optimisticRemovePutFailsRepo", true,
getFirstInstance());
+ = new
HazelcastAggregationRepository("optimisticRemoveStaleRepo", true,
getFirstInstance());
try {
repo.doStart();
- final String key = "optimisticRemovePutFailsKey";
- Exchange ex = createExchangeWithBody("This group must not be
lost.");
+ final String key = "optimisticRemoveStaleKey";
+ 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.put(anyString(), any())).thenThrow(new
HazelcastException("Simulated failure"));
- repo.persistedCache = failingPersistedCache;
-
- assertThrows(HazelcastException.class, () ->
repo.remove(context(), key, ex));
+ assertThrows(OptimisticLockingException.class, () ->
repo.remove(context(), key, staleEx));
- 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());
+ 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();
}
}
+ /**
+ * Regression test for CAMEL-26663: when the transaction that atomically
removes the entry from the operational
+ * cache and writes the recovery entry fails to commit, the group must
still be present in the operational map so it
+ * can be retried — it must not be silently lost.
+ * <p>
+ * This test drives the production {@code remove()} via a Mockito delegate
that intercepts
+ * {@link HazelcastInstance#newTransactionContext} and makes {@code
commitTransaction()} throw, so that if the
+ * production code were changed back to a plain remove-then-put, this test
would still catch the regression.
+ */
@Test
- public void checkOptimisticRemoveOfAStaleExchangeLeavesNothingToRecover()
throws Exception {
+ public void checkOptimisticRemoveKeepsTheGroupWhenTheTransactionFails()
throws Exception {
+ final String persistentRepoName =
"optimisticRemoveTxFailsRepo-completed";
+
+ // Add the group using the real instance
+ HazelcastAggregationRepository realRepo
+ = new HazelcastAggregationRepository(
+ "optimisticRemoveTxFailsRepo", persistentRepoName,
true, getFirstInstance());
+ realRepo.doStart();
+ final String key = "optimisticRemoveTxFailsKey";
+ Exchange ex = createExchangeWithBody("This group must not be lost.");
+ realRepo.add(context(), key, null, ex);
+ realRepo.doStop();
+
+ // Wrap the real instance with a Mockito delegate that makes
commitTransaction() throw
+ HazelcastInstance real = getFirstInstance();
+ HazelcastInstance hz = mock(HazelcastInstance.class,
AdditionalAnswers.delegatesTo(real));
+ doAnswer(inv -> {
+ TransactionContext ctx = mock(TransactionContext.class,
+
AdditionalAnswers.delegatesTo(real.newTransactionContext(inv.getArgument(0))));
+ doThrow(new TransactionException("Simulated commit failure"))
+ .when(ctx).commitTransaction();
+ return ctx;
+ }).when(hz).newTransactionContext(any(TransactionOptions.class));
+
+ // Drive the production remove() — the commit will fail, triggering
rollback and RuntimeCamelException
HazelcastAggregationRepository repo
- = new
HazelcastAggregationRepository("optimisticRemoveStaleRepo", true,
getFirstInstance());
-
+ = new
HazelcastAggregationRepository("optimisticRemoveTxFailsRepo",
persistentRepoName, true, hz);
try {
repo.doStart();
+ assertThrows(RuntimeCamelException.class, () ->
repo.remove(context(), key, ex));
- 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));
+ // The group must still be in the operational map — the rollback
restored it
+ Exchange getBackEx = realRepo.get(context(), key);
+ assertNotNull(getBackEx, "the group must still be in the
repository after a failed transaction");
+ assertEquals(ex.getIn().getBody(), getBackEx.getIn().getBody());
- 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());
+ // No recovery entry must have been written — the transaction was
rolled back before the put
+ assertTrue(realRepo.scan(context()).isEmpty(),
+ "no recovery entry must have been written after a
rolled-back transaction");
} finally {
repo.doStop();
}
}
+ /**
+ * Regression test for CAMEL-26796: in the optimistic remove() path, two
threads racing on the same group both hold
+ * the same exchange (same exchangeId). The losing thread must not corrupt
the winning thread's recovery entry.
+ * <p>
+ * Before the fix, the cleanup call on the losing thread would delete the
winning thread's recovery entry (they both
+ * had the same exchangeId), making the group irrecoverable on a
downstream failure. The fix uses a Hazelcast
+ * transaction to atomically apply the conditional remove and the recovery
write, so the loser's transaction sees
+ * the entry already removed and rolls back without touching the
persistent map.
+ * <p>
+ * Scenario: add a group, get it twice as {@code a} and {@code b} (same
content, same exchangeId), remove(a)
+ * succeeds, remove(b) throws {@link OptimisticLockingException}. After
the failed remove, recover(exchangeId) must
+ * still return the group so that downstream failures can be recovered.
+ * <p>
+ * Note: the transaction also closes the crash-between-two-steps window
that CAMEL-26663 originally fixed: if the
+ * JVM stops after the conditional remove but before the recovery write,
the transaction is rolled back on the
+ * cluster side (owner failure or timeout) and the group re-appears in the
operational repository.
+ */
@Test
- public void
checkOptimisticRemoveOfAStaleExchangeFailsWhenTheCleanupFails() throws
Exception {
- HazelcastAggregationRepository repo
- = new
HazelcastAggregationRepository("optimisticRemoveCleanupFailsRepo", true,
getFirstInstance());
+ public void checkOptimisticRemoveRaceDoesNotDeleteWinnersRecoveryEntry()
throws Exception {
+ final String persistentRepoName = "optimisticRaceRepo-completed";
+ HazelcastAggregationRepository repoOne
+ = new HazelcastAggregationRepository("optimisticRaceRepo",
persistentRepoName, true, getFirstInstance());
+ HazelcastAggregationRepository repoTwo
+ = new HazelcastAggregationRepository("optimisticRaceRepo",
persistentRepoName, true, getSecondInstance());
try {
- repo.doStart();
+ repoOne.doStart();
+ repoTwo.doStart();
- final String key = "optimisticRemoveCleanupFailsKey";
- Exchange ex = createExchangeWithBody("The current group.");
- repo.add(context(), key, null, ex);
- Exchange staleEx = createExchangeWithBody("A stale group.");
+ final String key = "raceKey";
+ Exchange ex = createExchangeWithBody("The group body.");
+ repoOne.add(context(), key, null, ex);
- IMap<String, DefaultExchangeHolder> failingPersistedCache =
mockMap();
- when(failingPersistedCache.remove(anyString())).thenThrow(new
HazelcastException("Simulated failure"));
- repo.persistedCache = failingPersistedCache;
+ // Both threads read the same stored exchange (same exchangeId,
same content)
+ Exchange a = repoOne.get(context(), key);
+ Exchange b = repoTwo.get(context(), key);
+ assertNotNull(a, "Thread A's exchange should not be null");
+ assertNotNull(b, "Thread B's exchange should not be null");
+ assertEquals(a.getExchangeId(), b.getExchangeId(), "Both threads
must hold the same exchangeId");
- assertThrows(OptimisticLockingException.class, () ->
repo.remove(context(), key, staleEx));
+ // Thread A wins the race: remove succeeds and a recovery entry is
written atomically
+ repoOne.remove(context(), key, a);
+
+ // Thread B loses: compare-and-remove fails (entry already removed
by A)
+ assertThrows(OptimisticLockingException.class, () ->
repoTwo.remove(context(), key, b));
+
+ // Thread A's recovery entry must still be there: if A's
downstream processing fails,
+ // the recovery task must be able to send the group
+ Exchange recovered = repoTwo.recover(context(), a.getExchangeId());
+ assertNotNull(recovered, "The winner's recovery entry must survive
the loser's failed remove");
+ assertEquals(ex.getIn().getBody(), recovered.getIn().getBody());
} finally {
- repo.doStop();
+ repoOne.doStop();
+ repoTwo.doStop();
}
}
- @SuppressWarnings("unchecked")
- private static IMap<String, DefaultExchangeHolder> mockMap() {
- return mock(IMap.class);
- }
}
