gnodet-bot commented on code in PR #29505:
URL: https://github.com/apache/camel/pull/29505#discussion_r4515444820
##########
components/camel-redis/src/main/java/org/apache/camel/component/redis/processor/aggregate/RedisAggregationRepository.java:
##########
@@ -296,21 +296,33 @@ public void setAllowSerializedHeaders(boolean
allowSerializedHeaders) {
public void remove(CamelContext camelContext, String key, Exchange
exchange) {
DefaultExchangeHolder holder = DefaultExchangeHolder.marshal(exchange,
true, allowSerializedHeaders);
if (optimistic) {
- LOG.trace("Removing an exchange with ID {} for key {} in an
optimistic manner.", exchange.getExchangeId(), key);
- if (!cache.remove(key, holder)) {
- LOG.warn(
- "Optimistic locking failed for exchange with key {}:
RMap#remove removed no Exchanges, while it's expected to remove one.",
- key);
- throw new OptimisticLockingException();
- }
- LOG.trace("Removed an exchange with ID {} for key {} in an
optimistic manner.", exchange.getExchangeId(), key);
+ LOG.trace("Removing 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);
}
+
+ if (!cache.remove(key, holder)) {
+ if (useRecovery) {
+ persistedCache.remove(exchange.getExchangeId());
Review Comment:
`persistedCache.remove()` is not guarded by a try/catch. If `persistedCache`
throws (e.g. Redis connectivity issue), the exception propagates out of
`remove()` as a storage exception rather than an `OptimisticLockingException`,
breaking the aggregator's retry logic which expects the latter.
Consider wrapping it:
```suggestion
try {
persistedCache.remove(exchange.getExchangeId());
} catch (Exception e) {
LOG.warn("Failed to remove exchange {} from persisted
cache during optimistic locking rollback", exchange.getExchangeId(), e);
}
```
--
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]
