This is an automated email from the ASF dual-hosted git repository.
davsclaus pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/camel.git
The following commit(s) were added to refs/heads/main by this push:
new 8c1e61d7b6fc CAMEL-24003: Fix thread-safety in BacklogTracer message
history queue
8c1e61d7b6fc is described below
commit 8c1e61d7b6fc3371d3cd97940bb25647aaf103ce
Author: Claus Ibsen <[email protected]>
AuthorDate: Mon Jul 13 09:46:35 2026 +0200
CAMEL-24003: Fix thread-safety in BacklogTracer message history queue
Fix race condition in BacklogTracer.traceEvent() where concurrent calls
from multiple threads (e.g. timer thread and Kafka consumer threads)
could corrupt the message history queue state. The compound queue
operations (peek/offer/addAll/clear) on provisionalHistoryQueue and
completeHistoryQueue are now synchronized to prevent events from being
dropped and the complete history becoming permanently stale.
Closes #24637
Co-Authored-By: Claude Opus 4.6 <[email protected]>
---
.../apache/camel/impl/debugger/BacklogTracer.java | 108 +++++++++++----------
1 file changed, 58 insertions(+), 50 deletions(-)
diff --git
a/core/camel-base-engine/src/main/java/org/apache/camel/impl/debugger/BacklogTracer.java
b/core/camel-base-engine/src/main/java/org/apache/camel/impl/debugger/BacklogTracer.java
index c716dfbcd11c..1e5d07c7d77f 100644
---
a/core/camel-base-engine/src/main/java/org/apache/camel/impl/debugger/BacklogTracer.java
+++
b/core/camel-base-engine/src/main/java/org/apache/camel/impl/debugger/BacklogTracer.java
@@ -78,6 +78,7 @@ public class BacklogTracer extends ServiceSupport implements
org.apache.camel.sp
private final ActivityEventNotifier activityEventNotifier = new
ActivityEventNotifier();
private int activitySize = 100;
private static final long INFLIGHT_EVICTION_MILLIS = 5 * 60 * 1000;
+ private final Object historyLock = new Object();
private volatile String lastCompletedBreadcrumbId;
private boolean removeOnDump = true;
private int bodyMaxChars = 32 * 1024;
@@ -218,59 +219,66 @@ public class BacklogTracer extends ServiceSupport
implements org.apache.camel.sp
// handle capturing events for last full completed exchange (aka
replay)
if (camelContext.isMessageHistory()) {
- var head = provisionalHistoryQueue.peek();
- String bid = null;
- String tid = null;
- if (head != null) {
- bid = head.getBreadcrumbId();
- tid = head.getExchangeId();
- }
- // correlate by breadcrumb ID when available (links exchanges
across broker boundaries)
- // fallback to exchange ID / correlation ID matching when
breadcrumb is not set
- boolean match;
- if (bid != null && event.getBreadcrumbId() != null) {
- match = bid.equals(event.getBreadcrumbId());
- } else {
- match = tid == null || tid.equals(event.getExchangeId()) ||
tid.equals(event.getCorrelationExchangeId());
- }
- // check if this event continues a previously completed breadcrumb
flow
- // (e.g. a downstream route connected via Kafka/SEDA that starts
after the originating route finished)
- boolean appendMode = false;
- if (head == null && event.getBreadcrumbId() != null
- &&
event.getBreadcrumbId().equals(lastCompletedBreadcrumbId)) {
- appendMode = true;
- } else if (head != null && event.getBreadcrumbId() != null
- &&
event.getBreadcrumbId().equals(lastCompletedBreadcrumbId)
- && head.getBreadcrumbId() != null
- &&
head.getBreadcrumbId().equals(lastCompletedBreadcrumbId)) {
- appendMode = true;
- }
- if (match || appendMode) {
- boolean added = provisionalHistoryQueue.offer(event);
- boolean original = head != null && event.getRouteId() != null
&& event.getRouteId().equals(head.getRouteId());
- if (event.isLast() && original) {
- if (appendMode) {
- // downstream route finished: merge into existing
complete history
- completeHistoryQueue.addAll(provisionalHistoryQueue);
- if (!added) {
- completeHistoryQueue.add(event);
- }
- } else {
- // originating route finished: replace complete history
- completeHistoryQueue.clear();
- completeHistoryQueue.addAll(provisionalHistoryQueue);
- if (!added) {
- completeHistoryQueue.add(event);
+ // synchronized to prevent race conditions when multiple threads
(e.g. timer thread
+ // and Kafka consumer threads) call traceEvent concurrently — the
peek/offer/addAll/clear
+ // sequence must be atomic to avoid dropping events or corrupting
the queue state
+ synchronized (historyLock) {
+ var head = provisionalHistoryQueue.peek();
+ String bid = null;
+ String tid = null;
+ if (head != null) {
+ bid = head.getBreadcrumbId();
+ tid = head.getExchangeId();
+ }
+ // correlate by breadcrumb ID when available (links exchanges
across broker boundaries)
+ // fallback to exchange ID / correlation ID matching when
breadcrumb is not set
+ boolean match;
+ if (bid != null && event.getBreadcrumbId() != null) {
+ match = bid.equals(event.getBreadcrumbId());
+ } else {
+ match = tid == null || tid.equals(event.getExchangeId())
+ || tid.equals(event.getCorrelationExchangeId());
+ }
+ // check if this event continues a previously completed
breadcrumb flow
+ // (e.g. a downstream route connected via Kafka/SEDA that
starts after the originating route finished)
+ boolean appendMode = false;
+ if (head == null && event.getBreadcrumbId() != null
+ &&
event.getBreadcrumbId().equals(lastCompletedBreadcrumbId)) {
+ appendMode = true;
+ } else if (head != null && event.getBreadcrumbId() != null
+ &&
event.getBreadcrumbId().equals(lastCompletedBreadcrumbId)
+ && head.getBreadcrumbId() != null
+ &&
head.getBreadcrumbId().equals(lastCompletedBreadcrumbId)) {
+ appendMode = true;
+ }
+ if (match || appendMode) {
+ boolean added = provisionalHistoryQueue.offer(event);
+ boolean original
+ = head != null && event.getRouteId() != null &&
event.getRouteId().equals(head.getRouteId());
+ if (event.isLast() && original) {
+ if (appendMode) {
+ // downstream route finished: merge into existing
complete history
+
completeHistoryQueue.addAll(provisionalHistoryQueue);
+ if (!added) {
+ completeHistoryQueue.add(event);
+ }
+ } else {
+ // originating route finished: replace complete
history
+ completeHistoryQueue.clear();
+
completeHistoryQueue.addAll(provisionalHistoryQueue);
+ if (!added) {
+ completeHistoryQueue.add(event);
+ }
+ lastCompletedBreadcrumbId =
event.getBreadcrumbId();
}
- lastCompletedBreadcrumbId = event.getBreadcrumbId();
+ provisionalHistoryQueue.clear();
}
- provisionalHistoryQueue.clear();
+ } else if (lastCompletedBreadcrumbId != null &&
event.getBreadcrumbId() != null
+ &&
event.getBreadcrumbId().equals(lastCompletedBreadcrumbId)) {
+ // late-arriving event from a downstream route (e.g.
second branch of a multicast via Kafka/SEDA)
+ // that arrived after the provisional queue was claimed by
a new exchange
+ completeHistoryQueue.add(event);
}
- } else if (lastCompletedBreadcrumbId != null &&
event.getBreadcrumbId() != null
- &&
event.getBreadcrumbId().equals(lastCompletedBreadcrumbId)) {
- // late-arriving event from a downstream route (e.g. second
branch of a multicast via Kafka/SEDA)
- // that arrived after the provisional queue was claimed by a
new exchange
- completeHistoryQueue.add(event);
}
}
