This is an automated email from the ASF dual-hosted git repository.
lizhimins pushed a commit to branch rocketmq-studio
in repository https://gitbox.apache.org/repos/asf/rocketmq-dashboard.git
The following commit(s) were added to refs/heads/rocketmq-studio by this push:
new 45535312 feat: add producer connection health summary (#1270)
45535312 is described below
commit 455353128ea055c41bad95f846c2c013ee8c1b70
Author: yx9o <[email protected]>
AuthorDate: Mon Aug 10 20:20:08 2026 +0800
feat: add producer connection health summary (#1270)
---
.../cluster/client/ProducerConnectionResultVO.java | 8 +-
...O.java => ProducerConnectionSummaryItemVO.java} | 7 +-
.../client/ProducerConnectionSummaryVO.java | 144 +++++++++++++++++++++
.../client/ProducerConnectionSummaryVOTest.java | 80 ++++++++++++
.../cluster/client/ProducerControllerTest.java | 6 +-
web/src/api/producer.test.ts | 70 +++++++++-
web/src/api/producer.ts | 115 +++++++++++++++-
web/src/i18n/translations.ts | 27 ++++
web/src/pages/studio/Producer.tsx | 111 +++++++++++++++-
web/src/pages/studio/__tests__/Producer.test.tsx | 107 ++++++++++++---
10 files changed, 641 insertions(+), 34 deletions(-)
diff --git
a/server/src/main/java/org/apache/rocketmq/studio/cluster/client/ProducerConnectionResultVO.java
b/server/src/main/java/org/apache/rocketmq/studio/cluster/client/ProducerConnectionResultVO.java
index d99ad97b..5a77d3fc 100644
---
a/server/src/main/java/org/apache/rocketmq/studio/cluster/client/ProducerConnectionResultVO.java
+++
b/server/src/main/java/org/apache/rocketmq/studio/cluster/client/ProducerConnectionResultVO.java
@@ -16,7 +16,6 @@
*/
package org.apache.rocketmq.studio.cluster.client;
-import lombok.AllArgsConstructor;
import lombok.Data;
import lombok.NoArgsConstructor;
@@ -24,7 +23,12 @@ import java.util.List;
@Data
@NoArgsConstructor
-@AllArgsConstructor
public class ProducerConnectionResultVO {
private List<ProducerConnectionVO> connectionSet;
+ private ProducerConnectionSummaryVO summary;
+
+ public ProducerConnectionResultVO(List<ProducerConnectionVO>
connectionSet) {
+ this.connectionSet = connectionSet == null ? List.of() : connectionSet;
+ this.summary = ProducerConnectionSummaryVO.from(this.connectionSet);
+ }
}
diff --git
a/server/src/main/java/org/apache/rocketmq/studio/cluster/client/ProducerConnectionResultVO.java
b/server/src/main/java/org/apache/rocketmq/studio/cluster/client/ProducerConnectionSummaryItemVO.java
similarity index 89%
copy from
server/src/main/java/org/apache/rocketmq/studio/cluster/client/ProducerConnectionResultVO.java
copy to
server/src/main/java/org/apache/rocketmq/studio/cluster/client/ProducerConnectionSummaryItemVO.java
index d99ad97b..d00a3ed3 100644
---
a/server/src/main/java/org/apache/rocketmq/studio/cluster/client/ProducerConnectionResultVO.java
+++
b/server/src/main/java/org/apache/rocketmq/studio/cluster/client/ProducerConnectionSummaryItemVO.java
@@ -20,11 +20,10 @@ import lombok.AllArgsConstructor;
import lombok.Data;
import lombok.NoArgsConstructor;
-import java.util.List;
-
@Data
@NoArgsConstructor
@AllArgsConstructor
-public class ProducerConnectionResultVO {
- private List<ProducerConnectionVO> connectionSet;
+public class ProducerConnectionSummaryItemVO {
+ private String value;
+ private long count;
}
diff --git
a/server/src/main/java/org/apache/rocketmq/studio/cluster/client/ProducerConnectionSummaryVO.java
b/server/src/main/java/org/apache/rocketmq/studio/cluster/client/ProducerConnectionSummaryVO.java
new file mode 100644
index 00000000..7294a2e1
--- /dev/null
+++
b/server/src/main/java/org/apache/rocketmq/studio/cluster/client/ProducerConnectionSummaryVO.java
@@ -0,0 +1,144 @@
+/*
+ * 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.rocketmq.studio.cluster.client;
+
+import lombok.Data;
+import lombok.NoArgsConstructor;
+
+import java.util.Comparator;
+import java.util.List;
+import java.util.Map;
+import java.util.Objects;
+import java.util.function.Function;
+import java.util.stream.Collectors;
+
+@Data
+@NoArgsConstructor
+public class ProducerConnectionSummaryVO {
+ public static final String READY = "READY";
+ public static final String WARNING = "WARNING";
+ public static final String UNAVAILABLE = "UNAVAILABLE";
+ public static final String NO_CONNECTIONS = "NO_CONNECTIONS";
+ public static final String DUPLICATE_CLIENT_ID = "DUPLICATE_CLIENT_ID";
+ public static final String MIXED_CLIENT_VERSION = "MIXED_CLIENT_VERSION";
+ public static final String INCOMPLETE_CLIENT_METADATA =
"INCOMPLETE_CLIENT_METADATA";
+
+ private int totalConnections;
+ private int uniqueClientCount;
+ private int uniqueAddressCount;
+ private int uniqueLanguageCount;
+ private int uniqueVersionCount;
+ private List<ProducerConnectionSummaryItemVO> languages = List.of();
+ private List<ProducerConnectionSummaryItemVO> versions = List.of();
+ private List<String> duplicateClientIds = List.of();
+ private List<String> warnings = List.of();
+ private String readiness = READY;
+
+ public static ProducerConnectionSummaryVO from(List<ProducerConnectionVO>
connections) {
+ List<ProducerConnectionVO> safeConnections = connections == null ?
List.of() : connections;
+ ProducerConnectionSummaryVO summary = new
ProducerConnectionSummaryVO();
+ summary.totalConnections = safeConnections.size();
+ summary.uniqueClientCount = countDistinct(safeConnections,
ProducerConnectionVO::getClientId);
+ summary.uniqueAddressCount = countDistinct(safeConnections,
ProducerConnectionVO::getClientAddr);
+ summary.languages = distribution(safeConnections,
ProducerConnectionVO::getLanguage);
+ summary.versions = distribution(safeConnections,
ProducerConnectionVO::getVersionDesc);
+ summary.uniqueLanguageCount = summary.languages.size();
+ summary.uniqueVersionCount = summary.versions.size();
+ summary.duplicateClientIds = duplicateClientIds(safeConnections);
+ summary.warnings = warnings(summary, safeConnections);
+ summary.readiness = readiness(summary);
+ return summary;
+ }
+
+ private static int countDistinct(
+ List<ProducerConnectionVO> connections,
Function<ProducerConnectionVO, String> extractor) {
+ return (int) connections.stream()
+ .map(extractor)
+ .filter(ProducerConnectionSummaryVO::hasText)
+ .distinct()
+ .count();
+ }
+
+ private static List<ProducerConnectionSummaryItemVO> distribution(
+ List<ProducerConnectionVO> connections,
Function<ProducerConnectionVO, String> extractor) {
+ return connections.stream()
+ .map(extractor)
+ .map(ProducerConnectionSummaryVO::normalizeDimension)
+ .collect(Collectors.groupingBy(Function.identity(),
Collectors.counting()))
+ .entrySet()
+ .stream()
+ .sorted(Comparator
+ .<Map.Entry<String,
Long>>comparingLong(Map.Entry::getValue)
+ .reversed()
+ .thenComparing(Map.Entry::getKey))
+ .map(entry -> new
ProducerConnectionSummaryItemVO(entry.getKey(), entry.getValue()))
+ .toList();
+ }
+
+ private static List<String> duplicateClientIds(List<ProducerConnectionVO>
connections) {
+ return connections.stream()
+ .map(ProducerConnectionVO::getClientId)
+ .filter(ProducerConnectionSummaryVO::hasText)
+ .collect(Collectors.groupingBy(Function.identity(),
Collectors.counting()))
+ .entrySet()
+ .stream()
+ .filter(entry -> entry.getValue() > 1)
+ .map(Map.Entry::getKey)
+ .sorted()
+ .toList();
+ }
+
+ private static List<String> warnings(
+ ProducerConnectionSummaryVO summary, List<ProducerConnectionVO>
connections) {
+ if (summary.totalConnections == 0) {
+ return List.of(NO_CONNECTIONS);
+ }
+ List<String> warnings = new java.util.ArrayList<>();
+ if (!summary.duplicateClientIds.isEmpty()) {
+ warnings.add(DUPLICATE_CLIENT_ID);
+ }
+ if (summary.uniqueVersionCount > 1) {
+ warnings.add(MIXED_CLIENT_VERSION);
+ }
+ if
(connections.stream().anyMatch(ProducerConnectionSummaryVO::hasIncompleteMetadata))
{
+ warnings.add(INCOMPLETE_CLIENT_METADATA);
+ }
+ return List.copyOf(warnings);
+ }
+
+ private static String readiness(ProducerConnectionSummaryVO summary) {
+ if (summary.totalConnections == 0) {
+ return UNAVAILABLE;
+ }
+ return summary.warnings.isEmpty() ? READY : WARNING;
+ }
+
+ private static boolean hasIncompleteMetadata(ProducerConnectionVO
connection) {
+ return !hasText(connection.getClientId())
+ || !hasText(connection.getClientAddr())
+ || !hasText(connection.getLanguage())
+ || !hasText(connection.getVersionDesc());
+ }
+
+ private static String normalizeDimension(String value) {
+ return hasText(value) ? value.trim() : "UNKNOWN";
+ }
+
+ private static boolean hasText(String value) {
+ return value != null && !value.trim().isEmpty() &&
!Objects.equals(value.trim(), "null");
+ }
+}
diff --git
a/server/src/test/java/org/apache/rocketmq/studio/cluster/client/ProducerConnectionSummaryVOTest.java
b/server/src/test/java/org/apache/rocketmq/studio/cluster/client/ProducerConnectionSummaryVOTest.java
new file mode 100644
index 00000000..24c5eb68
--- /dev/null
+++
b/server/src/test/java/org/apache/rocketmq/studio/cluster/client/ProducerConnectionSummaryVOTest.java
@@ -0,0 +1,80 @@
+/*
+ * 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.rocketmq.studio.cluster.client;
+
+import org.junit.jupiter.api.Test;
+
+import java.util.List;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+class ProducerConnectionSummaryVOTest {
+
+ @Test
+ void fromShouldMarkEmptyConnectionsUnavailable() {
+ ProducerConnectionSummaryVO summary =
ProducerConnectionSummaryVO.from(List.of());
+
+ assertThat(summary.getTotalConnections()).isZero();
+
assertThat(summary.getReadiness()).isEqualTo(ProducerConnectionSummaryVO.UNAVAILABLE);
+
assertThat(summary.getWarnings()).containsExactly(ProducerConnectionSummaryVO.NO_CONNECTIONS);
+ }
+
+ @Test
+ void fromShouldReportMixedVersionsAndDuplicateClients() {
+ ProducerConnectionSummaryVO summary =
ProducerConnectionSummaryVO.from(List.of(
+ connection("producer-a", "10.0.0.1:38888", "Java", "5.1.0"),
+ connection("producer-a", "10.0.0.2:38888", "Java", "5.1.0"),
+ connection("producer-b", "10.0.0.3:38888", "Go", "5.2.0")));
+
+ assertThat(summary.getTotalConnections()).isEqualTo(3);
+ assertThat(summary.getUniqueClientCount()).isEqualTo(2);
+ assertThat(summary.getUniqueAddressCount()).isEqualTo(3);
+ assertThat(summary.getLanguages())
+ .extracting(ProducerConnectionSummaryItemVO::getValue)
+ .containsExactly("Java", "Go");
+ assertThat(summary.getVersions())
+ .extracting(ProducerConnectionSummaryItemVO::getValue)
+ .containsExactly("5.1.0", "5.2.0");
+
assertThat(summary.getDuplicateClientIds()).containsExactly("producer-a");
+ assertThat(summary.getWarnings()).containsExactly(
+ ProducerConnectionSummaryVO.DUPLICATE_CLIENT_ID,
+ ProducerConnectionSummaryVO.MIXED_CLIENT_VERSION);
+
assertThat(summary.getReadiness()).isEqualTo(ProducerConnectionSummaryVO.WARNING);
+ }
+
+ @Test
+ void fromShouldWarnWhenConnectionMetadataIsIncomplete() {
+ ProducerConnectionSummaryVO summary =
ProducerConnectionSummaryVO.from(List.of(
+ connection("producer-a", "", null, "5.1.0")));
+
+
assertThat(summary.getReadiness()).isEqualTo(ProducerConnectionSummaryVO.WARNING);
+ assertThat(summary.getWarnings()).containsExactly(
+ ProducerConnectionSummaryVO.INCOMPLETE_CLIENT_METADATA);
+ assertThat(summary.getLanguages())
+ .extracting(ProducerConnectionSummaryItemVO::getValue)
+ .containsExactly("UNKNOWN");
+ }
+
+ private ProducerConnectionVO connection(String clientId, String address,
String language, String version) {
+ return ProducerConnectionVO.builder()
+ .clientId(clientId)
+ .clientAddr(address)
+ .language(language)
+ .versionDesc(version)
+ .build();
+ }
+}
diff --git
a/server/src/test/java/org/apache/rocketmq/studio/cluster/client/ProducerControllerTest.java
b/server/src/test/java/org/apache/rocketmq/studio/cluster/client/ProducerControllerTest.java
index c59b4134..5e824875 100644
---
a/server/src/test/java/org/apache/rocketmq/studio/cluster/client/ProducerControllerTest.java
+++
b/server/src/test/java/org/apache/rocketmq/studio/cluster/client/ProducerControllerTest.java
@@ -75,7 +75,11 @@ class ProducerControllerTest {
.andExpect(jsonPath("$.connectionSet[0].clientId").value("producer-1"))
.andExpect(jsonPath("$.connectionSet[0].clientAddr").value("10.0.0.1:38888"))
.andExpect(jsonPath("$.connectionSet[0].language").value("Java"))
-
.andExpect(jsonPath("$.connectionSet[0].versionDesc").value("5.1.0"));
+
.andExpect(jsonPath("$.connectionSet[0].versionDesc").value("5.1.0"))
+ .andExpect(jsonPath("$.summary.totalConnections").value(1))
+ .andExpect(jsonPath("$.summary.uniqueClientCount").value(1))
+ .andExpect(jsonPath("$.summary.uniqueAddressCount").value(1))
+ .andExpect(jsonPath("$.summary.readiness").value("READY"));
verify(producerConnectionService).listConnections("instance-1",
"order-topic", "pg-order");
}
diff --git a/web/src/api/producer.test.ts b/web/src/api/producer.test.ts
index 6303b075..8245cfd1 100644
--- a/web/src/api/producer.test.ts
+++ b/web/src/api/producer.test.ts
@@ -18,7 +18,12 @@
import MockAdapter from 'axios-mock-adapter';
import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest';
import client from './client';
-import { fetchProducerGroups, fetchTopicList, queryProducerConnection } from
'./producer';
+import {
+ buildProducerConnectionSummary,
+ fetchProducerGroups,
+ fetchTopicList,
+ queryProducerConnection,
+} from './producer';
const mock = new MockAdapter(client);
@@ -94,14 +99,71 @@ describe('Producer API', () => {
});
const result = await queryProducerConnection('instance-1', 'order-events',
'order-producer');
- expect(result).toHaveLength(2);
- expect(result[0].clientId).toBe('producer-1');
+ expect(result.connectionSet).toHaveLength(2);
+ expect(result.connectionSet[0].clientId).toBe('producer-1');
+ expect(result.summary.totalConnections).toBe(2);
+ expect(result.summary.readiness).toBe('READY');
});
it('handles empty producer connections', async () => {
mock.onGet('/producer/connection').reply(200, { connectionSet: [] });
const result = await queryProducerConnection('instance-1', 'topic',
'group');
- expect(result).toEqual([]);
+ expect(result.connectionSet).toEqual([]);
+ expect(result.summary.readiness).toBe('UNAVAILABLE');
+ expect(result.summary.warnings).toEqual(['NO_CONNECTIONS']);
+ });
+
+ it('uses backend producer connection summaries when provided', async () => {
+ mock.onGet('/producer/connection').reply(200, {
+ connectionSet: [],
+ summary: {
+ totalConnections: 0,
+ uniqueClientCount: 0,
+ uniqueAddressCount: 0,
+ uniqueLanguageCount: 0,
+ uniqueVersionCount: 0,
+ languages: [],
+ versions: [],
+ duplicateClientIds: [],
+ warnings: ['NO_CONNECTIONS'],
+ readiness: 'UNAVAILABLE',
+ },
+ });
+
+ const result = await queryProducerConnection('instance-1', 'topic',
'group');
+ expect(result.summary).toEqual({
+ totalConnections: 0,
+ uniqueClientCount: 0,
+ uniqueAddressCount: 0,
+ uniqueLanguageCount: 0,
+ uniqueVersionCount: 0,
+ languages: [],
+ versions: [],
+ duplicateClientIds: [],
+ warnings: ['NO_CONNECTIONS'],
+ readiness: 'UNAVAILABLE',
+ });
+ });
+
+ it('builds producer connection warning summaries for legacy responses', ()
=> {
+ const result = buildProducerConnectionSummary([
+ {
+ clientId: 'producer-a',
+ clientAddr: '10.0.0.1',
+ language: 'Java',
+ versionDesc: '5.1.0',
+ },
+ {
+ clientId: 'producer-a',
+ clientAddr: '10.0.0.2',
+ language: 'Go',
+ versionDesc: '5.2.0',
+ },
+ ]);
+
+ expect(result.readiness).toBe('WARNING');
+ expect(result.duplicateClientIds).toEqual(['producer-a']);
+ expect(result.warnings).toEqual(['DUPLICATE_CLIENT_ID',
'MIXED_CLIENT_VERSION']);
});
});
diff --git a/web/src/api/producer.ts b/web/src/api/producer.ts
index 61138103..5249a870 100644
--- a/web/src/api/producer.ts
+++ b/web/src/api/producer.ts
@@ -25,6 +25,34 @@ export interface ProducerConnection {
versionDesc: string;
}
+export type ProducerReadiness = 'READY' | 'WARNING' | 'UNAVAILABLE';
+
+export type ProducerConnectionWarning =
+ 'NO_CONNECTIONS' | 'DUPLICATE_CLIENT_ID' | 'MIXED_CLIENT_VERSION' |
'INCOMPLETE_CLIENT_METADATA';
+
+export interface ProducerConnectionSummaryItem {
+ value: string;
+ count: number;
+}
+
+export interface ProducerConnectionSummary {
+ totalConnections: number;
+ uniqueClientCount: number;
+ uniqueAddressCount: number;
+ uniqueLanguageCount: number;
+ uniqueVersionCount: number;
+ languages: ProducerConnectionSummaryItem[];
+ versions: ProducerConnectionSummaryItem[];
+ duplicateClientIds: string[];
+ warnings: ProducerConnectionWarning[];
+ readiness: ProducerReadiness;
+}
+
+export interface ProducerConnectionResult {
+ connectionSet: ProducerConnection[];
+ summary: ProducerConnectionSummary;
+}
+
interface TopicRecord {
name: string;
}
@@ -34,8 +62,85 @@ interface TopicListResponse {
topicList?: string[];
}
+interface ProducerConnectionResponse {
+ connectionSet?: ProducerConnection[];
+ summary?: ProducerConnectionSummary;
+}
+
// ─── API ────────────────────────────────────────────────────────
+const hasText = (value?: string | null) => Boolean(value?.trim() &&
value.trim() !== 'null');
+
+const normalizeDimension = (value?: string | null) => (hasText(value) ?
value!.trim() : 'UNKNOWN');
+
+const countDistinct = (
+ connections: ProducerConnection[],
+ extractor: (connection: ProducerConnection) => string,
+) => new Set(connections.map(extractor).filter(hasText)).size;
+
+const distribution = (
+ connections: ProducerConnection[],
+ extractor: (connection: ProducerConnection) => string,
+): ProducerConnectionSummaryItem[] => {
+ const counts = new Map<string, number>();
+ connections.forEach((connection) => {
+ const value = normalizeDimension(extractor(connection));
+ counts.set(value, (counts.get(value) ?? 0) + 1);
+ });
+ return [...counts]
+ .map(([value, count]) => ({ value, count }))
+ .sort((a, b) => b.count - a.count || a.value.localeCompare(b.value));
+};
+
+export function buildProducerConnectionSummary(
+ connections: ProducerConnection[],
+): ProducerConnectionSummary {
+ const duplicateClientIds = [
+ ...connections.reduce((counts, connection) => {
+ const clientId = connection.clientId?.trim();
+ if (clientId) counts.set(clientId, (counts.get(clientId) ?? 0) + 1);
+ return counts;
+ }, new Map<string, number>()),
+ ]
+ .filter(([, count]) => count > 1)
+ .map(([clientId]) => clientId)
+ .sort();
+ const languages = distribution(connections, (connection) =>
connection.language);
+ const versions = distribution(connections, (connection) =>
connection.versionDesc);
+ const warnings: ProducerConnectionWarning[] = [];
+
+ if (connections.length === 0) {
+ warnings.push('NO_CONNECTIONS');
+ } else {
+ if (duplicateClientIds.length > 0) warnings.push('DUPLICATE_CLIENT_ID');
+ if (versions.length > 1) warnings.push('MIXED_CLIENT_VERSION');
+ if (
+ connections.some(
+ (connection) =>
+ !hasText(connection.clientId) ||
+ !hasText(connection.clientAddr) ||
+ !hasText(connection.language) ||
+ !hasText(connection.versionDesc),
+ )
+ ) {
+ warnings.push('INCOMPLETE_CLIENT_METADATA');
+ }
+ }
+
+ return {
+ totalConnections: connections.length,
+ uniqueClientCount: countDistinct(connections, (connection) =>
connection.clientId),
+ uniqueAddressCount: countDistinct(connections, (connection) =>
connection.clientAddr),
+ uniqueLanguageCount: languages.length,
+ uniqueVersionCount: versions.length,
+ languages,
+ versions,
+ duplicateClientIds,
+ warnings,
+ readiness: connections.length === 0 ? 'UNAVAILABLE' : warnings.length > 0
? 'WARNING' : 'READY',
+ };
+}
+
/** Fetch topic names for a managed instance. */
export async function fetchTopicList(instanceId: string): Promise<string[]> {
const res = await client.get<TopicListResponse>('/topics', { params: {
instanceId } });
@@ -54,9 +159,13 @@ export async function queryProducerConnection(
instanceId: string,
topic: string,
producerGroup: string,
-): Promise<ProducerConnection[]> {
- const res = await client.get<{ connectionSet: ProducerConnection[]
}>('/producer/connection', {
+): Promise<ProducerConnectionResult> {
+ const res = await
client.get<ProducerConnectionResponse>('/producer/connection', {
params: { instanceId, topic, producerGroup },
});
- return res.data?.connectionSet ?? [];
+ const connectionSet = res.data?.connectionSet ?? [];
+ return {
+ connectionSet,
+ summary: res.data?.summary ??
buildProducerConnectionSummary(connectionSet),
+ };
}
diff --git a/web/src/i18n/translations.ts b/web/src/i18n/translations.ts
index ec89bf46..e484c0b4 100644
--- a/web/src/i18n/translations.ts
+++ b/web/src/i18n/translations.ts
@@ -1161,6 +1161,33 @@ const translations: Record<string, Record<Lang, string>>
= {
en: 'Failed to fetch producer connections',
},
'producer.noConnections': { zh: '暂无生产者连接', en: 'No producer connections
found' },
+ 'producer.readiness': { zh: '生产者连接健康', en: 'Producer connection health' },
+ 'producer.readinessREADY': { zh: '就绪', en: 'Ready' },
+ 'producer.readinessWARNING': { zh: '存在风险', en: 'Warning' },
+ 'producer.readinessUNAVAILABLE': { zh: '无可用连接', en: 'Unavailable' },
+ 'producer.connectionTotal': { zh: '连接数', en: 'Connections' },
+ 'producer.uniqueClients': { zh: '唯一客户端', en: 'Unique clients' },
+ 'producer.uniqueAddresses': { zh: '唯一地址', en: 'Unique addresses' },
+ 'producer.languageKinds': { zh: '语言种类', en: 'Languages' },
+ 'producer.versionKinds': { zh: '版本种类', en: 'Versions' },
+ 'producer.languageDistribution': { zh: '语言分布', en: 'Language distribution' },
+ 'producer.versionDistribution': { zh: '版本分布', en: 'Version distribution' },
+ 'producer.warningNoConnections': {
+ zh: '未发现活跃连接',
+ en: 'No active producer connection',
+ },
+ 'producer.warningDuplicateClientId': {
+ zh: '存在重复 Client ID',
+ en: 'Duplicate client IDs found',
+ },
+ 'producer.warningMixedVersion': {
+ zh: '存在多个客户端版本',
+ en: 'Multiple client versions found',
+ },
+ 'producer.warningIncompleteMetadata': {
+ zh: '连接元数据不完整',
+ en: 'Incomplete connection metadata',
+ },
// ─── Namespace ───
'ns.title': { zh: '命名空间管理', en: 'Namespace Management' },
diff --git a/web/src/pages/studio/Producer.tsx
b/web/src/pages/studio/Producer.tsx
index 132ccd37..7f2a782c 100644
--- a/web/src/pages/studio/Producer.tsx
+++ b/web/src/pages/studio/Producer.tsx
@@ -16,7 +16,19 @@
*/
import { useEffect, useState } from 'react';
-import { App, AutoComplete, Button, Card, Form, Select, Table } from 'antd';
+import {
+ Alert,
+ App,
+ AutoComplete,
+ Button,
+ Card,
+ Flex,
+ Form,
+ Select,
+ Statistic,
+ Table,
+ Tag,
+} from 'antd';
import { MagnifyingGlass } from '@phosphor-icons/react';
import { useLang } from '../../i18n/LangContext';
import {
@@ -24,15 +36,27 @@ import {
fetchTopicList,
queryProducerConnection,
type ProducerConnection,
+ type ProducerConnectionSummary,
+ type ProducerConnectionWarning,
+ type ProducerReadiness,
} from '../../api/producer';
import type { Instance } from '../../api/instance';
import { listInstances } from '../../services/instanceService';
+const readinessConfig: Record<ProducerReadiness, { color: string; type:
'success' | 'warning' }> = {
+ READY: { color: 'success', type: 'success' },
+ WARNING: { color: 'warning', type: 'warning' },
+ UNAVAILABLE: { color: 'error', type: 'warning' },
+};
+
const ProducerPage = () => {
const [form] = Form.useForm();
const [topicList, setTopicList] = useState<string[]>([]);
const [producerGroups, setProducerGroups] = useState<string[]>([]);
const [connectionList, setConnectionList] =
useState<ProducerConnection[]>([]);
+ const [connectionSummary, setConnectionSummary] =
useState<ProducerConnectionSummary | null>(
+ null,
+ );
const [instances, setInstances] = useState<Instance[]>([]);
const [selectedInstanceId, setSelectedInstanceId] = useState('');
const [loading, setLoading] = useState(false);
@@ -129,12 +153,14 @@ const ProducerPage = () => {
}
setLoading(true);
try {
- const connections = await queryProducerConnection(
+ const result = await queryProducerConnection(
selectedInstanceId,
values.selectedTopic,
values.producerGroup,
);
+ const connections = result.connectionSet;
setConnectionList(connections);
+ setConnectionSummary(result.summary);
if (connections.length === 0) {
message.info(t('producer.noConnections'));
}
@@ -167,6 +193,26 @@ const ProducerPage = () => {
},
];
+ const warningLabel: Record<ProducerConnectionWarning, string> = {
+ NO_CONNECTIONS: t('producer.warningNoConnections'),
+ DUPLICATE_CLIENT_ID: t('producer.warningDuplicateClientId'),
+ MIXED_CLIENT_VERSION: t('producer.warningMixedVersion'),
+ INCOMPLETE_CLIENT_METADATA: t('producer.warningIncompleteMetadata'),
+ };
+
+ const renderDistribution = (items: ProducerConnectionSummary['languages']) =>
+ items.length === 0 ? (
+ <Tag>{t('common.noData')}</Tag>
+ ) : (
+ <Flex gap={4} wrap>
+ {items.map((item) => (
+ <Tag key={item.value}>
+ {item.value}: {item.count}
+ </Tag>
+ ))}
+ </Flex>
+ );
+
return (
<div style={{ padding: 0 }}>
<div
@@ -233,10 +279,69 @@ const ProducerPage = () => {
</Form.Item>
</Form>
+ {connectionSummary && (
+ <div style={{ marginBottom: 20 }}>
+ <Alert
+ showIcon
+ type={readinessConfig[connectionSummary.readiness].type}
+ message={
+ <Flex align="center" gap={8} wrap>
+ <span>{t('producer.readiness')}</span>
+ <Tag
color={readinessConfig[connectionSummary.readiness].color}>
+ {t(`producer.readiness${connectionSummary.readiness}`)}
+ </Tag>
+ {connectionSummary.warnings.map((warning) => (
+ <Tag key={warning} color="warning">
+ {warningLabel[warning] ?? warning}
+ </Tag>
+ ))}
+ </Flex>
+ }
+ style={{ marginBottom: 12 }}
+ />
+ <Flex gap={24} wrap style={{ marginBottom: 12 }}>
+ <Statistic
+ title={t('producer.connectionTotal')}
+ value={connectionSummary.totalConnections}
+ />
+ <Statistic
+ title={t('producer.uniqueClients')}
+ value={connectionSummary.uniqueClientCount}
+ />
+ <Statistic
+ title={t('producer.uniqueAddresses')}
+ value={connectionSummary.uniqueAddressCount}
+ />
+ <Statistic
+ title={t('producer.languageKinds')}
+ value={connectionSummary.uniqueLanguageCount}
+ />
+ <Statistic
+ title={t('producer.versionKinds')}
+ value={connectionSummary.uniqueVersionCount}
+ />
+ </Flex>
+ <Flex gap={16} wrap>
+ <div>
+ <div style={{ color: '#8c8c8c', fontSize: 12, marginBottom: 6
}}>
+ {t('producer.languageDistribution')}
+ </div>
+ {renderDistribution(connectionSummary.languages)}
+ </div>
+ <div>
+ <div style={{ color: '#8c8c8c', fontSize: 12, marginBottom: 6
}}>
+ {t('producer.versionDistribution')}
+ </div>
+ {renderDistribution(connectionSummary.versions)}
+ </div>
+ </Flex>
+ </div>
+ )}
+
<Table
dataSource={connectionList}
columns={columns}
- rowKey="clientId"
+ rowKey={(record) => `${record.clientId}:${record.clientAddr}`}
pagination={false}
bordered
size="middle"
diff --git a/web/src/pages/studio/__tests__/Producer.test.tsx
b/web/src/pages/studio/__tests__/Producer.test.tsx
index 5bc65599..22236aa8 100644
--- a/web/src/pages/studio/__tests__/Producer.test.tsx
+++ b/web/src/pages/studio/__tests__/Producer.test.tsx
@@ -22,6 +22,8 @@ import { App } from 'antd';
import { LangProvider } from '../../../i18n/LangContext';
import ProducerPage from '../Producer';
import {
+ type ProducerConnection,
+ type ProducerConnectionResult,
fetchProducerGroups,
fetchTopicList,
queryProducerConnection,
@@ -62,6 +64,22 @@ const renderWithProviders = (ui: React.ReactElement) => {
);
};
+const producerResult = (connectionSet: ProducerConnection[]):
ProducerConnectionResult => ({
+ connectionSet,
+ summary: {
+ totalConnections: connectionSet.length,
+ uniqueClientCount: new Set(connectionSet.map((connection) =>
connection.clientId)).size,
+ uniqueAddressCount: new Set(connectionSet.map((connection) =>
connection.clientAddr)).size,
+ uniqueLanguageCount: new Set(connectionSet.map((connection) =>
connection.language)).size,
+ uniqueVersionCount: new Set(connectionSet.map((connection) =>
connection.versionDesc)).size,
+ languages: connectionSet.map((connection) => ({ value:
connection.language, count: 1 })),
+ versions: connectionSet.map((connection) => ({ value:
connection.versionDesc, count: 1 })),
+ duplicateClientIds: [],
+ warnings: connectionSet.length === 0 ? ['NO_CONNECTIONS'] : [],
+ readiness: connectionSet.length === 0 ? 'UNAVAILABLE' : 'READY',
+ },
+});
+
describe('ProducerPage', () => {
beforeEach(() => {
vi.clearAllMocks();
@@ -80,7 +98,7 @@ describe('ProducerPage', () => {
]);
vi.mocked(fetchTopicList).mockResolvedValue(['order-events',
'payment-events']);
vi.mocked(fetchProducerGroups).mockResolvedValue(['pg-order',
'pg-payment']);
- vi.mocked(queryProducerConnection).mockResolvedValue([]);
+ vi.mocked(queryProducerConnection).mockResolvedValue(producerResult([]));
});
it('loads topic options after mount', async () => {
@@ -117,14 +135,16 @@ describe('ProducerPage', () => {
it('queries producer connections with the required topic and group', async
() => {
const user = userEvent.setup();
- vi.mocked(queryProducerConnection).mockResolvedValue([
- {
- clientId: 'producer-1',
- clientAddr: '192.168.1.10',
- language: 'JAVA',
- versionDesc: '5.1.0',
- },
- ]);
+ vi.mocked(queryProducerConnection).mockResolvedValue(
+ producerResult([
+ {
+ clientId: 'producer-1',
+ clientAddr: '192.168.1.10',
+ language: 'JAVA',
+ versionDesc: '5.1.0',
+ },
+ ]),
+ );
renderWithProviders(<ProducerPage />);
await waitFor(() => expect(fetchTopicList).toHaveBeenCalledTimes(1));
@@ -144,6 +164,8 @@ describe('ProducerPage', () => {
);
});
expect(await screen.findByText('producer-1')).toBeInTheDocument();
+ expect(await screen.findByText('生产者连接健康')).toBeInTheDocument();
+ expect(screen.getByText('就绪')).toBeInTheDocument();
});
it('does not query without a producer group', async () => {
@@ -164,6 +186,55 @@ describe('ProducerPage', () => {
expect(queryProducerConnection).not.toHaveBeenCalled();
});
+ it('renders producer connection warnings from the summary', async () => {
+ const user = userEvent.setup();
+ vi.mocked(queryProducerConnection).mockResolvedValue({
+ connectionSet: [
+ {
+ clientId: 'producer-a',
+ clientAddr: '192.168.1.10',
+ language: 'JAVA',
+ versionDesc: '5.1.0',
+ },
+ {
+ clientId: 'producer-a',
+ clientAddr: '192.168.1.11',
+ language: 'JAVA',
+ versionDesc: '5.2.0',
+ },
+ ],
+ summary: {
+ totalConnections: 2,
+ uniqueClientCount: 1,
+ uniqueAddressCount: 2,
+ uniqueLanguageCount: 1,
+ uniqueVersionCount: 2,
+ languages: [{ value: 'JAVA', count: 2 }],
+ versions: [
+ { value: '5.1.0', count: 1 },
+ { value: '5.2.0', count: 1 },
+ ],
+ duplicateClientIds: ['producer-a'],
+ warnings: ['DUPLICATE_CLIENT_ID', 'MIXED_CLIENT_VERSION'],
+ readiness: 'WARNING',
+ },
+ });
+ renderWithProviders(<ProducerPage />);
+
+ await waitFor(() => expect(fetchTopicList).toHaveBeenCalledTimes(1));
+ const [, topicSelect, groupInput] = screen.getAllByRole('combobox');
+ fireEvent.mouseDown(topicSelect.parentElement!);
+ await user.click(
+ await screen.findByText('order-events', { selector:
'.ant-select-item-option-content' }),
+ );
+ await user.type(groupInput, 'order-producer');
+ await user.click(screen.getByRole('button', { name: /搜索/ }));
+
+ expect(await screen.findByText('存在重复 Client ID')).toBeInTheDocument();
+ expect(screen.getByText('存在多个客户端版本')).toBeInTheDocument();
+ expect(screen.getByText('JAVA: 2')).toBeInTheDocument();
+ });
+
it('keeps manual producer group queries available when suggestions fail',
async () => {
vi.mocked(fetchProducerGroups).mockRejectedValue(new Error('broker
unavailable'));
const user = userEvent.setup();
@@ -213,14 +284,16 @@ describe('ProducerPage', () => {
},
]);
vi.mocked(fetchTopicList).mockResolvedValueOnce(['order-events']).mockResolvedValueOnce([]);
- vi.mocked(queryProducerConnection).mockResolvedValue([
- {
- clientId: 'producer-1',
- clientAddr: '192.168.1.10',
- language: 'JAVA',
- versionDesc: '5.1.0',
- },
- ]);
+ vi.mocked(queryProducerConnection).mockResolvedValue(
+ producerResult([
+ {
+ clientId: 'producer-1',
+ clientAddr: '192.168.1.10',
+ language: 'JAVA',
+ versionDesc: '5.1.0',
+ },
+ ]),
+ );
const user = userEvent.setup();
const { container } = renderWithProviders(<ProducerPage />);
