CAMEL-24003: Add live activity tracking, CLI command, and TUI Activity tab

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 void traceEvent(BacklogTracerEventMessage event) {
 
         // 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);
             }
         }
 

docs/user-manual/modules/ROOT/pages/jbang-commands/camel-jbang-get-activity.adoc

@@ -0,0 +1,32 @@
+
+// AUTO-GENERATED by camel-package-maven-plugin - DO NOT EDIT THIS FILE
+= camel get activity
+
+Get recent completed exchange activity
+
+
+== Usage
+
+[source,bash]
+----
+camel get activity [options]
+----
+
+
+
+== Options
+
+[cols="2,5,1,2",options="header"]
+|===
+| Option | Description | Default | Type
+| `--filter` | Filter activity by route ID |  | String
+| `--json` | Output in JSON Format |  | boolean
+| `--limit` | Filter activity by limiting to the given number of rows |  | int
+| `--short-uri` | List endpoint URI without query parameters (short) |  | boolean
+| `--sort` | Sort by pid, name, age, elapsed, or since | since | String
+| `--watch` | Execute periodically and showing output fullscreen |  | boolean
+| `--wide-uri` | List endpoint URI in full details |  | boolean
+| `-h,--help` | Display the help and sub-commands |  | boolean
+|===
+
+

docs/user-manual/modules/ROOT/pages/jbang-commands/camel-jbang-get.adoc

@@ -18,6 +18,7 @@ camel get [options]
 [cols="2,5",options="header"]
 |===
 | Subcommand | Description
+| xref:jbang-commands/camel-jbang-get-activity.adoc[activity] | Get recent completed exchange activity
 | xref:jbang-commands/camel-jbang-get-bean.adoc[bean] | List beans in a running Camel integration
 | xref:jbang-commands/camel-jbang-get-blocked.adoc[blocked] | Get blocked messages of Camel integrations
 | xref:jbang-commands/camel-jbang-get-circuit-breaker.adoc[circuit-breaker] | Get status of Circuit Breaker EIPs

dsl/camel-cli-connector/src/main/java/org/apache/camel/cli/connector/LocalCliConnector.java

@@ -130,6 +130,7 @@ public class LocalCliConnector extends ServiceSupport implements CliConnector, C
     private File errorFile;
     private File receiveFile;
     private long receiveFilePos; // keep track of receive offset
+    private File activityFile;
     private byte[] lastSource;
     private ExpressionDefinition lastSourceExpression;
 
@@ -202,6 +203,7 @@ protected void doStart() throws Exception {
             errorFile = createLockFile(lockFile.getName() + "-error.json");
             debugFile = createLockFile(lockFile.getName() + "-debug.json");
             receiveFile = createLockFile(lockFile.getName() + "-receive.json");
+            activityFile = createLockFile(lockFile.getName() + "-activity.json");
             scheduledFuture = executor.scheduleWithFixedDelay(this::task, 0, delay, TimeUnit.MILLISECONDS);
             LOG.info("Camel CLI connector enabled");
         } else {
@@ -1693,6 +1695,20 @@ protected void traceTask() {
             LOG.trace("Error updating receive file: {} due to: {}. This exception is ignored.",
                     receiveFile, e.getMessage(), e);
         }
+        try {
+            DevConsole dc15 = camelContext.getCamelContextExtension().getContextPlugin(DevConsoleRegistry.class)
+                    .resolveById("activity");
+            if (dc15 != null) {
+                JsonObject json = (JsonObject) dc15.call(DevConsole.MediaType.JSON);
+                LOG.trace("Updating activity file: {}", activityFile);
+                String data = json.toJson() + System.lineSeparator();
+                IOHelper.writeText(data, activityFile);
+            }
+        } catch (Exception e) {
+            // ignore
+            LOG.trace("Error updating activity file: {} due to: {}. This exception is ignored.",
+                    activityFile, e.getMessage(), e);
+        }
     }
 
     private JsonObject collectMemory() {
@@ -1853,6 +1869,9 @@ protected void doStop() throws Exception {
         if (receiveFile != null) {
             FileUtil.deleteFile(receiveFile);
         }
+        if (activityFile != null) {
+            FileUtil.deleteFile(activityFile);
+        }
         if (executor != null) {
             camelContext.getExecutorServiceManager().shutdown(executor);
             executor = null;