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 650372c7b1de CAMEL-24003: Add activity queue to BacklogTracer with
EventNotifier
650372c7b1de is described below
commit 650372c7b1de64868e1076575116ba8a1c4f2c3e
Author: Claus Ibsen <[email protected]>
AuthorDate: Sat Jul 11 23:54:47 2026 +0200
CAMEL-24003: Add activity queue to BacklogTracer with EventNotifier
Adds a rolling window of recent exchange lifecycle activity to BacklogTracer
for live monitoring dashboards. Uses EventNotifier to capture exchange
created,
sent, completed, and failed events. Introduces BacklogTracerActivityMessage
as
a lightweight API (no body/headers) with EndpointSend entries tracking which
remote endpoints each exchange called. Bounded queue (default 100) with
ConcurrentHashMap-based in-flight tracking and 5-minute stale eviction.
Closes #24598
Co-Authored-By: Claude Opus 4.6 <[email protected]>
---
.../apache/camel/catalog/dev-consoles.properties | 1 +
.../camel/catalog/dev-consoles/activity.json | 15 ++
.../main/camel-main-configuration-metadata.json | 1 +
.../java/org/apache/camel/spi/BacklogTracer.java | 43 ++++
.../camel/spi/BacklogTracerActivityMessage.java | 103 ++++++++++
.../apache/camel/impl/debugger/BacklogTracer.java | 218 +++++++++++++++++++++
.../DefaultBacklogTracerActivityMessage.java | 139 +++++++++++++
.../apache/camel/impl/engine/DefaultChannel.java | 5 +
.../org/apache/camel/dev-console/activity.json | 15 ++
.../services/org/apache/camel/dev-console/activity | 2 +
.../org/apache/camel/dev-consoles.properties | 2 +-
.../camel/impl/console/ActivityDevConsole.java | 108 ++++++++++
.../TracerConfigurationPropertiesConfigurer.java | 7 +
.../camel-main-configuration-metadata.json | 1 +
core/camel-main/src/main/docs/main.adoc | 3 +-
.../org/apache/camel/main/BaseMainSupport.java | 1 +
.../camel/main/TracerConfigurationProperties.java | 23 +++
.../mbean/ManagedBacklogTracerMBean.java | 9 +
.../management/mbean/ManagedBacklogTracer.java | 15 ++
.../management/BacklogTracerActivityTest.java | 173 ++++++++++++++++
20 files changed, 882 insertions(+), 2 deletions(-)
diff --git
a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/dev-consoles.properties
b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/dev-consoles.properties
index 92fda01cffe3..ade3587e222b 100644
---
a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/dev-consoles.properties
+++
b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/dev-consoles.properties
@@ -1,3 +1,4 @@
+activity
aws-secrets
aws2-s3
azure-secrets
diff --git
a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/dev-consoles/activity.json
b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/dev-consoles/activity.json
new file mode 100644
index 000000000000..1c8f94954a0e
--- /dev/null
+++
b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/dev-consoles/activity.json
@@ -0,0 +1,15 @@
+{
+ "console": {
+ "kind": "console",
+ "group": "camel",
+ "name": "activity",
+ "title": "Camel Activity",
+ "description": "Recent completed exchange activity",
+ "deprecated": false,
+ "javaType": "org.apache.camel.impl.console.ActivityDevConsole",
+ "groupId": "org.apache.camel",
+ "artifactId": "camel-console",
+ "version": "4.22.0-SNAPSHOT"
+ }
+}
+
diff --git
a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/main/camel-main-configuration-metadata.json
b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/main/camel-main-configuration-metadata.json
index 034183e3dda7..9c5e7a7718e5 100644
---
a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/main/camel-main-configuration-metadata.json
+++
b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/main/camel-main-configuration-metadata.json
@@ -483,6 +483,7 @@
{ "name": "camel.threadpool.poolSize", "required": false, "description":
"Sets the default core pool size (threads to keep minimum in pool)",
"sourceType": "org.apache.camel.main.ThreadPoolConfigurationProperties",
"type": "integer", "javaType": "java.lang.Integer", "secret": false },
{ "name": "camel.threadpool.rejectedPolicy", "required": false,
"description": "Sets the default handler for tasks which cannot be executed by
the thread pool.", "sourceType":
"org.apache.camel.main.ThreadPoolConfigurationProperties", "type": "enum",
"javaType": "org.apache.camel.util.concurrent.ThreadPoolRejectedPolicy",
"secret": false, "enum": [ "Abort", "CallerRuns", "DiscardOldest", "Discard" ]
},
{ "name": "camel.threadpool.timeUnit", "required": false, "description":
"Sets the default time unit used for keep alive time", "sourceType":
"org.apache.camel.main.ThreadPoolConfigurationProperties", "type": "object",
"javaType": "java.util.concurrent.TimeUnit", "secret": false },
+ { "name": "camel.trace.activitySize", "required": false, "description":
"Defines how many completed exchange summaries to keep in the activity queue
(should be between 1 - 1000). The activity queue captures lightweight metadata
(no body or headers) each time an exchange completes a route, providing a
rolling window of recent exchange activity.", "sourceType":
"org.apache.camel.main.TracerConfigurationProperties", "type": "integer",
"javaType": "int", "defaultValue": 100, "secret": false },
{ "name": "camel.trace.backlogSize", "required": false, "description":
"Defines how many of the last messages to keep in the tracer (should be between
1 - 1000).", "sourceType":
"org.apache.camel.main.TracerConfigurationProperties", "type": "integer",
"javaType": "int", "defaultValue": 100, "secret": false },
{ "name": "camel.trace.bodyIncludeFiles", "required": false,
"description": "Whether to include the message body of file based messages. The
overhead is that the file content has to be read from the file.", "sourceType":
"org.apache.camel.main.TracerConfigurationProperties", "type": "boolean",
"javaType": "boolean", "defaultValue": true, "secret": false },
{ "name": "camel.trace.bodyIncludeStreams", "required": false,
"description": "Whether to include the message body of stream based messages.
If enabled then beware the stream may not be re-readable later. See more about
Stream Caching.", "sourceType":
"org.apache.camel.main.TracerConfigurationProperties", "type": "boolean",
"javaType": "boolean", "defaultValue": false, "secret": false },
diff --git
a/core/camel-api/src/main/java/org/apache/camel/spi/BacklogTracer.java
b/core/camel-api/src/main/java/org/apache/camel/spi/BacklogTracer.java
index d56acdc2dde5..5b7563ae8247 100644
--- a/core/camel-api/src/main/java/org/apache/camel/spi/BacklogTracer.java
+++ b/core/camel-api/src/main/java/org/apache/camel/spi/BacklogTracer.java
@@ -79,6 +79,24 @@ public interface BacklogTracer {
*/
void setBacklogSize(int backlogSize);
+ /**
+ * Number of completed exchange summaries to keep in the activity queue.
Default is 100.
+ * <p>
+ * The activity queue captures a lightweight summary (metadata only, no
body or headers) each time an exchange
+ * completes a route. This provides a rolling window of recent exchange
activity suitable for live monitoring.
+ *
+ * @since 4.22
+ */
+ int getActivitySize();
+
+ /**
+ * Number of completed exchange summaries to keep in the activity queue
(should be between 1 - 1000). Default is
+ * 100.
+ *
+ * @since 4.22
+ */
+ void setActivitySize(int activitySize);
+
/**
* Remove the currently traced messages when dump methods are invoked
*/
@@ -237,6 +255,31 @@ public interface BacklogTracer {
*/
Collection<BacklogTracerEventMessage> getLatestMessageHistory();
+ /**
+ * Get all activity summaries (without removing).
+ * <p>
+ * Activity summaries are lightweight snapshots captured when an exchange
completes. They contain exchange-level
+ * metadata (exchange ID, route, elapsed time, status) and a list of
remote endpoints that were called — no message
+ * body or headers.
+ *
+ * @since 4.22
+ */
+ Collection<BacklogTracerActivityMessage> getActivity();
+
+ /**
+ * Dumps all activity summaries.
+ *
+ * @since 4.22
+ */
+ List<BacklogTracerActivityMessage> dumpActivity();
+
+ /**
+ * Dumps all activity summaries as JSon.
+ *
+ * @since 4.22
+ */
+ String dumpActivityAsJSon();
+
/**
* Dumps all tracing data
*/
diff --git
a/core/camel-api/src/main/java/org/apache/camel/spi/BacklogTracerActivityMessage.java
b/core/camel-api/src/main/java/org/apache/camel/spi/BacklogTracerActivityMessage.java
new file mode 100644
index 000000000000..94f6a7a72453
--- /dev/null
+++
b/core/camel-api/src/main/java/org/apache/camel/spi/BacklogTracerActivityMessage.java
@@ -0,0 +1,103 @@
+/*
+ * 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.spi;
+
+import java.util.List;
+
+/**
+ * Lightweight summary of a completed exchange captured by the {@link
BacklogTracer} activity queue.
+ * <p>
+ * Unlike {@link BacklogTracerEventMessage} which captures full message
snapshots (body, headers, properties) at each
+ * route node, activity messages contain only exchange-level metadata and a
list of remote endpoints that were called
+ * during the exchange lifecycle. This makes them suitable for live monitoring
dashboards without the overhead of full
+ * tracing.
+ *
+ * @since 4.22
+ */
+public interface BacklogTracerActivityMessage {
+
+ /**
+ * Unique trace counter for this activity entry.
+ */
+ long getUid();
+
+ /**
+ * Timestamp when the exchange was created.
+ */
+ long getTimestamp();
+
+ /**
+ * The exchange id.
+ */
+ String getExchangeId();
+
+ /**
+ * The route id where the exchange originated.
+ */
+ String getRouteId();
+
+ /**
+ * The endpoint URI of the route consumer (the "from" endpoint).
+ */
+ String getFromEndpointUri();
+
+ /**
+ * Total elapsed time in milliseconds for the exchange.
+ */
+ long getElapsed();
+
+ /**
+ * Whether the exchange failed.
+ */
+ boolean isFailed();
+
+ /**
+ * The exception message if the exchange failed, or null if successful.
+ */
+ String getExceptionMessage();
+
+ /**
+ * List of remote endpoints that were called during the exchange lifecycle.
+ * <p>
+ * Only remote endpoints are captured (endpoints where {@link
org.apache.camel.Endpoint#isRemote()} returns true).
+ * Internal endpoints like direct, seda, timer, and log are excluded.
+ */
+ List<EndpointSend> getEndpointSends();
+
+ /**
+ * A single remote endpoint call captured during the exchange lifecycle.
+ *
+ * @since 4.22
+ */
+ interface EndpointSend {
+
+ /**
+ * The endpoint URI that was called.
+ */
+ String getEndpointUri();
+
+ /**
+ * Whether this is a remote endpoint.
+ */
+ boolean isRemoteEndpoint();
+
+ /**
+ * Time taken in milliseconds for this endpoint call.
+ */
+ long getElapsed();
+ }
+}
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 17ac2eb04972..29e77353857a 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
@@ -22,6 +22,7 @@ import java.util.Collections;
import java.util.List;
import java.util.Objects;
import java.util.Queue;
+import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.atomic.AtomicLong;
@@ -30,14 +31,19 @@ import org.apache.camel.Endpoint;
import org.apache.camel.Exchange;
import org.apache.camel.ExchangePropertyKey;
import org.apache.camel.NamedNode;
+import org.apache.camel.NonManagedService;
import org.apache.camel.Predicate;
import org.apache.camel.Route;
+import org.apache.camel.spi.BacklogTracerActivityMessage;
import org.apache.camel.spi.BacklogTracerEventMessage;
+import org.apache.camel.spi.CamelEvent;
import org.apache.camel.spi.Language;
import org.apache.camel.support.CamelContextHelper;
import org.apache.camel.support.LoggerHelper;
import org.apache.camel.support.MessageHelper;
import org.apache.camel.support.PatternHelper;
+import org.apache.camel.support.SimpleEventNotifierSupport;
+import org.apache.camel.support.service.ServiceHelper;
import org.apache.camel.support.service.ServiceSupport;
import org.apache.camel.util.StringHelper;
import org.apache.camel.util.json.JsonArray;
@@ -66,6 +72,12 @@ public class BacklogTracer extends ServiceSupport implements
org.apache.camel.sp
// use tracer to capture additional information for capturing latest
completed exchange message-history
private final Queue<BacklogTracerEventMessage> provisionalHistoryQueue =
new LinkedBlockingQueue<>(MAX_BACKLOG_SIZE);
private final Queue<BacklogTracerEventMessage> completeHistoryQueue = new
LinkedBlockingQueue<>(MAX_BACKLOG_SIZE + 1);
+ // rolling window of completed exchange activity for live monitoring
+ private final Queue<BacklogTracerActivityMessage> activityQueue = new
LinkedBlockingQueue<>(MAX_BACKLOG_SIZE);
+ private final ConcurrentHashMap<String,
DefaultBacklogTracerActivityMessage> inflightActivity = new
ConcurrentHashMap<>();
+ private final ActivityEventNotifier activityEventNotifier = new
ActivityEventNotifier();
+ private int activitySize = 100;
+ private static final long INFLIGHT_EVICTION_MILLIS = 5 * 60 * 1000;
private volatile String lastCompletedBreadcrumbId;
private boolean removeOnDump = true;
private int bodyMaxChars = 32 * 1024;
@@ -260,6 +272,7 @@ public class BacklogTracer extends ServiceSupport
implements org.apache.camel.sp
completeHistoryQueue.add(event);
}
}
+
if (!enabled) {
return;
}
@@ -290,6 +303,15 @@ public class BacklogTracer extends ServiceSupport
implements org.apache.camel.sp
}
}
+ private void drainActivity() {
+ int drain = activityQueue.size() - activitySize + 1;
+ if (drain > 0) {
+ for (int i = 0; i < drain; i++) {
+ activityQueue.poll();
+ }
+ }
+ }
+
private boolean shouldTraceFilter(Exchange exchange) {
return predicate.matches(exchange);
}
@@ -331,6 +353,23 @@ public class BacklogTracer extends ServiceSupport
implements org.apache.camel.sp
this.backlogSize = backlogSize;
}
+ @Override
+ public int getActivitySize() {
+ return activitySize;
+ }
+
+ @Override
+ public void setActivitySize(int activitySize) {
+ if (activitySize <= 0) {
+ throw new IllegalArgumentException("The activity size must be a
positive number, was: " + activitySize);
+ }
+ if (activitySize > MAX_BACKLOG_SIZE) {
+ throw new IllegalArgumentException(
+ "The activity size cannot be greater than the max size of
" + MAX_BACKLOG_SIZE + ", was: " + activitySize);
+ }
+ this.activitySize = activitySize;
+ }
+
@Override
public boolean isRemoveOnDump() {
return removeOnDump;
@@ -478,6 +517,62 @@ public class BacklogTracer extends ServiceSupport
implements org.apache.camel.sp
return Collections.unmodifiableCollection(completeHistoryQueue);
}
+ @Override
+ public Collection<BacklogTracerActivityMessage> getActivity() {
+ return Collections.unmodifiableCollection(activityQueue);
+ }
+
+ @Override
+ public List<BacklogTracerActivityMessage> dumpActivity() {
+ List<BacklogTracerActivityMessage> answer = new
ArrayList<>(activityQueue);
+ if (isRemoveOnDump()) {
+ activityQueue.clear();
+ }
+ return answer;
+ }
+
+ @Override
+ public String dumpActivityAsJSon() {
+ List<BacklogTracerActivityMessage> events = dumpActivity();
+
+ JsonObject root = new JsonObject();
+ JsonArray arr = new JsonArray();
+ root.put("activity", arr);
+ for (BacklogTracerActivityMessage event : events) {
+ JsonObject jo = new JsonObject();
+ jo.put("uid", event.getUid());
+ jo.put("exchangeId", event.getExchangeId());
+ if (event.getRouteId() != null) {
+ jo.put("routeId", event.getRouteId());
+ }
+ if (event.getFromEndpointUri() != null) {
+ jo.put("fromEndpointUri", event.getFromEndpointUri());
+ }
+ if (event.getTimestamp() > 0) {
+ jo.put("timestamp", event.getTimestamp());
+ }
+ jo.put("elapsed", event.getElapsed());
+ jo.put("failed", event.isFailed());
+ if (event.getExceptionMessage() != null) {
+ jo.put("exception", event.getExceptionMessage());
+ }
+ List<BacklogTracerActivityMessage.EndpointSend> sends =
event.getEndpointSends();
+ if (sends != null && !sends.isEmpty()) {
+ JsonArray sa = new JsonArray();
+ for (BacklogTracerActivityMessage.EndpointSend send : sends) {
+ JsonObject so = new JsonObject();
+ so.put("endpointUri", send.getEndpointUri());
+ so.put("remoteEndpoint", send.isRemoteEndpoint());
+ so.put("elapsed", send.getElapsed());
+ sa.add(so);
+ }
+ jo.put("endpointSends", sa);
+ }
+ arr.add(jo);
+ }
+ return root.toJson();
+ }
+
public List<BacklogTracerEventMessage> dumpTracedMessages(String nodeId) {
List<BacklogTracerEventMessage> answer = new ArrayList<>();
if (nodeId != null) {
@@ -585,6 +680,8 @@ public class BacklogTracer extends ServiceSupport
implements org.apache.camel.sp
queue.clear();
completeHistoryQueue.clear();
provisionalHistoryQueue.clear();
+ activityQueue.clear();
+ inflightActivity.clear();
}
@Override
@@ -592,9 +689,130 @@ public class BacklogTracer extends ServiceSupport
implements org.apache.camel.sp
return traceCounter.incrementAndGet();
}
+ @Override
+ protected void doStart() throws Exception {
+
camelContext.getManagementStrategy().addEventNotifier(activityEventNotifier);
+ ServiceHelper.startService(activityEventNotifier);
+ }
+
@Override
protected void doStop() throws Exception {
+ ServiceHelper.stopService(activityEventNotifier);
+
camelContext.getManagementStrategy().removeEventNotifier(activityEventNotifier);
+ inflightActivity.clear();
clear();
}
+ private void onExchangeCreated(CamelEvent.ExchangeCreatedEvent event) {
+ evictStaleInflight();
+
+ Exchange exchange = event.getExchange();
+ String exchangeId = exchange.getExchangeId();
+ long timestamp = event.getTimestamp() > 0 ? event.getTimestamp() :
System.currentTimeMillis();
+
+ DefaultBacklogTracerActivityMessage activity = new
DefaultBacklogTracerActivityMessage(
+ incrementTraceCounter(), timestamp, exchangeId);
+ inflightActivity.put(exchangeId, activity);
+ }
+
+ private void onExchangeSent(CamelEvent.ExchangeSentEvent event) {
+ Endpoint endpoint = event.getEndpoint();
+ if (!endpoint.isRemote()) {
+ return;
+ }
+
+ String exchangeId = event.getExchange().getExchangeId();
+ DefaultBacklogTracerActivityMessage activity =
inflightActivity.get(exchangeId);
+ if (activity != null) {
+ activity.addEndpointSend(endpoint.getEndpointUri(),
endpoint.isRemote(), event.getTimeTaken());
+ }
+ }
+
+ private void onExchangeCompleted(CamelEvent.ExchangeCompletedEvent event) {
+ Exchange exchange = event.getExchange();
+ String exchangeId = exchange.getExchangeId();
+ DefaultBacklogTracerActivityMessage activity =
inflightActivity.remove(exchangeId);
+ if (activity != null) {
+ String routeId = exchange.getFromRouteId();
+ String fromEndpointUri = resolveFromEndpointUri(routeId);
+ activity.complete(routeId, fromEndpointUri,
exchange.getClock().elapsed(), false, null);
+ drainActivity();
+ activityQueue.offer(activity);
+ }
+ }
+
+ private void onExchangeFailed(CamelEvent.ExchangeFailedEvent event) {
+ Exchange exchange = event.getExchange();
+ String exchangeId = exchange.getExchangeId();
+ DefaultBacklogTracerActivityMessage activity =
inflightActivity.remove(exchangeId);
+ if (activity != null) {
+ String exMessage = null;
+ Exception cause = exchange.getException();
+ if (cause != null) {
+ exMessage = cause.getMessage();
+ }
+ String routeId = exchange.getFromRouteId();
+ String fromEndpointUri = resolveFromEndpointUri(routeId);
+ activity.complete(routeId, fromEndpointUri,
exchange.getClock().elapsed(), true, exMessage);
+ drainActivity();
+ activityQueue.offer(activity);
+ }
+ }
+
+ private String resolveFromEndpointUri(String routeId) {
+ if (routeId != null) {
+ Route route = camelContext.getRoute(routeId);
+ if (route != null && route.getConsumer() != null) {
+ return route.getConsumer().getEndpoint().getEndpointUri();
+ }
+ }
+ return null;
+ }
+
+ private void evictStaleInflight() {
+ if (inflightActivity.isEmpty()) {
+ return;
+ }
+ long cutoff = System.currentTimeMillis() - INFLIGHT_EVICTION_MILLIS;
+ inflightActivity.entrySet().removeIf(e -> e.getValue().getTimestamp()
< cutoff);
+ }
+
+ /**
+ * Inner EventNotifier that captures exchange lifecycle events for
activity monitoring.
+ */
+ private class ActivityEventNotifier extends SimpleEventNotifierSupport
implements NonManagedService {
+
+ ActivityEventNotifier() {
+ // enable the exchange events we need for activity tracking
+ setIgnoreExchangeEvents(false);
+ setIgnoreExchangeCreatedEvent(false);
+ setIgnoreExchangeCompletedEvent(false);
+ setIgnoreExchangeFailedEvents(false);
+ setIgnoreExchangeSentEvents(false);
+ // ignore all non-exchange events
+ setIgnoreCamelContextInitEvents(true);
+ setIgnoreCamelContextEvents(true);
+ setIgnoreRouteEvents(true);
+ setIgnoreServiceEvents(true);
+ setIgnoreStepEvents(true);
+ }
+
+ @Override
+ public boolean isDisabled() {
+ return !enabled && !standby;
+ }
+
+ @Override
+ public void notify(CamelEvent event) throws Exception {
+ if (event instanceof CamelEvent.ExchangeCreatedEvent ece) {
+ onExchangeCreated(ece);
+ } else if (event instanceof CamelEvent.ExchangeSentEvent ese) {
+ onExchangeSent(ese);
+ } else if (event instanceof CamelEvent.ExchangeCompletedEvent ece)
{
+ onExchangeCompleted(ece);
+ } else if (event instanceof CamelEvent.ExchangeFailedEvent efe) {
+ onExchangeFailed(efe);
+ }
+ }
+ }
}
diff --git
a/core/camel-base-engine/src/main/java/org/apache/camel/impl/debugger/DefaultBacklogTracerActivityMessage.java
b/core/camel-base-engine/src/main/java/org/apache/camel/impl/debugger/DefaultBacklogTracerActivityMessage.java
new file mode 100644
index 000000000000..d50feac72b78
--- /dev/null
+++
b/core/camel-base-engine/src/main/java/org/apache/camel/impl/debugger/DefaultBacklogTracerActivityMessage.java
@@ -0,0 +1,139 @@
+/*
+ * 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.impl.debugger;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+
+import org.apache.camel.spi.BacklogTracerActivityMessage;
+
+/**
+ * Default implementation of {@link BacklogTracerActivityMessage}.
+ */
+public final class DefaultBacklogTracerActivityMessage implements
BacklogTracerActivityMessage {
+
+ private final long uid;
+ private final long timestamp;
+ private final String exchangeId;
+ private String routeId;
+ private String fromEndpointUri;
+ private final List<DefaultEndpointSend> endpointSends = new ArrayList<>();
+ private long elapsed;
+ private boolean failed;
+ private String exceptionMessage;
+
+ public DefaultBacklogTracerActivityMessage(long uid, long timestamp,
String exchangeId) {
+ this.uid = uid;
+ this.timestamp = timestamp;
+ this.exchangeId = exchangeId;
+ }
+
+ @Override
+ public long getUid() {
+ return uid;
+ }
+
+ @Override
+ public long getTimestamp() {
+ return timestamp;
+ }
+
+ @Override
+ public String getExchangeId() {
+ return exchangeId;
+ }
+
+ @Override
+ public String getRouteId() {
+ return routeId;
+ }
+
+ @Override
+ public String getFromEndpointUri() {
+ return fromEndpointUri;
+ }
+
+ @Override
+ public long getElapsed() {
+ return elapsed;
+ }
+
+ @Override
+ public boolean isFailed() {
+ return failed;
+ }
+
+ @Override
+ public String getExceptionMessage() {
+ return exceptionMessage;
+ }
+
+ @Override
+ public List<EndpointSend> getEndpointSends() {
+ synchronized (endpointSends) {
+ return Collections.unmodifiableList(new
ArrayList<>(endpointSends));
+ }
+ }
+
+ public void addEndpointSend(String endpointUri, boolean remoteEndpoint,
long elapsed) {
+ synchronized (endpointSends) {
+ endpointSends.add(new DefaultEndpointSend(endpointUri,
remoteEndpoint, elapsed));
+ }
+ }
+
+ public void complete(
+ String routeId, String fromEndpointUri,
+ long elapsed, boolean failed, String exceptionMessage) {
+ this.routeId = routeId;
+ this.fromEndpointUri = fromEndpointUri;
+ this.elapsed = elapsed;
+ this.failed = failed;
+ this.exceptionMessage = exceptionMessage;
+ }
+
+ /**
+ * Default implementation of {@link EndpointSend}.
+ */
+ public static final class DefaultEndpointSend implements EndpointSend {
+
+ private final String endpointUri;
+ private final boolean remoteEndpoint;
+ private final long elapsed;
+
+ public DefaultEndpointSend(String endpointUri, boolean remoteEndpoint,
long elapsed) {
+ this.endpointUri = endpointUri;
+ this.remoteEndpoint = remoteEndpoint;
+ this.elapsed = elapsed;
+ }
+
+ @Override
+ public String getEndpointUri() {
+ return endpointUri;
+ }
+
+ @Override
+ public boolean isRemoteEndpoint() {
+ return remoteEndpoint;
+ }
+
+ @Override
+ public long getElapsed() {
+ return elapsed;
+ }
+ }
+}
diff --git
a/core/camel-base-engine/src/main/java/org/apache/camel/impl/engine/DefaultChannel.java
b/core/camel-base-engine/src/main/java/org/apache/camel/impl/engine/DefaultChannel.java
index 432612f68c93..2efa27c39a68 100644
---
a/core/camel-base-engine/src/main/java/org/apache/camel/impl/engine/DefaultChannel.java
+++
b/core/camel-base-engine/src/main/java/org/apache/camel/impl/engine/DefaultChannel.java
@@ -312,6 +312,11 @@ public class DefaultChannel extends CamelInternalProcessor
implements Channel {
tracer.setTraceTemplates(camelContext.isBacklogTracingTemplates());
tracer.setTraceRests(camelContext.isBacklogTracingRests());
camelContext.getCamelContextExtension().addContextPlugin(BacklogTracer.class,
tracer);
+ try {
+ camelContext.addService(tracer);
+ } catch (Exception e) {
+ throw new RuntimeException(e);
+ }
}
return tracer;
}
diff --git
a/core/camel-console/src/generated/resources/META-INF/org/apache/camel/dev-console/activity.json
b/core/camel-console/src/generated/resources/META-INF/org/apache/camel/dev-console/activity.json
new file mode 100644
index 000000000000..1c8f94954a0e
--- /dev/null
+++
b/core/camel-console/src/generated/resources/META-INF/org/apache/camel/dev-console/activity.json
@@ -0,0 +1,15 @@
+{
+ "console": {
+ "kind": "console",
+ "group": "camel",
+ "name": "activity",
+ "title": "Camel Activity",
+ "description": "Recent completed exchange activity",
+ "deprecated": false,
+ "javaType": "org.apache.camel.impl.console.ActivityDevConsole",
+ "groupId": "org.apache.camel",
+ "artifactId": "camel-console",
+ "version": "4.22.0-SNAPSHOT"
+ }
+}
+
diff --git
a/core/camel-console/src/generated/resources/META-INF/services/org/apache/camel/dev-console/activity
b/core/camel-console/src/generated/resources/META-INF/services/org/apache/camel/dev-console/activity
new file mode 100644
index 000000000000..fe3ec9d28204
--- /dev/null
+++
b/core/camel-console/src/generated/resources/META-INF/services/org/apache/camel/dev-console/activity
@@ -0,0 +1,2 @@
+# Generated by camel build tools - do NOT edit this file!
+class=org.apache.camel.impl.console.ActivityDevConsole
diff --git
a/core/camel-console/src/generated/resources/META-INF/services/org/apache/camel/dev-consoles.properties
b/core/camel-console/src/generated/resources/META-INF/services/org/apache/camel/dev-consoles.properties
index 4f7ab2e10103..95209ff8cce9 100644
---
a/core/camel-console/src/generated/resources/META-INF/services/org/apache/camel/dev-consoles.properties
+++
b/core/camel-console/src/generated/resources/META-INF/services/org/apache/camel/dev-consoles.properties
@@ -1,5 +1,5 @@
# Generated by camel build tools - do NOT edit this file!
-dev-consoles=bean blocked browse circuit-breaker consumer context datasource
debug endpoint errors eval-language event gc health heap-histogram inflight
internal-tasks java-security jfr-memory-leak jvm log memory message-history
processor producer properties receive reload rest rest-spec route
route-controller route-dump route-group route-structure route-topology send
service simple-language source sql-query sql-trace startup-recorder
system-properties thread top trace transformers type- [...]
+dev-consoles=activity bean blocked browse circuit-breaker consumer context
datasource debug endpoint errors eval-language event gc health heap-histogram
inflight internal-tasks java-security jfr-memory-leak jvm log memory
message-history processor producer properties receive reload rest rest-spec
route route-controller route-dump route-group route-structure route-topology
send service simple-language source sql-query sql-trace startup-recorder
system-properties thread top trace transform [...]
groupId=org.apache.camel
artifactId=camel-console
version=4.22.0-SNAPSHOT
diff --git
a/core/camel-console/src/main/java/org/apache/camel/impl/console/ActivityDevConsole.java
b/core/camel-console/src/main/java/org/apache/camel/impl/console/ActivityDevConsole.java
new file mode 100644
index 000000000000..4082259daafa
--- /dev/null
+++
b/core/camel-console/src/main/java/org/apache/camel/impl/console/ActivityDevConsole.java
@@ -0,0 +1,108 @@
+/*
+ * 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.impl.console;
+
+import java.util.List;
+import java.util.Map;
+
+import org.apache.camel.spi.BacklogTracer;
+import org.apache.camel.spi.BacklogTracerActivityMessage;
+import org.apache.camel.spi.annotations.DevConsole;
+import org.apache.camel.support.console.AbstractDevConsole;
+import org.apache.camel.util.json.JsonArray;
+import org.apache.camel.util.json.JsonObject;
+
+@DevConsole(name = "activity", displayName = "Camel Activity", description =
"Recent completed exchange activity")
+public class ActivityDevConsole extends AbstractDevConsole {
+
+ public ActivityDevConsole() {
+ super("camel", "activity", "Camel Activity", "Recent completed
exchange activity");
+ }
+
+ @Override
+ protected String doCallText(Map<String, Object> options) {
+ StringBuilder sb = new StringBuilder();
+
+ BacklogTracer tracer =
getCamelContext().getCamelContextExtension().getContextPlugin(BacklogTracer.class);
+ if (tracer != null) {
+ sb.append("Activity Size:
").append(tracer.getActivitySize()).append("\n");
+ for (BacklogTracerActivityMessage event : tracer.getActivity()) {
+ sb.append(String.format(" %s | %s | %s | %dms | %s",
+ event.getExchangeId(),
+ event.getRouteId(),
+ event.getFromEndpointUri() != null ?
event.getFromEndpointUri() : "",
+ event.getElapsed(),
+ event.isFailed() ? "FAILED" : "OK"));
+ List<BacklogTracerActivityMessage.EndpointSend> sends =
event.getEndpointSends();
+ if (sends != null && !sends.isEmpty()) {
+ sb.append(" | sent: ").append(sends.size());
+ }
+ sb.append("\n");
+ }
+ }
+
+ return sb.toString();
+ }
+
+ @Override
+ protected JsonObject doCallJson(Map<String, Object> options) {
+ JsonObject root = new JsonObject();
+
+ BacklogTracer tracer =
getCamelContext().getCamelContextExtension().getContextPlugin(BacklogTracer.class);
+ if (tracer != null) {
+ root.put("activitySize", tracer.getActivitySize());
+
+ JsonArray arr = new JsonArray();
+ root.put("activity", arr);
+ for (BacklogTracerActivityMessage event : tracer.getActivity()) {
+ JsonObject jo = new JsonObject();
+ jo.put("uid", event.getUid());
+ jo.put("exchangeId", event.getExchangeId());
+ if (event.getRouteId() != null) {
+ jo.put("routeId", event.getRouteId());
+ }
+ if (event.getFromEndpointUri() != null) {
+ jo.put("fromEndpointUri", event.getFromEndpointUri());
+ }
+ if (event.getTimestamp() > 0) {
+ jo.put("timestamp", event.getTimestamp());
+ }
+ jo.put("elapsed", event.getElapsed());
+ jo.put("failed", event.isFailed());
+ if (event.getExceptionMessage() != null) {
+ jo.put("exception", event.getExceptionMessage());
+ }
+ List<BacklogTracerActivityMessage.EndpointSend> sends =
event.getEndpointSends();
+ if (sends != null && !sends.isEmpty()) {
+ JsonArray sa = new JsonArray();
+ for (BacklogTracerActivityMessage.EndpointSend send :
sends) {
+ JsonObject so = new JsonObject();
+ so.put("endpointUri", send.getEndpointUri());
+ so.put("remoteEndpoint", send.isRemoteEndpoint());
+ so.put("elapsed", send.getElapsed());
+ sa.add(so);
+ }
+ jo.put("endpointSends", sa);
+ }
+ arr.add(jo);
+ }
+ }
+
+ return root;
+ }
+
+}
diff --git
a/core/camel-main/src/generated/java/org/apache/camel/main/TracerConfigurationPropertiesConfigurer.java
b/core/camel-main/src/generated/java/org/apache/camel/main/TracerConfigurationPropertiesConfigurer.java
index 2c7cf5cafed8..811931a0aad8 100644
---
a/core/camel-main/src/generated/java/org/apache/camel/main/TracerConfigurationPropertiesConfigurer.java
+++
b/core/camel-main/src/generated/java/org/apache/camel/main/TracerConfigurationPropertiesConfigurer.java
@@ -22,6 +22,7 @@ public class TracerConfigurationPropertiesConfigurer extends
org.apache.camel.su
private static final Map<String, Object> ALL_OPTIONS;
static {
Map<String, Object> map = new CaseInsensitiveMap();
+ map.put("ActivitySize", int.class);
map.put("BacklogSize", int.class);
map.put("BodyIncludeFiles", boolean.class);
map.put("BodyIncludeStreams", boolean.class);
@@ -43,6 +44,8 @@ public class TracerConfigurationPropertiesConfigurer extends
org.apache.camel.su
public boolean configure(CamelContext camelContext, Object obj, String
name, Object value, boolean ignoreCase) {
org.apache.camel.main.TracerConfigurationProperties target =
(org.apache.camel.main.TracerConfigurationProperties) obj;
switch (ignoreCase ? name.toLowerCase() : name) {
+ case "activitysize":
+ case "activitySize": target.setActivitySize(property(camelContext,
int.class, value)); return true;
case "backlogsize":
case "backlogSize": target.setBacklogSize(property(camelContext,
int.class, value)); return true;
case "bodyincludefiles":
@@ -81,6 +84,8 @@ public class TracerConfigurationPropertiesConfigurer extends
org.apache.camel.su
@Override
public Class<?> getOptionType(String name, boolean ignoreCase) {
switch (ignoreCase ? name.toLowerCase() : name) {
+ case "activitysize":
+ case "activitySize": return int.class;
case "backlogsize":
case "backlogSize": return int.class;
case "bodyincludefiles":
@@ -115,6 +120,8 @@ public class TracerConfigurationPropertiesConfigurer
extends org.apache.camel.su
public Object getOptionValue(Object obj, String name, boolean ignoreCase) {
org.apache.camel.main.TracerConfigurationProperties target =
(org.apache.camel.main.TracerConfigurationProperties) obj;
switch (ignoreCase ? name.toLowerCase() : name) {
+ case "activitysize":
+ case "activitySize": return target.getActivitySize();
case "backlogsize":
case "backlogSize": return target.getBacklogSize();
case "bodyincludefiles":
diff --git
a/core/camel-main/src/generated/resources/META-INF/camel-main-configuration-metadata.json
b/core/camel-main/src/generated/resources/META-INF/camel-main-configuration-metadata.json
index 034183e3dda7..9c5e7a7718e5 100644
---
a/core/camel-main/src/generated/resources/META-INF/camel-main-configuration-metadata.json
+++
b/core/camel-main/src/generated/resources/META-INF/camel-main-configuration-metadata.json
@@ -483,6 +483,7 @@
{ "name": "camel.threadpool.poolSize", "required": false, "description":
"Sets the default core pool size (threads to keep minimum in pool)",
"sourceType": "org.apache.camel.main.ThreadPoolConfigurationProperties",
"type": "integer", "javaType": "java.lang.Integer", "secret": false },
{ "name": "camel.threadpool.rejectedPolicy", "required": false,
"description": "Sets the default handler for tasks which cannot be executed by
the thread pool.", "sourceType":
"org.apache.camel.main.ThreadPoolConfigurationProperties", "type": "enum",
"javaType": "org.apache.camel.util.concurrent.ThreadPoolRejectedPolicy",
"secret": false, "enum": [ "Abort", "CallerRuns", "DiscardOldest", "Discard" ]
},
{ "name": "camel.threadpool.timeUnit", "required": false, "description":
"Sets the default time unit used for keep alive time", "sourceType":
"org.apache.camel.main.ThreadPoolConfigurationProperties", "type": "object",
"javaType": "java.util.concurrent.TimeUnit", "secret": false },
+ { "name": "camel.trace.activitySize", "required": false, "description":
"Defines how many completed exchange summaries to keep in the activity queue
(should be between 1 - 1000). The activity queue captures lightweight metadata
(no body or headers) each time an exchange completes a route, providing a
rolling window of recent exchange activity.", "sourceType":
"org.apache.camel.main.TracerConfigurationProperties", "type": "integer",
"javaType": "int", "defaultValue": 100, "secret": false },
{ "name": "camel.trace.backlogSize", "required": false, "description":
"Defines how many of the last messages to keep in the tracer (should be between
1 - 1000).", "sourceType":
"org.apache.camel.main.TracerConfigurationProperties", "type": "integer",
"javaType": "int", "defaultValue": 100, "secret": false },
{ "name": "camel.trace.bodyIncludeFiles", "required": false,
"description": "Whether to include the message body of file based messages. The
overhead is that the file content has to be read from the file.", "sourceType":
"org.apache.camel.main.TracerConfigurationProperties", "type": "boolean",
"javaType": "boolean", "defaultValue": true, "secret": false },
{ "name": "camel.trace.bodyIncludeStreams", "required": false,
"description": "Whether to include the message body of stream based messages.
If enabled then beware the stream may not be re-readable later. See more about
Stream Caching.", "sourceType":
"org.apache.camel.main.TracerConfigurationProperties", "type": "boolean",
"javaType": "boolean", "defaultValue": false, "secret": false },
diff --git a/core/camel-main/src/main/docs/main.adoc
b/core/camel-main/src/main/docs/main.adoc
index 20d29d34c351..f19111d9f174 100644
--- a/core/camel-main/src/main/docs/main.adoc
+++ b/core/camel-main/src/main/docs/main.adoc
@@ -282,11 +282,12 @@ The camel.debug supports 15 options, which are listed
below.
=== Camel Tracer configurations
-The camel.trace supports 14 options, which are listed below.
+The camel.trace supports 15 options, which are listed below.
[width="100%",cols="2,5,^1,2",options="header"]
|===
| Name | Description | Default | Type
+| *camel.trace.activitySize* | Defines how many completed exchange summaries
to keep in the activity queue (should be between 1 - 1000). The activity queue
captures lightweight metadata (no body or headers) each time an exchange
completes a route, providing a rolling window of recent exchange activity. |
100 | int
| *camel.trace.backlogSize* | Defines how many of the last messages to keep in
the tracer (should be between 1 - 1000). | 100 | int
| *camel.trace.bodyIncludeFiles* | Whether to include the message body of file
based messages. The overhead is that the file content has to be read from the
file. | true | boolean
| *camel.trace.bodyIncludeStreams* | Whether to include the message body of
stream based messages. If enabled then beware the stream may not be re-readable
later. See more about Stream Caching. | false | boolean
diff --git
a/core/camel-main/src/main/java/org/apache/camel/main/BaseMainSupport.java
b/core/camel-main/src/main/java/org/apache/camel/main/BaseMainSupport.java
index b933427bf2c7..26ce5b2e0bb7 100644
--- a/core/camel-main/src/main/java/org/apache/camel/main/BaseMainSupport.java
+++ b/core/camel-main/src/main/java/org/apache/camel/main/BaseMainSupport.java
@@ -2503,6 +2503,7 @@ public abstract class BaseMainSupport extends BaseService
{
tracer.setEnabled(config.isEnabled());
tracer.setStandby(config.isStandby());
tracer.setBacklogSize(config.getBacklogSize());
+ tracer.setActivitySize(config.getActivitySize());
tracer.setRemoveOnDump(config.isRemoveOnDump());
tracer.setBodyMaxChars(config.getBodyMaxChars());
tracer.setBodyIncludeStreams(config.isBodyIncludeStreams());
diff --git
a/core/camel-main/src/main/java/org/apache/camel/main/TracerConfigurationProperties.java
b/core/camel-main/src/main/java/org/apache/camel/main/TracerConfigurationProperties.java
index 753eb5f0f2c3..de5e6939c335 100644
---
a/core/camel-main/src/main/java/org/apache/camel/main/TracerConfigurationProperties.java
+++
b/core/camel-main/src/main/java/org/apache/camel/main/TracerConfigurationProperties.java
@@ -34,6 +34,8 @@ public class TracerConfigurationProperties implements
BootstrapCloseable {
private boolean standby;
@Metadata(label = "advanced", defaultValue = "100")
private int backlogSize = 100;
+ @Metadata(label = "advanced", defaultValue = "100")
+ private int activitySize = 100;
@Metadata(label = "advanced", defaultValue = "true")
private boolean removeOnDump = true;
@Metadata(label = "advanced", defaultValue = "32768")
@@ -104,6 +106,19 @@ public class TracerConfigurationProperties implements
BootstrapCloseable {
this.backlogSize = backlogSize;
}
+ public int getActivitySize() {
+ return activitySize;
+ }
+
+ /**
+ * Defines how many completed exchange summaries to keep in the activity
queue (should be between 1 - 1000). The
+ * activity queue captures lightweight metadata (no body or headers) each
time an exchange completes a route,
+ * providing a rolling window of recent exchange activity.
+ */
+ public void setActivitySize(int activitySize) {
+ this.activitySize = activitySize;
+ }
+
public boolean isRemoveOnDump() {
return removeOnDump;
}
@@ -253,6 +268,14 @@ public class TracerConfigurationProperties implements
BootstrapCloseable {
return this;
}
+ /**
+ * Defines how many completed exchange summaries to keep in the activity
queue (should be between 1 - 1000).
+ */
+ public TracerConfigurationProperties withActivitySize(int activitySize) {
+ this.activitySize = activitySize;
+ return this;
+ }
+
/**
* Whether all traced messages should be removed when the tracer is
dumping. By default, the messages are removed,
* which means that dumping will not contain previous dumped messages.
diff --git
a/core/camel-management-api/src/main/java/org/apache/camel/api/management/mbean/ManagedBacklogTracerMBean.java
b/core/camel-management-api/src/main/java/org/apache/camel/api/management/mbean/ManagedBacklogTracerMBean.java
index d254c9a27c6f..2e63926b8bd7 100644
---
a/core/camel-management-api/src/main/java/org/apache/camel/api/management/mbean/ManagedBacklogTracerMBean.java
+++
b/core/camel-management-api/src/main/java/org/apache/camel/api/management/mbean/ManagedBacklogTracerMBean.java
@@ -45,6 +45,12 @@ public interface ManagedBacklogTracerMBean {
@ManagedAttribute(description = "Number of maximum traced messages in
total to keep in the backlog (FIFO queue)")
void setBacklogSize(int backlogSize);
+ @ManagedAttribute(description = "Number of completed exchange summaries to
keep in the activity queue")
+ int getActivitySize();
+
+ @ManagedAttribute(description = "Number of completed exchange summaries to
keep in the activity queue")
+ void setActivitySize(int activitySize);
+
@ManagedAttribute(description = "Whether to remove traced message from
backlog when dumping trace messages")
boolean isRemoveOnDump();
@@ -135,6 +141,9 @@ public interface ManagedBacklogTracerMBean {
@ManagedOperation(description = "Dumps latest completed exchange message
history in JSon format")
String dumpLatestMessageHistoryAsJSon();
+ @ManagedOperation(description = "Dumps completed exchange activity
summaries in JSon format")
+ String dumpActivityAsJSon();
+
@ManagedOperation(description = "Clears the backlog")
void clear();
diff --git
a/core/camel-management/src/main/java/org/apache/camel/management/mbean/ManagedBacklogTracer.java
b/core/camel-management/src/main/java/org/apache/camel/management/mbean/ManagedBacklogTracer.java
index 098d954ec52c..bd449b0d4134 100644
---
a/core/camel-management/src/main/java/org/apache/camel/management/mbean/ManagedBacklogTracer.java
+++
b/core/camel-management/src/main/java/org/apache/camel/management/mbean/ManagedBacklogTracer.java
@@ -86,6 +86,16 @@ public class ManagedBacklogTracer implements
ManagedBacklogTracerMBean {
backlogTracer.setBacklogSize(backlogSize);
}
+ @Override
+ public int getActivitySize() {
+ return backlogTracer.getActivitySize();
+ }
+
+ @Override
+ public void setActivitySize(int activitySize) {
+ backlogTracer.setActivitySize(activitySize);
+ }
+
@Override
public boolean isRemoveOnDump() {
return backlogTracer.isRemoveOnDump();
@@ -236,6 +246,11 @@ public class ManagedBacklogTracer implements
ManagedBacklogTracerMBean {
return backlogTracer.dumpLatestMessageHistoryAsJSon();
}
+ @Override
+ public String dumpActivityAsJSon() {
+ return backlogTracer.dumpActivityAsJSon();
+ }
+
@Override
public void clear() {
backlogTracer.clear();
diff --git
a/core/camel-management/src/test/java/org/apache/camel/management/BacklogTracerActivityTest.java
b/core/camel-management/src/test/java/org/apache/camel/management/BacklogTracerActivityTest.java
new file mode 100644
index 000000000000..b245b4470829
--- /dev/null
+++
b/core/camel-management/src/test/java/org/apache/camel/management/BacklogTracerActivityTest.java
@@ -0,0 +1,173 @@
+/*
+ * 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 javax.management.Attribute;
+import javax.management.MBeanServer;
+import javax.management.ObjectName;
+
+import org.apache.camel.builder.RouteBuilder;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.condition.DisabledOnOs;
+import org.junit.jupiter.api.condition.OS;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+@DisabledOnOs(OS.AIX)
+public class BacklogTracerActivityTest extends ManagementTestSupport {
+
+ @SuppressWarnings("unchecked")
+ @Test
+ public void testActivityCapture() throws Exception {
+ MBeanServer mbeanServer = getMBeanServer();
+ ObjectName on
+ = new ObjectName(
+ "org.apache.camel:context=" +
context.getManagementName() + ",type=tracer,name=BacklogTracer");
+ assertNotNull(on);
+ assertTrue(mbeanServer.isRegistered(on));
+
+ Integer activitySize = (Integer) mbeanServer.getAttribute(on,
"ActivitySize");
+ assertEquals(100, activitySize.intValue());
+
+ getMockEndpoint("mock:foo").expectedMessageCount(2);
+ getMockEndpoint("mock:bar").expectedMessageCount(2);
+
+ template.sendBody("direct:start", "Hello World");
+ template.sendBody("direct:start", "Bye World");
+
+ assertMockEndpointsSatisfied();
+
+ String json = (String) mbeanServer.invoke(on, "dumpActivityAsJSon",
null, null);
+ assertNotNull(json);
+ assertTrue(json.contains("activity"));
+ assertTrue(json.contains("exchangeId"));
+ assertTrue(json.contains("routeId"));
+ assertTrue(json.contains("elapsed"));
+ assertTrue(json.contains("\"failed\":false"));
+ assertTrue(json.contains("fromEndpointUri"));
+ }
+
+ @SuppressWarnings("unchecked")
+ @Test
+ public void testActivityOverflow() throws Exception {
+ MBeanServer mbeanServer = getMBeanServer();
+ ObjectName on
+ = new ObjectName(
+ "org.apache.camel:context=" +
context.getManagementName() + ",type=tracer,name=BacklogTracer");
+ assertNotNull(on);
+ assertTrue(mbeanServer.isRegistered(on));
+
+ // set activity size to 5
+ mbeanServer.setAttribute(on, new Attribute("ActivitySize", 5));
+ Integer activitySize = (Integer) mbeanServer.getAttribute(on,
"ActivitySize");
+ assertEquals(5, activitySize.intValue());
+
+ getMockEndpoint("mock:foo").expectedMessageCount(10);
+ getMockEndpoint("mock:bar").expectedMessageCount(10);
+
+ for (int i = 0; i < 10; i++) {
+ template.sendBody("direct:start", "Message " + i);
+ }
+
+ assertMockEndpointsSatisfied();
+
+ // with removeOnDump=true (default), dumpActivity returns and clears
+ // the activity queue should have been drained to at most 5 entries
+ String json = (String) mbeanServer.invoke(on, "dumpActivityAsJSon",
null, null);
+ assertNotNull(json);
+
+ // count exchangeId occurrences to verify bounded size
+ int count = 0;
+ int idx = 0;
+ while ((idx = json.indexOf("\"exchangeId\"", idx)) != -1) {
+ count++;
+ idx++;
+ }
+ assertEquals(5, count, "Activity queue should be bounded to
activitySize=5");
+ }
+
+ @SuppressWarnings("unchecked")
+ @Test
+ public void testActivityClearOnDump() throws Exception {
+ MBeanServer mbeanServer = getMBeanServer();
+ ObjectName on
+ = new ObjectName(
+ "org.apache.camel:context=" +
context.getManagementName() + ",type=tracer,name=BacklogTracer");
+ assertNotNull(on);
+ assertTrue(mbeanServer.isRegistered(on));
+
+ getMockEndpoint("mock:foo").expectedMessageCount(2);
+ getMockEndpoint("mock:bar").expectedMessageCount(2);
+
+ template.sendBody("direct:start", "Hello World");
+ template.sendBody("direct:start", "Bye World");
+
+ assertMockEndpointsSatisfied();
+
+ // first dump should have entries
+ String json1 = (String) mbeanServer.invoke(on, "dumpActivityAsJSon",
null, null);
+ assertNotNull(json1);
+ assertTrue(json1.contains("exchangeId"));
+
+ // second dump should be empty (removeOnDump=true by default)
+ String json2 = (String) mbeanServer.invoke(on, "dumpActivityAsJSon",
null, null);
+ assertNotNull(json2);
+ assertFalse(json2.contains("exchangeId"), "Activity should be cleared
after dump");
+ }
+
+ @SuppressWarnings("unchecked")
+ @Test
+ public void testActivityCapturesFailed() throws Exception {
+ MBeanServer mbeanServer = getMBeanServer();
+ ObjectName on
+ = new ObjectName(
+ "org.apache.camel:context=" +
context.getManagementName() + ",type=tracer,name=BacklogTracer");
+ assertNotNull(on);
+ assertTrue(mbeanServer.isRegistered(on));
+
+ getMockEndpoint("mock:foo").expectedMessageCount(1);
+ getMockEndpoint("mock:bar").expectedMessageCount(1);
+
+ template.sendBody("direct:start", "Hello World");
+
+ assertMockEndpointsSatisfied();
+
+ String json = (String) mbeanServer.invoke(on, "dumpActivityAsJSon",
null, null);
+ assertNotNull(json);
+ assertTrue(json.contains("\"failed\":false"));
+ assertTrue(json.contains("\"routeId\":\"route1\""));
+ }
+
+ @Override
+ protected RouteBuilder createRouteBuilder() {
+ return new RouteBuilder() {
+ @Override
+ public void configure() {
+ context.setUseBreadcrumb(false);
+ context.setBacklogTracing(true);
+
+ from("direct:start")
+ .to("mock:foo").id("foo")
+ .to("mock:bar").id("bar");
+ }
+ };
+ }
+
+}
