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 a4d45f948160 CAMEL-26883: camel-core - BacklogDebugger: a resumed
exchange no longer waits on the breakpoint of another exchange (#29787)
a4d45f948160 is described below
commit a4d45f948160aeb1078beb7262d58bc7cf6f44ca
Author: allthingssecurity <[email protected]>
AuthorDate: Fri Nov 13 15:01:44 2026 +0530
CAMEL-26883: camel-core - BacklogDebugger: a resumed exchange no longer
waits on the breakpoint of another exchange (#29787)
Co-Authored-By: Claude Opus 5.5 <[email protected]>
---
.../impl/debugger/DefaultBacklogDebugger.java | 44 ++++--
.../BacklogDebuggerResumedBeforeSuspendTest.java | 158 +++++++++++++++++++++
2 files changed, 188 insertions(+), 14 deletions(-)
diff --git
a/core/camel-base-engine/src/main/java/org/apache/camel/impl/debugger/DefaultBacklogDebugger.java
b/core/camel-base-engine/src/main/java/org/apache/camel/impl/debugger/DefaultBacklogDebugger.java
index 7b2e8005a31a..fb62e6435b05 100644
---
a/core/camel-base-engine/src/main/java/org/apache/camel/impl/debugger/DefaultBacklogDebugger.java
+++
b/core/camel-base-engine/src/main/java/org/apache/camel/impl/debugger/DefaultBacklogDebugger.java
@@ -16,13 +16,13 @@
*/
package org.apache.camel.impl.debugger;
-import java.util.ArrayDeque;
import java.util.Collections;
import java.util.Deque;
import java.util.LinkedHashSet;
import java.util.List;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ConcurrentLinkedDeque;
import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
@@ -81,7 +81,8 @@ public final class DefaultBacklogDebugger extends
ServiceSupport implements Back
private final ConcurrentMap<String, BacklogTracerEventMessage>
suspendedBreakpointMessages = new ConcurrentHashMap<>();
private final AtomicReference<CountDownLatch> suspend = new
AtomicReference<>();
- private final Deque<String> singleStepExchangeId = new ArrayDeque<>();
+ // changed by the threads of the debugger client and of the exchanges
+ private final Deque<String> singleStepExchangeId = new
ConcurrentLinkedDeque<>();
private final AtomicBoolean stepOverMode = new AtomicBoolean();
private final AtomicBoolean skipOverMode = new AtomicBoolean();
@@ -924,6 +925,14 @@ public final class DefaultBacklogDebugger extends
ServiceSupport implements Back
isBodyIncludeStreams(), isBodyIncludeFiles(),
getBodyMaxChars());
}
+ /**
+ * Removes the message at the node if it is the message of the given
exchange (another exchange may be suspended at
+ * the node by now).
+ */
+ private void removeMessageOf(String nodeId, String exchangeId) {
+ suspendedBreakpointMessages.computeIfPresent(nodeId, (k, m) ->
exchangeId.equals(m.getExchangeId()) ? null : m);
+ }
+
/**
* Represents a {@link org.apache.camel.spi.Breakpoint} that has a {@link
Condition} on a specific node id.
*/
@@ -970,20 +979,27 @@ public final class DefaultBacklogDebugger extends
ServiceSupport implements Back
first, false, uid, timestamp, source, fromRouteId,
routeId, toNode, toNodeParentId, null, null,
toNodeShortName, toNodeLabel, level, exchangeId,
correlationExchangeId, breadcrumbId,
false, false, data);
- suspendedBreakpointMessages.put(nodeId, msg);
- // suspend at this breakpoint
+ // suspend at this breakpoint, with the suspended exchange that
matchProcess registered for this exchange:
+ // the breakpoint may have been resumed before this exchange got
here, and another exchange may be suspended
+ // at this node by now, which this exchange must neither wait for
nor replace in the debugger
final SuspendedExchange se = suspendedBreakpoints.get(nodeId);
- if (se != null) {
+ if (se != null && se.getExchange() == exchange) {
+ // publish the message only while this exchange is the one
suspended at the node, in one step with the
+ // check, as a resume and another exchange can come in between
+ suspendedBreakpointMessages.compute(nodeId, (k, m) ->
suspendedBreakpoints.get(k) == se ? msg : m);
// now wait until we should continue
logger.log(String.format("NodeBreakpoint at node %s is waiting
to continue for exchangeId: %s", toNode,
exchangeId));
try {
boolean hit = se.awaitExchange(fallbackTimeout);
+ // the message of this exchange only (a resume removes it
already); by exchange id, as the debugger
+ // replaces the message when the body or a header is
changed at the breakpoint
+ removeMessageOf(nodeId, exchangeId);
if (!hit) {
- // remove breakpoint as it timed out
- suspendedBreakpointMessages.remove(nodeId);
- suspendedBreakpoints.remove(nodeId);
+ // remove breakpoint as it timed out (only the
suspended exchange of this exchange: another
+ // exchange may be suspended at this node after a
resume)
+ suspendedBreakpoints.remove(nodeId, se);
logger.log(
String.format("NodeBreakpoint at node %s timed
out and is continued exchangeId: %s", toNode,
exchangeId),
@@ -1075,9 +1091,9 @@ public final class DefaultBacklogDebugger extends
ServiceSupport implements Back
try {
boolean hit = se.awaitExchange(fallbackTimeout);
if (!hit) {
- // remove breakpoint as it timed out
- suspendedBreakpointMessages.remove(toNode);
- suspendedBreakpoints.remove(toNode);
+ // remove breakpoint as it timed out (only the ones of
this exchange)
+ removeMessageOf(toNode, exchange.getExchangeId());
+ suspendedBreakpoints.remove(toNode, se);
logger.log(
String.format("StepBreakpoint at node %s timed out
and is continued exchangeId: %s", toNode,
exchange.getExchangeId()),
@@ -1194,9 +1210,9 @@ public final class DefaultBacklogDebugger extends
ServiceSupport implements Back
try {
boolean hit = se.awaitExchange(fallbackTimeout);
if (!hit) {
- // remove breakpoint as it timed out
- suspendedBreakpointMessages.remove(toNode);
- suspendedBreakpoints.remove(toNode);
+ // remove breakpoint as it timed out (only the ones of
this exchange)
+ removeMessageOf(toNode, exchange.getExchangeId());
+ suspendedBreakpoints.remove(toNode, se);
logger.log(
String.format("StepBreakpoint at node %s timed out
and is continued exchangeId: %s", toNode,
exchange.getExchangeId()),
diff --git
a/core/camel-management/src/test/java/org/apache/camel/management/BacklogDebuggerResumedBeforeSuspendTest.java
b/core/camel-management/src/test/java/org/apache/camel/management/BacklogDebuggerResumedBeforeSuspendTest.java
new file mode 100644
index 000000000000..1585f0a7ce0d
--- /dev/null
+++
b/core/camel-management/src/test/java/org/apache/camel/management/BacklogDebuggerResumedBeforeSuspendTest.java
@@ -0,0 +1,158 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.camel.management;
+
+import java.util.Set;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
+
+import org.apache.camel.builder.RouteBuilder;
+import org.apache.camel.component.mock.MockEndpoint;
+import org.apache.camel.spi.BacklogDebugger;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.condition.DisabledOnOs;
+import org.junit.jupiter.api.condition.OS;
+
+import static org.awaitility.Awaitility.await;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertNull;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+/**
+ * A breakpoint that is resumed while the exchange that matched it has not
suspended yet, and another exchange that
+ * suspends at the same node meanwhile: each exchange waits for its own
resume, and the debugger shows the exchange that
+ * is suspended.
+ */
+@DisabledOnOs(OS.AIX)
+public class BacklogDebuggerResumedBeforeSuspendTest extends
ManagementTestSupport {
+
+ private final CountDownLatch dumping = new CountDownLatch(1);
+ private final CountDownLatch dumped = new CountDownLatch(1);
+ private BacklogDebugger debugger;
+
+ @Override
+ @BeforeEach
+ public void setUp() throws Exception {
+ super.setUp();
+ debugger = context.hasService(BacklogDebugger.class);
+ debugger.enableDebugger();
+ // the exchanges of a failed test continue after this, not after the
default 300 seconds
+ debugger.setFallbackTimeout(20);
+ }
+
+ @AfterEach
+ void resumeAll() {
+ dumped.countDown();
+ if (debugger != null) {
+ debugger.resumeAll();
+ }
+ }
+
+ @Test
+ public void testResumedBeforeTheExchangeSuspended() throws Exception {
+ debugger.addBreakpoint("bar");
+ MockEndpoint mock = getMockEndpoint("mock:result");
+
+ // the first exchange matches the breakpoint, and is held while the
debugger dumps its message
+ template.sendBody("seda:start", new SlowBody());
+ assertTrue(dumping.await(10, TimeUnit.SECONDS));
+ await().atMost(5, TimeUnit.SECONDS)
+ .untilAsserted(() -> assertEquals(Set.of("bar"),
debugger.getSuspendedBreakpointNodeIds()));
+
+ // the user resumes the breakpoint, and a second exchange suspends at
it
+ debugger.resumeBreakpoint("bar");
+ template.sendBody("seda:start", "Second");
+ await().atMost(5, TimeUnit.SECONDS)
+ .untilAsserted(() ->
assertTrue(String.valueOf(debugger.dumpTracedMessagesAsXml("bar")).contains("Second")));
+ String second = debugger.getSuspendedExchangeIds().iterator().next();
+
+ // the first exchange was resumed: it continues and does not wait for
the resume of the second
+ mock.expectedBodiesReceived("Slow");
+ dumped.countDown();
+ MockEndpoint.assertIsSatisfied(10, TimeUnit.SECONDS, mock);
+
+ // the debugger still shows the second exchange, suspended at the
breakpoint
+ assertEquals(Set.of(second), debugger.getSuspendedExchangeIds());
+ String dump = debugger.dumpTracedMessagesAsXml("bar");
+ assertTrue(dump.contains(second), dump);
+ assertTrue(dump.contains("Second"), dump);
+
+ resetMocks();
+ mock.expectedBodiesReceived("Second");
+ debugger.resumeBreakpoint("bar");
+ MockEndpoint.assertIsSatisfied(10, TimeUnit.SECONDS, mock);
+ }
+
+ @Test
+ public void testTimeoutAfterTheBodyWasChangedLeavesNoMessage() throws
Exception {
+ debugger.setFallbackTimeout(2);
+ debugger.addBreakpoint("bar");
+ MockEndpoint mock = getMockEndpoint("mock:result");
+ mock.expectedBodiesReceived("Changed");
+
+ template.sendBody("seda:start", "Hello");
+ await().atMost(5, TimeUnit.SECONDS)
+ .untilAsserted(() -> assertEquals(Set.of("bar"),
debugger.getSuspendedBreakpointNodeIds()));
+ // the debugger replaces the message of the node when the body is
changed
+ debugger.setMessageBodyOnBreakpoint("bar", "Changed");
+
+ // the breakpoint times out: the exchange continues, and nothing is
left at the node
+ MockEndpoint.assertIsSatisfied(10, TimeUnit.SECONDS, mock);
+ assertEquals(Set.of(), debugger.getSuspendedBreakpointNodeIds());
+ assertNull(debugger.dumpTracedMessagesAsXml("bar"));
+ }
+
+ @Override
+ protected RouteBuilder createRouteBuilder() {
+ return new RouteBuilder() {
+ @Override
+ public void configure() {
+ context.setDebugging(true);
+
+ from("seda:start?concurrentConsumers=2")
+ .setProperty("num", constant(1)).id("foo")
+ .setHeader("node", constant("bar")).id("bar")
+ .convertBodyTo(String.class)
+ .to("mock:result");
+ }
+ };
+ }
+
+ /**
+ * A body that holds the dump of its message by the debugger at the
breakpoint (its conversion to a String).
+ */
+ private final class SlowBody {
+ private final AtomicBoolean first = new AtomicBoolean(true);
+
+ @Override
+ public String toString() {
+ // only once this exchange matched the breakpoint at bar (the
debugger dumps the message there)
+ if (debugger.getSuspendedBreakpointNodeIds().contains("bar") &&
first.compareAndSet(true, false)) {
+ dumping.countDown();
+ try {
+ dumped.await(20, TimeUnit.SECONDS);
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ }
+ }
+ return "Slow";
+ }
+ }
+}
