You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@camel.apache.org by da...@apache.org on 2026/07/13 06:19:02 UTC
(camel) branch fix/CAMEL-24003-backlog-tracer-thread-safety created (now 7c034e743daf)
This is an automated email from the ASF dual-hosted git repository.
davsclaus pushed a change to branch fix/CAMEL-24003-backlog-tracer-thread-safety
in repository https://gitbox.apache.org/repos/asf/camel.git
at 7c034e743daf CAMEL-24003: Fix thread-safety in BacklogTracer message history queue
This branch includes the following new commits:
new 7c034e743daf CAMEL-24003: Fix thread-safety in BacklogTracer message history queue
The 1 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
(camel) 01/01: CAMEL-24003: Fix thread-safety in BacklogTracer message history queue
Posted by da...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
davsclaus pushed a commit to branch fix/CAMEL-24003-backlog-tracer-thread-safety
in repository https://gitbox.apache.org/repos/asf/camel.git
commit 7c034e743dafb4302e494f2139dd43ba03e5c3ff
Author: Claus Ibsen <cl...@gmail.com>
AuthorDate: Mon Jul 13 08:18:47 2026 +0200
CAMEL-24003: Fix thread-safety in BacklogTracer message history queue
Co-Authored-By: Claude Opus 4.6 <no...@anthropic.com>
Signed-off-by: Claus Ibsen <cl...@gmail.com>
---
.../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);
}
}