gnodet-bot commented on code in PR #29564:
URL: https://github.com/apache/camel/pull/29564#discussion_r4549007726
##########
components/camel-hazelcast/src/main/java/org/apache/camel/processor/aggregate/hazelcast/HazelcastAggregationRepository.java:
##########
@@ -358,27 +362,65 @@ public void setHazelcastInstance(HazelcastInstance
hazelcastInstance) {
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 cache.remove(key, holder) followed by
persistedCache.put(...) has two failure modes:
+ // 1. 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. 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 loser sees the entry still present (or
already gone), 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 {}: IMap#remove removed no Exchanges, while it's expected to remove one.",
Review Comment:
🔧 **Nit:** The error message says `IMap#remove` but this code path calls
`tCache.remove(key, holder)` on a `TransactionalMap`, not an `IMap`. The
message in the `useRecovery=false` branch (line ~418) is correct since
`cache.remove` is an `IMap` there.
```suggestion
LOG.error(
"Optimistic locking failed for exchange with
key {}: TransactionalMap#remove removed no Exchanges, while it's expected to
remove one.",
key);
```
##########
components/camel-hazelcast/src/test/java/org/apache/camel/processor/aggregate/hazelcast/HazelcastAggregationRepositoryOperationsTest.java:
##########
@@ -193,78 +192,156 @@ public void checkOptimisticPersistentRemove() throws
Exception {
}
@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-26395 / CAMEL-26663: when the transaction
that atomically removes the entry from the
+ * operational cache and writes the recovery entry is rolled back (due to
any failure), the group must still be
+ * present in the operational map so it can be retried — it must not be
silently lost.
+ * <p>
+ * The original test for this guarantee (from CAMEL-26663) used Mockito to
inject a mock {@code persistedCache} IMap
+ * that throws on {@code put}. With the transaction-based fix for
CAMEL-26796, the code path goes through a
+ * {@link TransactionalMap} obtained from the {@code TransactionContext},
which bypasses a mock on the plain
+ * {@code persistedCache} field. This test therefore uses a subclass that
simulates a failure mid-transaction by
+ * rolling back explicitly after the conditional remove succeeds,
confirming that the transaction rollback restores
+ * the entry to the operational map.
+ */
@Test
- public void checkOptimisticRemoveOfAStaleExchangeLeavesNothingToRecover()
throws Exception {
+ public void checkOptimisticRemoveKeepsTheGroupWhenTheTransactionFails()
throws Exception {
+ final String persistentRepoName =
"optimisticRemoveTxFailsRepo-completed";
HazelcastAggregationRepository repo
- = new
HazelcastAggregationRepository("optimisticRemoveStaleRepo", true,
getFirstInstance());
+ = new
RollbackAfterRemoveRepository("optimisticRemoveTxFailsRepo",
persistentRepoName, getFirstInstance());
try {
repo.doStart();
- final String key = "optimisticRemoveStaleKey";
- Exchange ex = createExchangeWithBody("The current group.");
+ final String key = "optimisticRemoveTxFailsKey";
+ Exchange ex = createExchangeWithBody("This group must not be
lost.");
repo.add(context(), key, null, ex);
- Exchange staleEx = createExchangeWithBody("A stale group.");
- assertThrows(OptimisticLockingException.class, () ->
repo.remove(context(), key, staleEx));
+ // The subclass rolls the transaction back after the conditional
remove, simulating a commit failure.
+ assertThrows(RuntimeCamelException.class, () ->
repo.remove(context(), key, ex));
- 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());
+ // The group must still be in the operational map — the rollback
restored it.
+ Exchange getBackEx = repo.get(context(), key);
+ assertNotNull(getBackEx, "the group must still be in the
repository after a failed transaction");
+ assertEquals(ex.getIn().getBody(), getBackEx.getIn().getBody());
+
+ // No recovery entry must have been written — the transaction was
rolled back before the put.
+ assertTrue(repo.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 restart
+ * and the group re-appears in the operational repository.
Review Comment:
🔧 **Nit:** With `ONE_PHASE`, the uncommitted transaction is not rolled back
on JVM restart — it is rolled back on the cluster side when the transaction
owner leaves or the transaction times out. `"on restart"` is misleading.
```suggestion
* 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 commit, the
uncommitted transaction is rolled back
* automatically (via transaction timeout or owner departure) and the
group re-appears in the operational repository.
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
