davsclaus commented on code in PR #27989:
URL: https://github.com/apache/camel/pull/27989#discussion_r4301786220
##########
core/camel-core-processor/src/main/java/org/apache/camel/processor/aggregate/AggregateProcessor.java:
##########
@@ -1580,6 +1580,24 @@ private void onEviction(Listener.Type type, String key,
String exchangeId) {
if (type != Listener.Type.Evict) {
return;
}
+ try {
+ doOnEviction(key, exchangeId);
+ } catch (Exception e) {
+ // the entry is no longer in the timeout map, and the timeout
map ignores errors of its listeners, so
+ // without a new entry a group the repository could not read
or remove would never time out (an entry
+ // that a new message for the key has put in the meantime is
kept)
+ long delay = getCompletionTimeoutCheckerInterval();
+ LOG.warn("Error completing the group with correlation key: {}
by its completion timeout,"
+ + " will try again in {} ms. Caused by: {}",
+ key, delay, e.getMessage());
+ LOG.debug("Error completing the group with correlation key: {}
by its completion timeout", key, e);
+ if (isRunAllowed()) {
+ putIfAbsent(key, exchangeId, delay);
Review Comment:
Non-blocking: for a permanent failure (for example a stored exchange that
can no longer be unmarshalled after an upgrade), each such key is now retried,
and logged at WARN, every `completionTimeoutCheckerInterval` (1 s by default)
for ever. `purge()` runs under the aggregator `lock`, so with a store that is
down and slow to fail (a connect timeout), every check also holds back incoming
exchanges for the length of those calls. This is still better than losing the
group silently. If you want to bound it, two options: log the first failure at
WARN and the repeats at DEBUG, or back off, for example by retrying after
`Math.max(getCompletionTimeoutCheckerInterval(), getCompletionTimeout())`.
##########
core/camel-core-processor/src/main/java/org/apache/camel/processor/aggregate/AggregateProcessor.java:
##########
@@ -1580,6 +1580,24 @@ private void onEviction(Listener.Type type, String key,
String exchangeId) {
if (type != Listener.Type.Evict) {
return;
}
+ try {
+ doOnEviction(key, exchangeId);
+ } catch (Exception e) {
+ // the entry is no longer in the timeout map, and the timeout
map ignores errors of its listeners, so
+ // without a new entry a group the repository could not read
or remove would never time out (an entry
+ // that a new message for the key has put in the meantime is
kept)
+ long delay = getCompletionTimeoutCheckerInterval();
+ LOG.warn("Error completing the group with correlation key: {}
by its completion timeout,"
+ + " will try again in {} ms. Caused by: {}",
+ key, delay, e.getMessage());
Review Comment:
Nit: `e.getMessage()` is `null` for some exceptions (an NPE, or some
JDBC/driver wrappers), which logs `Caused by: null`. Including the exception
type makes the WARN useful without the DEBUG stack trace:
```suggestion
key, delay, e.toString());
```
--
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]
