gnodet-bot commented on code in PR #29505:
URL: https://github.com/apache/camel/pull/29505#discussion_r4552965181
##########
components/camel-redis/src/main/java/org/apache/camel/component/redis/processor/aggregate/RedisAggregationRepository.java:
##########
@@ -58,6 +60,19 @@ public class RedisAggregationRepository extends
ServiceSupport
private Map<String, DefaultExchangeHolder> persistedCache;
private boolean shutdownRedisson;
+ private static final String OPTIMISTIC_REMOVE_SCRIPT = """
+ local current = redis.call('HGET', KEYS[1], ARGV[1])
+
+ if not current or current ~= ARGV[2] then
+ return 0
+ end
+
+ redis.call('HSET', KEYS[2], ARGV[3], ARGV[4])
+ redis.call('HDEL', KEYS[1], ARGV[1])
+
+ return 1
+ """;
Review Comment:
💡 **Missing comment explaining the script's invariant and codec dependency.**
The Lua script implements compare-and-swap: it reads `HGET(active_map,
key)`, compares it against `ARGV[2]` (the serialized `holder`), and only
proceeds if they match. Two non-obvious facts that readers must know to trust
this script:
1. **Invariant:** The equality check (`current ~= ARGV[2]`) is the atomic
guard — if the active value has been updated by another thread between `get()`
and `remove()`, the comparison fails and the script returns `0`, triggering
`OptimisticLockingException`. This is the entire point of the atomic script vs.
the previous `RMap#remove(key, value)` call.
2. **Codec dependency:** `ARGV[2]` is the serialized `holder` object. For
the byte-level `~=` comparison to work correctly, the codec used by Redisson to
serialize the value stored via `HSET` in `add()` must produce identical bytes
when the same `Exchange` is serialized again here. If the codec is
non-deterministic or uses timestamps/random IVs, the comparison will never
match. Consider asserting this assumption with a comment.
Suggested addition above the field:
```suggestion
/**
* Atomic Lua script for optimistic removal with recovery.
*
* <p>Implements compare-and-swap: reads the current value from the
active map (KEYS[1])
* and compares it byte-for-byte with the expected value (ARGV[2], the
serialized holder).
* If they match, atomically writes the entry to the recovery map
(KEYS[2]) and removes it
* from the active map. Returns 1 on success, 0 on CAS mismatch.
*
* <p><b>Codec requirement:</b> The serialization of {@code holder}
passed as ARGV[2] must
* produce bytes identical to those stored by {@code add()} for the
equality check to work.
* This holds as long as Redisson's configured codec is deterministic
(the default is).
*/
private static final String OPTIMISTIC_REMOVE_SCRIPT = """
##########
components/camel-redis/src/test/java/org/apache/camel/component/redis/processor/aggregate/integration/RedisAggregationRepositoryOperationsIT.java:
##########
@@ -151,6 +153,104 @@ public void testOptimisticRemoveWithRecovery() {
}
}
+ @Test
+ public void testOptimisticRemoveWhenRecoveryWriteFails() {
+ String mapName = "optimisticRemoveRecoveryFailure";
+ RedisAggregationRepository repo = createRepo(mapName, true);
+
+ try {
+ // 1. Add an exchange to the active map
+ Exchange exchange = createExchange("recoverable");
+ repo.add(context, "key1", null, exchange);
+
+ // 2. Create a Redis String at the recovery map's key
+ String recoveryMapName = mapName + "-completed";
+ repo.getRedisson()
+ .getBucket(recoveryMapName)
+ .set("wrong-type");
+
+ // 3. Attempt removal; recovery write should fail
+ assertThrows(RedisException.class,
+ () -> repo.remove(context, "key1", exchange));
+
+ // 4. Verify the exchange still exists in the active map
+ Exchange result = repo.get(context, "key1");
+ assertNotNull(result);
+ assertEquals("recoverable", result.getIn().getBody(String.class));
+
+ } finally {
+ repo.stop();
+ }
+ }
+
+ @Test
+ public void testOptimisticRemoveCleansUpRecoveryOnLockingFailure() {
Review Comment:
⚠️ **Test name is misleading — no cleanup occurs on locking failure.**
The test is named `testOptimisticRemoveCleansUpRecoveryOnLockingFailure`,
which implies that a recovery entry was first created and then cleaned up when
optimistic locking fails. But that's not what happens.
Looking at the Lua script: when the CAS check fails (`current ~= ARGV[2]`),
the script returns `0` **before** executing the `HSET` into the recovery map.
So when locking fails, **no recovery entry is ever written** — there is nothing
to clean up.
The `assertNull(repo.recover(context, stale.getExchangeId()))` at step 6
therefore passes vacuously: it verifies that the recovery map is empty, which
is trivially true because the atomic script never wrote to it.
The test *does* cover an important property (failed CAS never contaminates
the recovery map), but the name should reflect what actually happens:
```suggestion
public void
testOptimisticRemoveDoesNotPopulateRecoveryOnLockingFailure() {
```
Also consider adding a comment explaining *why* the assert passes — the
script's atomicity guarantees no partial write occurred.
--
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]
