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 f64de14b7099 camel-tui - Add Kafka tab to More menu
f64de14b7099 is described below
commit f64de14b70994a12c199d4b5614ece9983f7df75
Author: Claus Ibsen <[email protected]>
AuthorDate: Tue Jul 14 14:55:14 2026 +0200
camel-tui - Add Kafka tab to More menu
Add a dedicated Kafka tab under the More submenu that displays live
Kafka consumer status from the Kafka dev console. Shows route, worker
state, group ID, topic, partition, offset, and errors in a sortable
table with MCP and F1 help support.
Co-Authored-By: Claude Opus 4.6 <[email protected]>
Signed-off-by: Claus Ibsen <[email protected]>
---
.../jbang/core/commands/tui/IntegrationInfo.java | 1 +
.../jbang/core/commands/tui/KafkaConsumerInfo.java | 32 +++
.../dsl/jbang/core/commands/tui/KafkaTab.java | 237 +++++++++++++++++++++
.../dsl/jbang/core/commands/tui/StatusParser.java | 39 ++++
.../dsl/jbang/core/commands/tui/TabRegistry.java | 5 +-
.../dsl/jbang/core/commands/tui/TuiIcons.java | 1 +
.../jbang/core/commands/tui/TabRegistryTest.java | 8 +-
7 files changed, 318 insertions(+), 5 deletions(-)
diff --git
a/dsl/camel-jbang/camel-jbang-plugin-tui/src/main/java/org/apache/camel/dsl/jbang/core/commands/tui/IntegrationInfo.java
b/dsl/camel-jbang/camel-jbang-plugin-tui/src/main/java/org/apache/camel/dsl/jbang/core/commands/tui/IntegrationInfo.java
index b0201105a1c7..6598567a4198 100644
---
a/dsl/camel-jbang/camel-jbang-plugin-tui/src/main/java/org/apache/camel/dsl/jbang/core/commands/tui/IntegrationInfo.java
+++
b/dsl/camel-jbang/camel-jbang-plugin-tui/src/main/java/org/apache/camel/dsl/jbang/core/commands/tui/IntegrationInfo.java
@@ -83,6 +83,7 @@ class IntegrationInfo {
final List<HealthCheckInfo> healthChecks = new ArrayList<>();
final List<EndpointInfo> endpoints = new ArrayList<>();
final List<CircuitBreakerInfo> circuitBreakers = new ArrayList<>();
+ final List<KafkaConsumerInfo> kafkaConsumers = new ArrayList<>();
int errorCount;
volatile List<ErrorInfo> errors = new ArrayList<>();
volatile List<ActivityEntry> activity = List.of();
diff --git
a/dsl/camel-jbang/camel-jbang-plugin-tui/src/main/java/org/apache/camel/dsl/jbang/core/commands/tui/KafkaConsumerInfo.java
b/dsl/camel-jbang/camel-jbang-plugin-tui/src/main/java/org/apache/camel/dsl/jbang/core/commands/tui/KafkaConsumerInfo.java
new file mode 100644
index 000000000000..ade2907e5a86
--- /dev/null
+++
b/dsl/camel-jbang/camel-jbang-plugin-tui/src/main/java/org/apache/camel/dsl/jbang/core/commands/tui/KafkaConsumerInfo.java
@@ -0,0 +1,32 @@
+/*
+ * 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.dsl.jbang.core.commands.tui;
+
+class KafkaConsumerInfo {
+ String routeId;
+ String uri;
+ String threadId;
+ String state;
+ String lastError;
+ String groupId;
+ String groupInstanceId;
+ String memberId;
+ int generationId;
+ String lastTopic;
+ int lastPartition;
+ long lastOffset;
+}
diff --git
a/dsl/camel-jbang/camel-jbang-plugin-tui/src/main/java/org/apache/camel/dsl/jbang/core/commands/tui/KafkaTab.java
b/dsl/camel-jbang/camel-jbang-plugin-tui/src/main/java/org/apache/camel/dsl/jbang/core/commands/tui/KafkaTab.java
new file mode 100644
index 000000000000..945ab7f1b289
--- /dev/null
+++
b/dsl/camel-jbang/camel-jbang-plugin-tui/src/main/java/org/apache/camel/dsl/jbang/core/commands/tui/KafkaTab.java
@@ -0,0 +1,237 @@
+/*
+ * 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.dsl.jbang.core.commands.tui;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Locale;
+
+import dev.tamboui.layout.Constraint;
+import dev.tamboui.layout.Rect;
+import dev.tamboui.style.Style;
+import dev.tamboui.terminal.Frame;
+import dev.tamboui.text.Span;
+import dev.tamboui.widgets.block.Block;
+import dev.tamboui.widgets.block.BorderType;
+import dev.tamboui.widgets.block.Borders;
+import dev.tamboui.widgets.table.Cell;
+import dev.tamboui.widgets.table.Row;
+import dev.tamboui.widgets.table.Table;
+import org.apache.camel.util.json.JsonArray;
+import org.apache.camel.util.json.JsonObject;
+
+import static org.apache.camel.dsl.jbang.core.commands.tui.TuiHelper.*;
+
+class KafkaTab extends AbstractTableTab {
+
+ KafkaTab(MonitorContext ctx) {
+ super(ctx, "route", "state", "group", "topic");
+ }
+
+ @Override
+ protected int getRowCount() {
+ IntegrationInfo info = ctx.findSelectedIntegration();
+ return info != null ? info.kafkaConsumers.size() : 0;
+ }
+
+ @Override
+ public void onTabSelected() {
+ IntegrationInfo info = ctx.findSelectedIntegration();
+ if (info != null && !info.kafkaConsumers.isEmpty() &&
tableState.selected() == null) {
+ tableState.select(0);
+ }
+ }
+
+ @Override
+ protected void renderContent(Frame frame, Rect area, IntegrationInfo info)
{
+ List<KafkaConsumerInfo> sorted = new ArrayList<>(info.kafkaConsumers);
+ sorted.sort(this::sortKafka);
+
+ List<Row> rows = new ArrayList<>();
+ for (KafkaConsumerInfo ki : sorted) {
+ String state = ki.state != null ? ki.state : "";
+ Style stateStyle = switch (state.toLowerCase(Locale.ROOT)) {
+ case "running" -> Theme.success();
+ case "paused", "resume_requested" -> Theme.warning();
+ default -> state.isEmpty() ? Style.EMPTY : Theme.error();
+ };
+ String displayState = capitalize(state);
+ String topic = ki.lastTopic != null ? ki.lastTopic : "";
+ String partition = ki.lastTopic != null ?
String.valueOf(ki.lastPartition) : "";
+ String offset = ki.lastTopic != null ?
String.valueOf(ki.lastOffset) : "";
+ String error = ki.lastError != null ? ki.lastError : "";
+ String uri = ki.uri != null ? ki.uri : "";
+
+ rows.add(Row.from(
+ Cell.from(Span.styled(ki.routeId != null ? ki.routeId :
"", Style.EMPTY.fg(Theme.accent()))),
+ Cell.from(Span.styled(displayState, stateStyle)),
+ Cell.from(ki.groupId != null ? ki.groupId : ""),
+ Cell.from(topic),
+ rightCell(partition, 9),
+ rightCell(offset, 10),
+ Cell.from(Span.styled(error, error.isEmpty() ? Style.EMPTY
: Theme.error())),
+ Cell.from(Span.styled(uri, Style.EMPTY.dim()))));
+ }
+
+ if (rows.isEmpty()) {
+ rows.add(emptyRow("No Kafka consumers", 8));
+ }
+
+ Table table = Table.builder()
+ .rows(rows)
+ .header(Row.from(
+ Cell.from(Span.styled(sortLabel("ROUTE", "route"),
sortStyle("route"))),
+ Cell.from(Span.styled(sortLabel("STATUS", "state"),
sortStyle("state"))),
+ Cell.from(Span.styled(sortLabel("GROUP-ID", "group"),
sortStyle("group"))),
+ Cell.from(Span.styled(sortLabel("TOPIC", "topic"),
sortStyle("topic"))),
+ rightCell("PARTITION", 9, Style.EMPTY.bold()),
+ rightCell("OFFSET", 10, Style.EMPTY.bold()),
+ Cell.from(Span.styled("ERROR", Style.EMPTY.bold())),
+ Cell.from(Span.styled("ENDPOINT",
Style.EMPTY.bold()))))
+ .widths(
+ Constraint.length(20),
+ Constraint.length(12),
+ Constraint.length(20),
+ Constraint.length(20),
+ Constraint.length(9),
+ Constraint.length(10),
+ Constraint.length(30),
+ Constraint.fill())
+ .highlightStyle(Theme.selectionBg())
+ .highlightSpacing(Table.HighlightSpacing.ALWAYS)
+
.block(Block.builder().borderType(BorderType.ROUNDED).borders(Borders.ALL)
+ .title(" Kafka ").build())
+ .build();
+
+ lastTableArea = area;
+ frame.renderStatefulWidget(table, area, tableState);
+ renderScrollbar(frame, sorted.size());
+ }
+
+ private int sortKafka(KafkaConsumerInfo a, KafkaConsumerInfo b) {
+ int result = switch (sort) {
+ case "state" -> compareStr(a.state, b.state);
+ case "group" -> compareStr(a.groupId, b.groupId);
+ case "topic" -> compareStr(a.lastTopic, b.lastTopic);
+ default -> compareStr(a.routeId, b.routeId);
+ };
+ return sortReversed ? -result : result;
+ }
+
+ private static String capitalize(String s) {
+ if (s == null || s.isEmpty()) {
+ return "";
+ }
+ String lower = s.toLowerCase(Locale.ROOT);
+ return Character.toUpperCase(lower.charAt(0)) + lower.substring(1);
+ }
+
+ @Override
+ public SelectionContext getSelectionContext() {
+ IntegrationInfo info = ctx.findSelectedIntegration();
+ if (info == null || info.kafkaConsumers.isEmpty()) {
+ return null;
+ }
+ List<KafkaConsumerInfo> sorted = new ArrayList<>(info.kafkaConsumers);
+ sorted.sort(this::sortKafka);
+ List<String> items = sorted.stream()
+ .map(ki -> (ki.routeId != null ? ki.routeId : "") + " " +
(ki.groupId != null ? ki.groupId : ""))
+ .toList();
+ Integer sel = tableState.selected();
+ return new SelectionContext("table", items, sel != null ? sel : -1,
items.size(), "Kafka");
+ }
+
+ @Override
+ public String description() {
+ return "Apache Kafka consumer status (group, topic, partition,
offset)";
+ }
+
+ @Override
+ public String getHelpText() {
+ return """
+ # Kafka
+
+ Displays live status for all Kafka consumers in the selected
integration.
+ Data comes from the Kafka dev console built into
`camel-kafka`, which
+ tracks each consumer worker thread's group membership, current
position,
+ and health state.
+
+ ## Table Columns
+
+ - **ROUTE** — Route ID that owns this Kafka consumer
+ - **STATUS** — Worker state: `Running` (green, actively
polling),
+ `Paused` (yellow, consumer paused), or error states (red)
+ - **GROUP-ID** — Kafka consumer group ID this consumer belongs
to
+ - **TOPIC** — Topic of the last consumed record
+ - **PARTITION** — Partition number of the last consumed record
+ - **OFFSET** — Offset of the last consumed record
+ - **ERROR** — Last error message if the consumer is unhealthy
+ - **ENDPOINT** — Full Camel endpoint URI for this consumer
+
+ ## What to Look For
+
+ - **All consumers Running**: Normal operation. Offsets should
be
+ advancing steadily if messages are flowing.
+ - **Consumer Paused**: The consumer has been programmatically
paused
+ (e.g., by a route policy or manual suspension).
+ - **Error column populated**: The consumer worker is unhealthy.
+ Check the error message for connection issues, authentication
+ failures, or deserialization errors.
+ - **Offset not advancing**: If messages are being produced but
the
+ offset stays the same, the consumer may be stuck or the topic
+ may have no new messages on that partition.
+
+ ## Keys
+
+ - `Up/Down` — select consumer
+ - `s` — cycle sort column
+ - `S` — reverse sort order
+ """;
+ }
+
+ @Override
+ public JsonObject getTableDataAsJson() {
+ IntegrationInfo info = ctx.findSelectedIntegration();
+ if (info == null) {
+ return null;
+ }
+ JsonObject result = new JsonObject();
+ result.put("tab", "Kafka");
+ JsonArray rows = new JsonArray();
+ List<KafkaConsumerInfo> sorted = new ArrayList<>(info.kafkaConsumers);
+ sorted.sort(this::sortKafka);
+ for (KafkaConsumerInfo ki : sorted) {
+ JsonObject row = new JsonObject();
+ row.put("routeId", ki.routeId);
+ row.put("state", ki.state);
+ row.put("groupId", ki.groupId);
+ row.put("lastTopic", ki.lastTopic);
+ row.put("lastPartition", ki.lastPartition);
+ row.put("lastOffset", ki.lastOffset);
+ if (ki.lastError != null) {
+ row.put("lastError", ki.lastError);
+ }
+ row.put("uri", ki.uri);
+ rows.add(row);
+ }
+ result.put("rows", rows);
+ result.put("totalRows", info.kafkaConsumers.size());
+ Integer sel = tableState.selected();
+ result.put("selectedIndex", sel != null ? sel : -1);
+ return result;
+ }
+}
diff --git
a/dsl/camel-jbang/camel-jbang-plugin-tui/src/main/java/org/apache/camel/dsl/jbang/core/commands/tui/StatusParser.java
b/dsl/camel-jbang/camel-jbang-plugin-tui/src/main/java/org/apache/camel/dsl/jbang/core/commands/tui/StatusParser.java
index 85482474c610..f0a3be50cd7a 100644
---
a/dsl/camel-jbang/camel-jbang-plugin-tui/src/main/java/org/apache/camel/dsl/jbang/core/commands/tui/StatusParser.java
+++
b/dsl/camel-jbang/camel-jbang-plugin-tui/src/main/java/org/apache/camel/dsl/jbang/core/commands/tui/StatusParser.java
@@ -382,6 +382,9 @@ final class StatusParser {
}
}
+ // Parse Kafka consumers
+ parseKafkaSection(root, info);
+
// Parse error count from error registry
JsonObject errorsObj = (JsonObject) root.get("errors");
if (errorsObj != null) {
@@ -997,6 +1000,42 @@ final class StatusParser {
}
}
+ private static void parseKafkaSection(JsonObject root, IntegrationInfo
info) {
+ JsonObject kafkaObj = (JsonObject) root.get("kafka");
+ if (kafkaObj == null) {
+ return;
+ }
+ JsonArray consumers = (JsonArray) kafkaObj.get("kafkaConsumers");
+ if (consumers == null) {
+ return;
+ }
+ for (Object c : consumers) {
+ JsonObject cj = (JsonObject) c;
+ String routeId = cj.getString("routeId");
+ String uri = cj.getString("uri");
+ JsonArray workers = (JsonArray) cj.get("workers");
+ if (workers != null) {
+ for (Object w : workers) {
+ JsonObject wo = (JsonObject) w;
+ KafkaConsumerInfo ki = new KafkaConsumerInfo();
+ ki.routeId = routeId;
+ ki.uri = uri;
+ ki.threadId = wo.getString("threadId");
+ ki.state = wo.getString("state");
+ ki.lastError = wo.getString("lastError");
+ ki.groupId = wo.getString("groupId");
+ ki.groupInstanceId = wo.getString("groupInstanceId");
+ ki.memberId = wo.getString("memberId");
+ ki.generationId = wo.getIntegerOrDefault("generationId",
0);
+ ki.lastTopic = wo.getString("lastTopic");
+ ki.lastPartition = wo.getIntegerOrDefault("lastPartition",
0);
+ ki.lastOffset = wo.getLongOrDefault("lastOffset", 0);
+ info.kafkaConsumers.add(ki);
+ }
+ }
+ }
+ }
+
@SuppressWarnings("unchecked")
private static void parseMicrometerMeters(
JsonObject micrometerObj, String section, String type,
IntegrationInfo info) {
diff --git
a/dsl/camel-jbang/camel-jbang-plugin-tui/src/main/java/org/apache/camel/dsl/jbang/core/commands/tui/TabRegistry.java
b/dsl/camel-jbang/camel-jbang-plugin-tui/src/main/java/org/apache/camel/dsl/jbang/core/commands/tui/TabRegistry.java
index ff8ba0099fdf..06c9d3ffdeff 100644
---
a/dsl/camel-jbang/camel-jbang-plugin-tui/src/main/java/org/apache/camel/dsl/jbang/core/commands/tui/TabRegistry.java
+++
b/dsl/camel-jbang/camel-jbang-plugin-tui/src/main/java/org/apache/camel/dsl/jbang/core/commands/tui/TabRegistry.java
@@ -93,6 +93,7 @@ class TabRegistry {
private SpansTab spansTab;
private ProcessTab processTab;
private OverviewTab overviewTab;
+ private KafkaTab kafkaTab;
private DataSourceTab dataSourceTab;
private SqlQueryTab sqlQueryTab;
private SqlTraceTab sqlTraceTab;
@@ -113,6 +114,7 @@ class TabRegistry {
diagramTab = new DiagramTab(ctx);
routesTab = new RoutesTab(ctx);
consumersTab = new ConsumersTab(ctx);
+ kafkaTab = new KafkaTab(ctx);
dataSourceTab = new DataSourceTab(ctx);
heapHistogramTab = new HeapHistogramTab(ctx);
memoryLeakTab = new MemoryLeakTab(ctx);
@@ -161,9 +163,10 @@ class TabRegistry {
new MoreTab(TuiIcons.TAB_HEAP, "Heap Histogram", "&Heap
Histogram", heapHistogramTab),
new MoreTab(TuiIcons.TAB_INFLIGHT, "Inflight", "In&flight",
inflightTab),
new MoreTab(TuiIcons.TAB_DATASOURCE, "JDBC DataSource", "&JDBC
DataSource", dataSourceTab),
+ new MoreTab(TuiIcons.TAB_KAFKA, "Kafka", "&Kafka", kafkaTab),
new MoreTab(TuiIcons.TAB_MAVEN_DEPENDENCIES, "Maven
Dependencies", "Maven &Dependencies", mavenDependenciesTab),
new MoreTab(TuiIcons.TAB_MEMORY, "Memory", "&Memory",
memoryTab),
- new MoreTab(TuiIcons.TAB_MEMORY_LEAK, "Memory Leak", "Memory
Lea&k", memoryLeakTab),
+ new MoreTab(TuiIcons.TAB_MEMORY_LEAK, "Memory Leak", "Memor&y
Leak", memoryLeakTab),
new MoreTab(TuiIcons.TAB_METRICS, "Metrics", "Metr&ics",
metricsTab),
new MoreTab(TuiIcons.TAB_SQL_QUERY, "SQL Query", "S&QL Query",
sqlQueryTab),
new MoreTab(TuiIcons.TAB_SQL_TRACE, "SQL Trace", "SQL T&race",
sqlTraceTab),
diff --git
a/dsl/camel-jbang/camel-jbang-plugin-tui/src/main/java/org/apache/camel/dsl/jbang/core/commands/tui/TuiIcons.java
b/dsl/camel-jbang/camel-jbang-plugin-tui/src/main/java/org/apache/camel/dsl/jbang/core/commands/tui/TuiIcons.java
index c80d022373ca..95237e6e990e 100644
---
a/dsl/camel-jbang/camel-jbang-plugin-tui/src/main/java/org/apache/camel/dsl/jbang/core/commands/tui/TuiIcons.java
+++
b/dsl/camel-jbang/camel-jbang-plugin-tui/src/main/java/org/apache/camel/dsl/jbang/core/commands/tui/TuiIcons.java
@@ -134,6 +134,7 @@ final class TuiIcons {
static final String TAB_DATASOURCE = "💿";
static final String TAB_HEAP = MEMORY;
static final String TAB_INFLIGHT = RESET;
+ static final String TAB_KAFKA = "📨";
static final String TAB_MAVEN_DEPENDENCIES = "📜";
static final String TAB_MEMORY = MEMORY;
static final String TAB_MEMORY_LEAK = "💧";
diff --git
a/dsl/camel-jbang/camel-jbang-plugin-tui/src/test/java/org/apache/camel/dsl/jbang/core/commands/tui/TabRegistryTest.java
b/dsl/camel-jbang/camel-jbang-plugin-tui/src/test/java/org/apache/camel/dsl/jbang/core/commands/tui/TabRegistryTest.java
index 2560954442bc..db15e29f4996 100644
---
a/dsl/camel-jbang/camel-jbang-plugin-tui/src/test/java/org/apache/camel/dsl/jbang/core/commands/tui/TabRegistryTest.java
+++
b/dsl/camel-jbang/camel-jbang-plugin-tui/src/test/java/org/apache/camel/dsl/jbang/core/commands/tui/TabRegistryTest.java
@@ -82,8 +82,8 @@ class TabRegistryTest {
}
@Test
- void moreTabsHasTwentyOneEntries() {
- assertEquals(21, registry.moreTabs().size());
+ void moreTabsHasTwentyTwoEntries() {
+ assertEquals(22, registry.moreTabs().size());
}
@Test
@@ -109,8 +109,8 @@ class TabRegistryTest {
// MORE_SHORTCUTS array carried before the MoreTab refactor. A label
edit that repoints a key must fail here.
List<Character> shortcuts =
registry.moreTabs().stream().map(TabRegistry.MoreTab::shortcut).toList();
assertEquals(
- List.of('B', 'W', 'C', 'A', 'G', 'N', 'V', 'E', 'H', 'F', 'J',
'D', 'M', 'K', 'I', 'Q', 'R', 'O', 'P', 'S',
- 'T'),
+ List.of('B', 'W', 'C', 'A', 'G', 'N', 'V', 'E', 'H', 'F', 'J',
'K', 'D', 'M', 'Y', 'I', 'Q', 'R', 'O', 'P',
+ 'S', 'T'),
shortcuts, "More tab shortcut letters must match the
historical sequence");
}
