davsclaus commented on code in PR #29363:
URL: https://github.com/apache/camel/pull/29363#discussion_r4498177435
##########
components/camel-chaos/src/main/java/org/apache/camel/component/chaos/ChaosManager.java:
##########
@@ -386,6 +387,74 @@ boolean isCounting() {
return counting;
}
+ // the faults of the copies of circuit breakers, by the id of the exchange
they are a copy of, counted when that
+ // exchange is done (CAMEL-26701); bounded, and counted with the copy when
no event comes for a minute
+ private static final int MAX_WAITING = 10_000;
+ private static final long WAIT_FOR_ORIGINAL_MILLIS = 60_000;
+ private final ConcurrentMap<String, WaitingFaults> waitingForOriginal =
new ConcurrentHashMap<>();
+
+ /** The faults of copies of an exchange, and how the last copy was done. */
+ private static final class WaitingFaults {
+ final ChaosExchangeState.Faults faults;
+ final boolean copyFailed;
+ final boolean copyHandled;
+ final long since;
+
+ WaitingFaults(ChaosExchangeState.Faults faults, boolean copyFailed,
boolean copyHandled, long since) {
+ this.faults = faults;
+ this.copyFailed = copyFailed;
+ this.copyHandled = copyHandled;
+ this.since = since;
+ }
+ }
+
+ /**
+ * The faults of a copy of a circuit breaker wait for the exchange it is a
copy of.
+ *
+ * @return false when too many wait already, so the faults count now, with
the copy
+ */
+ boolean waitForOriginal(String exchangeId, ChaosExchangeState.Faults
faults, boolean copyFailed, boolean copyHandled) {
+ long now = TimeSource.currentTimeMillis();
+ countWaitedTooLong(now);
+ if (waitingForOriginal.size() >= MAX_WAITING) {
+ return false;
+ }
+ // another copy when the circuit breaker runs again at a redelivery of
the exchange: one message
+ waitingForOriginal.merge(exchangeId, new WaitingFaults(faults,
copyFailed, copyHandled, now), (before, copy) -> {
+ before.faults.merge(copy.faults);
+ return new WaitingFaults(before.faults, copy.copyFailed,
copy.copyHandled, copy.since);
+ });
+ return true;
+ }
+
+ /** Whether faults wait for an exchange to be done, so the events of done
exchanges are listened to. */
+ boolean isWaitingForOriginals() {
+ return !waitingForOriginal.isEmpty();
+ }
+
+ /** An exchange is done: the faults of the copies of circuit breakers that
wait for it count. */
+ void originalDone(Exchange exchange) {
+ WaitingFaults waiting =
waitingForOriginal.remove(exchange.getExchangeId());
+ if (waiting != null) {
+ waiting.faults.count(exchange);
+ }
+ }
+
+ /** The faults that waited a minute for an exchange count with how their
copy was done. */
+ void countWaitedTooLong(long now) {
+ if (waitingForOriginal.isEmpty()) {
+ return;
+ }
+ waitingForOriginal.entrySet().removeIf(e -> {
+ WaitingFaults w = e.getValue();
+ if (now - w.since < WAIT_FOR_ORIGINAL_MILLIS) {
+ return false;
+ }
+ w.faults.count(w.copyFailed, w.copyHandled, false);
+ return true;
+ });
+ }
Review Comment:
Thanks, fixed in a3048dcc19df. I went a step further than the suggestion.
The merge function changes `before.faults` in place, so collecting the entries
first would still count an entry that a copy merged into just before it was
removed. Now each timed-out entry is removed with `remove(key, value)`, which
succeeds only if no merge replaced the value. The faults then count once, after
they have left the map, so nothing else can change them. A new test,
`theFaultsOfACopyCountOnceWithItsOutcomeWhenTheExchangeIsNotDoneInAMinute`,
covers the timeout path: two copies of one exchange count once, as handled,
after a minute.
_Claude Code on behalf of Claus Ibsen_
--
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]
