gnodet-bot commented on code in PR #29564:
URL: https://github.com/apache/camel/pull/29564#discussion_r4549719323
##########
components/camel-hazelcast/src/test/java/org/apache/camel/processor/aggregate/hazelcast/HazelcastAggregationRepositoryOperationsTest.java:
##########
@@ -193,78 +195,137 @@ 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-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 com.hazelcast.core.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
com.hazelcast.transaction.TransactionException("Simulated commit failure"))
Review Comment:
**Nit:** `TransactionContext` and `TransactionOptions` are imported, but
`TransactionException` is used as a fully-qualified name here. Add the import
for consistency:
```suggestion
doThrow(new TransactionException("Simulated commit failure"))
```
And add to the import block:
```java
import com.hazelcast.transaction.TransactionException;
```
##########
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:
Review Comment:
**Nit:** The comment describes a "remove then put" two-step approach, but
the original code (before this PR) was actually put-first:
`persistedCache.put(...)` before `cache.remove(key, holder)`. With put-first,
failure mode 1 (the cleanup race, CAMEL-26796) is straightforward — Thread B
writes its holder to `persistedCache`, overlaying Thread A's entry, then Thread
A's conditional remove succeeds while Thread B's fails and calls
`removeFromRecoverableStorage(a.getExchangeId())`. With the "remove then put"
order described here, failure mode 1 would not arise in the same way (there is
nothing in `persistedCache` to clean up when the loser's remove fails, since
the winner hasn't put yet). Suggest:
```suggestion
// A plain persistedCache.put(...) followed by
cache.remove(key, holder) (the pre-fix order) has two failure modes:
```
Or simply drop the order detail and say `// A plain two-step approach` — the
transaction explanation below is self-contained.
--
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]
