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 0849e3f1b feat(topic): add route topology diagnostics (#2836)
0849e3f1b is described below
commit 0849e3f1bb29752a0bacfd6b5aebc2b281e7ca38
Author: coder999o <[email protected]>
AuthorDate: Mon Aug 31 20:26:40 2026 +0800
feat(topic): add route topology diagnostics (#2836)
---
.../studio/instance/topic/BrokerRouteVO.java | 11 +
.../provider/apache/RocketMQMetadataProvider.java | 49 +-
.../studio/instance/topic/TopicControllerTest.java | 26 +-
.../apache/RocketMQMetadataProviderTest.java | 41 ++
web/src/api/metadata.ts | 8 +
web/src/mock/topics.ts | 30 ++
.../pages/instance/__tests__/TopicPage.test.tsx | 57 ++-
web/src/pages/instance/topic.tsx | 320 +++++++++++--
web/src/services/topicService.test.ts | 4 +
web/src/services/topicService.ts | 7 +-
web/src/utils/topicRouteDiagnostics.test.ts | 181 ++++++++
web/src/utils/topicRouteDiagnostics.ts | 509 +++++++++++++++++++++
12 files changed, 1193 insertions(+), 50 deletions(-)
diff --git
a/server/src/main/java/org/apache/rocketmq/studio/instance/topic/BrokerRouteVO.java
b/server/src/main/java/org/apache/rocketmq/studio/instance/topic/BrokerRouteVO.java
index 20234c808..46b0bbb96 100644
---
a/server/src/main/java/org/apache/rocketmq/studio/instance/topic/BrokerRouteVO.java
+++
b/server/src/main/java/org/apache/rocketmq/studio/instance/topic/BrokerRouteVO.java
@@ -22,6 +22,9 @@ import lombok.Builder;
import lombok.Data;
import lombok.NoArgsConstructor;
+import java.util.List;
+import java.util.Map;
+
@Data
@Builder
@NoArgsConstructor
@@ -29,7 +32,15 @@ import lombok.NoArgsConstructor;
public class BrokerRouteVO {
private String brokerName;
private String brokerAddr;
+ private String masterAddr;
+ private Map<Long, String> brokerAddrs;
+ private List<Long> brokerIds;
+ private int replicaCount;
private int writeQueues;
private int readQueues;
private TopicPerm perm;
+ private int permCode;
+ private boolean readable;
+ private boolean writable;
+ private int topicSysFlag;
}
diff --git
a/server/src/main/java/org/apache/rocketmq/studio/provider/apache/RocketMQMetadataProvider.java
b/server/src/main/java/org/apache/rocketmq/studio/provider/apache/RocketMQMetadataProvider.java
index d3ec8ce94..61a199623 100644
---
a/server/src/main/java/org/apache/rocketmq/studio/provider/apache/RocketMQMetadataProvider.java
+++
b/server/src/main/java/org/apache/rocketmq/studio/provider/apache/RocketMQMetadataProvider.java
@@ -66,6 +66,7 @@ import java.util.ArrayList;
import java.util.Collections;
import java.util.HashMap;
import java.util.HashSet;
+import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
import java.util.Set;
@@ -417,20 +418,24 @@ public class RocketMQMetadataProvider implements
MetadataProvider {
if (routeData.getQueueDatas() != null) {
for (QueueData qd : routeData.getQueueDatas()) {
BrokerData bd = brokerDataMap.get(qd.getBrokerName());
- String brokerAddr = "";
- if (bd != null && bd.getBrokerAddrs() != null &&
!bd.getBrokerAddrs().isEmpty()) {
- brokerAddr = bd.getBrokerAddrs().get(MixAll.MASTER_ID);
- if (brokerAddr == null) {
- brokerAddr =
bd.getBrokerAddrs().values().iterator().next();
- }
- }
+ Map<Long, String> brokerAddrs = orderedBrokerAddrs(bd);
+ String masterAddr = brokerAddrs.get(MixAll.MASTER_ID);
+ int perm = qd.getPerm();
routes.add(BrokerRouteVO.builder()
.brokerName(qd.getBrokerName())
- .brokerAddr(brokerAddr)
+ .brokerAddr(selectBrokerAddr(brokerAddrs))
+ .masterAddr(masterAddr)
+ .brokerAddrs(brokerAddrs)
+ .brokerIds(new ArrayList<>(brokerAddrs.keySet()))
+ .replicaCount(countReplicaAddrs(brokerAddrs))
.writeQueues(qd.getWriteQueueNums())
.readQueues(qd.getReadQueueNums())
- .perm(mapPerm(qd.getPerm()))
+ .perm(mapPerm(perm))
+ .permCode(perm)
+ .readable(PermName.isReadable(perm))
+ .writable(PermName.isWriteable(perm))
+ .topicSysFlag(qd.getTopicSysFlag())
.build());
}
}
@@ -441,6 +446,32 @@ public class RocketMQMetadataProvider implements
MetadataProvider {
}
}
+ private Map<Long, String> orderedBrokerAddrs(BrokerData brokerData) {
+ if (brokerData == null || brokerData.getBrokerAddrs() == null
+ || brokerData.getBrokerAddrs().isEmpty()) {
+ return Map.of();
+ }
+ Map<Long, String> sorted = new LinkedHashMap<>();
+ brokerData.getBrokerAddrs().entrySet().stream()
+ .sorted(Map.Entry.comparingByKey())
+ .forEach(entry -> sorted.put(entry.getKey(),
entry.getValue()));
+ return sorted;
+ }
+
+ private String selectBrokerAddr(Map<Long, String> brokerAddrs) {
+ String masterAddr = brokerAddrs.get(MixAll.MASTER_ID);
+ if (masterAddr != null) {
+ return masterAddr;
+ }
+ return brokerAddrs.values().stream().findFirst().orElse("");
+ }
+
+ private int countReplicaAddrs(Map<Long, String> brokerAddrs) {
+ return (int) brokerAddrs.keySet().stream()
+ .filter(id -> !Long.valueOf(MixAll.MASTER_ID).equals(id))
+ .count();
+ }
+
@Override
public List<TopicConsumerVO> getTopicConsumers(String instanceId, String
name) {
return getTopicConsumersPage(instanceId, name, 1,
TOPIC_CONSUMER_SCAN_LIMIT).getItems();
diff --git
a/server/src/test/java/org/apache/rocketmq/studio/instance/topic/TopicControllerTest.java
b/server/src/test/java/org/apache/rocketmq/studio/instance/topic/TopicControllerTest.java
index 4fa2b1c99..31792bef2 100644
---
a/server/src/test/java/org/apache/rocketmq/studio/instance/topic/TopicControllerTest.java
+++
b/server/src/test/java/org/apache/rocketmq/studio/instance/topic/TopicControllerTest.java
@@ -18,6 +18,7 @@
package org.apache.rocketmq.studio.instance.topic;
import com.fasterxml.jackson.databind.ObjectMapper;
+import org.apache.rocketmq.studio.common.domain.enums.TopicPerm;
import org.junit.jupiter.api.Test;
import org.mockito.ArgumentCaptor;
import org.springframework.beans.factory.annotation.Autowired;
@@ -92,12 +93,33 @@ class TopicControllerTest {
@Test
void topicRuntimeDiagnosticsShouldPassSelectedInstance() throws Exception {
- when(metadataService.getTopicRoutes("instance-a",
"orders")).thenReturn(List.of());
+ BrokerRouteVO route = BrokerRouteVO.builder()
+ .brokerName("broker-a")
+ .brokerAddr("10.0.0.1:10911")
+ .masterAddr("10.0.0.1:10911")
+ .brokerAddrs(Map.of(0L, "10.0.0.1:10911", 1L,
"10.0.0.2:10911"))
+ .brokerIds(List.of(0L, 1L))
+ .replicaCount(1)
+ .writeQueues(8)
+ .readQueues(8)
+ .perm(TopicPerm.RW)
+ .permCode(6)
+ .readable(true)
+ .writable(true)
+ .topicSysFlag(0)
+ .build();
+ when(metadataService.getTopicRoutes("instance-a",
"orders")).thenReturn(List.of(route));
when(metadataService.getTopicConsumers("instance-a",
"orders")).thenReturn(List.of());
mockMvc.perform(get("/api/topics/orders/routes").param("instanceId",
"instance-a"))
.andExpect(status().isOk())
- .andExpect(jsonPath("$.code").value(200));
+ .andExpect(jsonPath("$.code").value(200))
+ .andExpect(jsonPath("$.data[0].brokerName").value("broker-a"))
+
.andExpect(jsonPath("$.data[0].masterAddr").value("10.0.0.1:10911"))
+ .andExpect(jsonPath("$.data[0].brokerIds[1]").value(1))
+ .andExpect(jsonPath("$.data[0].replicaCount").value(1))
+ .andExpect(jsonPath("$.data[0].readable").value(true))
+ .andExpect(jsonPath("$.data[0].writable").value(true));
mockMvc.perform(get("/api/topics/orders/consumers").param("instanceId",
"instance-a"))
.andExpect(status().isOk())
.andExpect(jsonPath("$.code").value(200));
diff --git
a/server/src/test/java/org/apache/rocketmq/studio/provider/apache/RocketMQMetadataProviderTest.java
b/server/src/test/java/org/apache/rocketmq/studio/provider/apache/RocketMQMetadataProviderTest.java
index eec2cbe81..3fdd47f6b 100644
---
a/server/src/test/java/org/apache/rocketmq/studio/provider/apache/RocketMQMetadataProviderTest.java
+++
b/server/src/test/java/org/apache/rocketmq/studio/provider/apache/RocketMQMetadataProviderTest.java
@@ -280,6 +280,47 @@ class RocketMQMetadataProviderTest {
});
}
+ @Test
+ void getTopicRoutesShouldExposeBrokerTopologyDetails() throws Exception {
+ DefaultMQAdminExt admin = mock(DefaultMQAdminExt.class);
+ QueueData queueData = new QueueData();
+ queueData.setBrokerName("broker-a");
+ queueData.setReadQueueNums(4);
+ queueData.setWriteQueueNums(6);
+ queueData.setPerm(6);
+ queueData.setTopicSysFlag(1);
+
+ HashMap<Long, String> brokerAddrs = new LinkedHashMap<>();
+ brokerAddrs.put(1L, "10.0.0.2:10911");
+ brokerAddrs.put(0L, "10.0.0.1:10911");
+ brokerAddrs.put(2L, "10.0.0.3:10911");
+ BrokerData brokerData = new BrokerData();
+ brokerData.setBrokerName("broker-a");
+ brokerData.setBrokerAddrs(brokerAddrs);
+
+ TopicRouteData routeData = new TopicRouteData();
+ routeData.setQueueDatas(List.of(queueData));
+ routeData.setBrokerDatas(List.of(brokerData));
+ when(admin.examineTopicRouteInfo("TopicA")).thenReturn(routeData);
+
+ List<BrokerRouteVO> routes =
newLiveProvider(admin).getTopicRoutes(null, "TopicA");
+
+ assertThat(routes).singleElement().satisfies(route -> {
+ assertThat(route.getBrokerAddr()).isEqualTo("10.0.0.1:10911");
+ assertThat(route.getMasterAddr()).isEqualTo("10.0.0.1:10911");
+ assertThat(route.getBrokerAddrs()).containsExactly(
+ Map.entry(0L, "10.0.0.1:10911"),
+ Map.entry(1L, "10.0.0.2:10911"),
+ Map.entry(2L, "10.0.0.3:10911"));
+ assertThat(route.getBrokerIds()).containsExactly(0L, 1L, 2L);
+ assertThat(route.getReplicaCount()).isEqualTo(2);
+ assertThat(route.getPermCode()).isEqualTo(6);
+ assertThat(route.isReadable()).isTrue();
+ assertThat(route.isWritable()).isTrue();
+ assertThat(route.getTopicSysFlag()).isEqualTo(1);
+ });
+ }
+
@Test
void getTopicConsumersShouldUseSelectedInstanceRuntimeClient() {
TopicConsumerPageVO consumers = TopicConsumerPageVO.builder()
diff --git a/web/src/api/metadata.ts b/web/src/api/metadata.ts
index 4096cd38d..118fdf81d 100644
--- a/web/src/api/metadata.ts
+++ b/web/src/api/metadata.ts
@@ -33,9 +33,17 @@ export interface TopicQuery {
export interface BrokerRoute {
brokerName: string;
brokerAddr: string;
+ masterAddr?: string;
+ brokerAddrs?: Record<string, string>;
+ brokerIds?: number[];
+ replicaCount?: number;
writeQueues: number;
readQueues: number;
perm: string;
+ permCode?: number;
+ readable?: boolean;
+ writable?: boolean;
+ topicSysFlag?: number;
}
export interface ConsumerGroupInfo {
diff --git a/web/src/mock/topics.ts b/web/src/mock/topics.ts
index 7fe18df0d..22c0edd40 100644
--- a/web/src/mock/topics.ts
+++ b/web/src/mock/topics.ts
@@ -240,9 +240,17 @@ export const topics: Topic[] = [
export interface BrokerRoute {
brokerName: string;
brokerAddr: string;
+ masterAddr?: string;
+ brokerAddrs?: Record<string, string>;
+ brokerIds?: number[];
+ replicaCount?: number;
writeQueues: number;
readQueues: number;
perm: string;
+ permCode?: number;
+ readable?: boolean;
+ writable?: boolean;
+ topicSysFlag?: number;
}
export const topicRoutes: Record<string, BrokerRoute[]> = {
@@ -250,16 +258,38 @@ export const topicRoutes: Record<string, BrokerRoute[]> =
{
{
brokerName: 'broker-a-0',
brokerAddr: '10.0.1.10:10911',
+ masterAddr: '10.0.1.10:10911',
+ brokerAddrs: {
+ '0': '10.0.1.10:10911',
+ '1': '10.0.1.12:10911',
+ },
+ brokerIds: [0, 1],
+ replicaCount: 1,
writeQueues: 8,
readQueues: 8,
perm: 'RW',
+ permCode: 6,
+ readable: true,
+ writable: true,
+ topicSysFlag: 0,
},
{
brokerName: 'broker-b-0',
brokerAddr: '10.0.1.11:10911',
+ masterAddr: '10.0.1.11:10911',
+ brokerAddrs: {
+ '0': '10.0.1.11:10911',
+ '1': '10.0.1.13:10911',
+ },
+ brokerIds: [0, 1],
+ replicaCount: 1,
writeQueues: 8,
readQueues: 8,
perm: 'RW',
+ permCode: 6,
+ readable: true,
+ writable: true,
+ topicSysFlag: 0,
},
],
'user-activity-log': [
diff --git a/web/src/pages/instance/__tests__/TopicPage.test.tsx
b/web/src/pages/instance/__tests__/TopicPage.test.tsx
index 41ff78f62..df2e54cd0 100644
--- a/web/src/pages/instance/__tests__/TopicPage.test.tsx
+++ b/web/src/pages/instance/__tests__/TopicPage.test.tsx
@@ -21,7 +21,7 @@ import userEvent from '@testing-library/user-event';
import { MemoryRouter, Route, Routes } from 'react-router-dom';
import { App } from 'antd';
import { LangProvider } from '../../../i18n/LangContext';
-import type { Topic } from '../../../api/metadata';
+import type { BrokerRoute, Topic } from '../../../api/metadata';
import { parseMessageProperties } from '../../../utils/messageProperties';
import TopicPage from '../topic';
@@ -342,6 +342,61 @@ describe('TopicPage', () => {
expect(topicServiceMocks.getTopicRoutes).toHaveBeenLastCalledWith('topic-01',
'instance-a');
});
+ it('renders topic route health diagnostics in the detail modal', async () =>
{
+ const user = userEvent.setup();
+ const routes: BrokerRoute[] = [
+ {
+ brokerName: 'broker-a',
+ brokerAddr: '',
+ masterAddr: '',
+ brokerAddrs: {
+ '1': '10.0.0.2:10911',
+ },
+ brokerIds: [1],
+ replicaCount: 1,
+ writeQueues: 12,
+ readQueues: 0,
+ perm: 'WO',
+ permCode: 2,
+ readable: false,
+ writable: true,
+ topicSysFlag: 0,
+ },
+ {
+ brokerName: 'broker-b',
+ brokerAddr: '10.0.0.3:10911',
+ masterAddr: '10.0.0.3:10911',
+ brokerAddrs: {
+ '0': '10.0.0.3:10911',
+ },
+ brokerIds: [0],
+ replicaCount: 0,
+ writeQueues: 2,
+ readQueues: 8,
+ perm: 'RW',
+ permCode: 6,
+ readable: true,
+ writable: true,
+ topicSysFlag: 0,
+ },
+ ];
+ mockTopicsList([buildTopics(1)[0]]);
+ topicServiceMocks.getTopicRoutes.mockResolvedValue(routes);
+ renderWithProviders();
+
+ await user.click(await screen.findByRole('button', { name: /详情/ }));
+
+ expect(await screen.findByText('路由诊断：不可用')).toBeInTheDocument();
+ expect(screen.getByText('可写 Broker')).toBeInTheDocument();
+ expect(screen.getByText('14 个写队列')).toBeInTheDocument();
+ expect(screen.getByText('Replica 1')).toBeInTheDocument();
+ expect(screen.getAllByText('写队列分布不均').length).toBeGreaterThan(0);
+ expect(screen.getByText('broker-a：读队列不可用')).toBeInTheDocument();
+ expect(
+ screen.getByText('检查 Broker 是否仍向 NameServer 注册，并确认 master 节点可达。'),
+ ).toBeInTheDocument();
+ });
+
it('keeps failed topics selected after a partially successful batch
deletion', async () => {
const user = userEvent.setup();
mockTopicsList(buildTopics(3));
diff --git a/web/src/pages/instance/topic.tsx b/web/src/pages/instance/topic.tsx
index 58d4961fc..a93506ae1 100644
--- a/web/src/pages/instance/topic.tsx
+++ b/web/src/pages/instance/topic.tsx
@@ -40,6 +40,7 @@ import {
Spin,
message,
App,
+ Progress,
} from 'antd';
import type { TableColumnsType } from 'antd';
import {
@@ -52,6 +53,9 @@ import {
SyncOutlined,
PlusCircleOutlined,
MinusCircleOutlined,
+ CheckCircleOutlined,
+ ExclamationCircleOutlined,
+ WarningOutlined,
} from '@ant-design/icons';
import PageHeader from '../../components/PageHeader';
import InfoBanner from '../../components/InfoBanner';
@@ -81,6 +85,12 @@ import {
import { buildCsv, downloadCsv, type CsvColumn } from '../../utils/download';
import { parseMessageProperties } from '../../utils/messageProperties';
import { tableScrollX } from '../../utils/table';
+import {
+ analyzeTopicRoutes,
+ type RouteDiagnosticIssue,
+ type RouteDiagnosticStatus,
+ type RouteDistribution,
+} from '../../utils/topicRouteDiagnostics';
const { Text } = Typography;
@@ -288,6 +298,22 @@ const formatDateTime = (iso?: string): string => {
return `${d.getFullYear()}-${pad(d.getMonth() + 1)}-${pad(d.getDate())}
${pad(d.getHours())}:${pad(d.getMinutes())}:${pad(d.getSeconds())}`;
};
+const ROUTE_STATUS_META: Record<
+ RouteDiagnosticStatus,
+ { color: string; label: string; icon: React.ReactNode }
+> = {
+ healthy: { color: 'success', label: '健康', icon: <CheckCircleOutlined /> },
+ warning: { color: 'warning', label: '关注', icon: <WarningOutlined /> },
+ critical: { color: 'error', label: '异常', icon: <ExclamationCircleOutlined />
},
+};
+
+const ISSUE_SEVERITY_COLOR: Record<RouteDiagnosticIssue['severity'], string> =
{
+ warning: 'warning',
+ critical: 'error',
+};
+
+const formatPercent = (value: number) => `${value.toFixed(value % 1 === 0 ? 0
: 1)}%`;
+
// ═══════════════════════════════════════════════════════════════════
const TopicPage = () => {
const { t } = useLang();
@@ -686,17 +712,114 @@ const TopicPage = () => {
},
];
+ const renderRouteStatusTag = (status: RouteDiagnosticStatus) => {
+ const meta = ROUTE_STATUS_META[status];
+ return (
+ <Tag color={meta.color} icon={meta.icon}>
+ {meta.label}
+ </Tag>
+ );
+ };
+
+ const renderRouteIssueTags = (issues: RouteDiagnosticIssue[]) => {
+ if (issues.length === 0) return <Text type="secondary">无</Text>;
+ return (
+ <Space size={[4, 4]} wrap>
+ {issues.slice(0, 3).map((item) => (
+ <Tag key={item.id} color={ISSUE_SEVERITY_COLOR[item.severity]}>
+ {item.title}
+ </Tag>
+ ))}
+ {issues.length > 3 && <Tag>+{issues.length - 3}</Tag>}
+ </Space>
+ );
+ };
+
// ─── Route table columns ──────────────────────────────────────
- const routeColumns: TableColumnsType<BrokerRoute> = [
- { title: 'Broker 名称', dataIndex: 'brokerName', key: 'brokerName' },
- { title: 'Broker 地址', dataIndex: 'brokerAddr', key: 'brokerAddr' },
- { title: '写队列', dataIndex: 'writeQueues', key: 'writeQueues' },
- { title: '读队列', dataIndex: 'readQueues', key: 'readQueues' },
+ const routeColumns: TableColumnsType<RouteDistribution> = [
+ {
+ title: 'Broker',
+ dataIndex: 'brokerName',
+ key: 'brokerName',
+ width: 170,
+ render: (_: string, record) => (
+ <Space direction="vertical" size={2}>
+ <Text strong>{record.brokerName}</Text>
+ {renderRouteStatusTag(record.status)}
+ </Space>
+ ),
+ },
+ {
+ title: '地址拓扑',
+ key: 'brokerAddr',
+ width: 260,
+ render: (_: unknown, record) => (
+ <Space direction="vertical" size={2} style={{ width: '100%' }}>
+ <Text code copyable style={{ fontSize: 14 }}>
+ {record.brokerAddr}
+ </Text>
+ {record.masterAddr && record.masterAddr !== record.brokerAddr && (
+ <Text type="secondary" style={{ fontSize: 14 }}>
+ Master {record.masterAddr}
+ </Text>
+ )}
+ <Space size={4} wrap>
+ {record.brokerIds.length > 0 ? (
+ record.brokerIds.map((id) => (
+ <Tag key={id} color={id === '0' ? 'blue' : undefined}>
+ {id === '0' ? 'Master' : `Replica ${id}`}
+ </Tag>
+ ))
+ ) : (
+ <Tag color="warning">地址未知</Tag>
+ )}
+ </Space>
+ </Space>
+ ),
+ },
+ {
+ title: '队列分布',
+ key: 'queues',
+ width: 220,
+ render: (_: unknown, record) => (
+ <Space direction="vertical" size={4} style={{ width: '100%' }}>
+ <div>
+ <Flex justify="space-between">
+ <Text>写队列 {record.writeQueues}</Text>
+ <Text type="secondary">{formatPercent(record.writeShare)}</Text>
+ </Flex>
+ <Progress percent={record.writeShare} showInfo={false}
size="small" />
+ </div>
+ <div>
+ <Flex justify="space-between">
+ <Text>读队列 {record.readQueues}</Text>
+ <Text type="secondary">{formatPercent(record.readShare)}</Text>
+ </Flex>
+ <Progress percent={record.readShare} showInfo={false} size="small"
/>
+ </div>
+ </Space>
+ ),
+ },
{
title: '权限',
dataIndex: 'perm',
key: 'perm',
- render: (p: string) => <Tag>{PERM_LABEL[p] || p}</Tag>,
+ width: 130,
+ render: (_: string, record) => (
+ <Space direction="vertical" size={4}>
+ <Tag>{PERM_LABEL[record.perm] || record.perm}</Tag>
+ <Space size={4}>
+ <Tag color={record.readable ? 'success' : 'error'}>读</Tag>
+ <Tag color={record.writable ? 'success' : 'error'}>写</Tag>
+ </Space>
+ </Space>
+ ),
+ },
+ {
+ title: '诊断',
+ key: 'diagnostics',
+ width: 220,
+ render: (_: unknown, record) => renderRouteIssueTags(record.issues),
},
];
@@ -747,6 +870,158 @@ const TopicPage = () => {
},
];
+ const renderRouteMetric = (label: string, value: React.ReactNode, extra?:
React.ReactNode) => (
+ <Col xs={12} md={6}>
+ <div
+ style={{
+ border: '1px solid #f0f0f0',
+ borderRadius: 6,
+ padding: '10px 12px',
+ minHeight: 78,
+ background: '#fafafa',
+ }}
+ >
+ <Text type="secondary" style={{ display: 'block', fontSize: 14 }}>
+ {label}
+ </Text>
+ <Text strong style={{ fontSize: 20, fontVariantNumeric: 'tabular-nums'
}}>
+ {value}
+ </Text>
+ {extra && (
+ <div style={{ marginTop: 2 }}>
+ <Text type="secondary" style={{ fontSize: 14 }}>
+ {extra}
+ </Text>
+ </div>
+ )}
+ </div>
+ </Col>
+ );
+
+ const renderRouteIssues = (issues: RouteDiagnosticIssue[]) => {
+ if (issues.length === 0) return null;
+ return (
+ <div
+ data-testid="topic-route-issues"
+ style={{ border: '1px solid #f0f0f0', borderRadius: 6, padding: 12 }}
+ >
+ <Text strong style={{ display: 'block', marginBottom: 8 }}>
+ 诊断项
+ </Text>
+ <Space direction="vertical" size={8} style={{ width: '100%' }}>
+ {issues.map((item) => (
+ <Flex key={item.id} align="flex-start" gap={8}>
+ <Tag color={ISSUE_SEVERITY_COLOR[item.severity]} style={{
marginTop: 1 }}>
+ {item.severity === 'critical' ? '异常' : '关注'}
+ </Tag>
+ <div>
+ <Text strong>
+ {item.brokerName ? `${item.brokerName}：${item.title}` :
item.title}
+ </Text>
+ <Text type="secondary" style={{ display: 'block' }}>
+ {item.description}
+ </Text>
+ </div>
+ </Flex>
+ ))}
+ </Space>
+ </div>
+ );
+ };
+
+ const renderRouteRecommendations = (recommendations: string[]) => {
+ if (recommendations.length === 0) return null;
+ return (
+ <InfoBanner
+ title="建议处理"
+ description={
+ <Space direction="vertical" size={2}>
+ {recommendations.map((item) => (
+ <Text key={item} style={{ fontSize: 14 }}>
+ {item}
+ </Text>
+ ))}
+ </Space>
+ }
+ />
+ );
+ };
+
+ const renderRouteSection = (topic: Topic) => {
+ const routes = getRoutes(topic.name);
+ const diagnostics = analyzeTopicRoutes(routes);
+ const summary = diagnostics.summary;
+
+ return (
+ <>
+ <Text strong style={{ fontSize: 14, display: 'block', marginBottom: 12
}}>
+ 路由信息
+ </Text>
+ {!detailLoading && (
+ <Space direction="vertical" size={12} style={{ width: '100%',
marginBottom: 12 }}>
+ <Alert
+ type={diagnostics.statusColor}
+ showIcon
+ message={`路由诊断：${diagnostics.statusText}`}
+ description={
+ diagnostics.status === 'healthy'
+ ? `共 ${summary.brokerCount} 个 Broker，写队列
${summary.totalWriteQueues} 个，读队列 ${summary.totalReadQueues} 个。`
+ : `发现 ${diagnostics.issues.length} 个诊断项，优先处理异常标记的 Broker。`
+ }
+ action={
+ routes.length === 0 ? (
+ <Button
+ size="small"
+ type="primary"
+ loading={rebuilding}
+ onClick={() => void rebuildTopic(topic)}
+ >
+ 在 Broker 上重建
+ </Button>
+ ) : undefined
+ }
+ />
+ <Row gutter={[12, 12]}>
+ {renderRouteMetric(
+ 'Broker 数',
+ summary.brokerCount,
+ `${summary.addressCount} 个地址`,
+ )}
+ {renderRouteMetric(
+ '可写 Broker',
+ summary.writableBrokerCount,
+ `${summary.totalWriteQueues} 个写队列`,
+ )}
+ {renderRouteMetric(
+ '可读 Broker',
+ summary.readableBrokerCount,
+ `${summary.totalReadQueues} 个读队列`,
+ )}
+ {renderRouteMetric(
+ 'Replica 数',
+ summary.replicaCount,
+ summary.writeSkew.gap > 0 || summary.readSkew.gap > 0
+ ? `队列差距 写 ${summary.writeSkew.gap} / 读
${summary.readSkew.gap}`
+ : '队列均衡',
+ )}
+ </Row>
+ {renderRouteIssues(diagnostics.issues)}
+ {renderRouteRecommendations(diagnostics.recommendations)}
+ </Space>
+ )}
+ <Table<RouteDistribution>
+ columns={routeColumns}
+ dataSource={detailLoading ? [] : diagnostics.distributions}
+ rowKey="key"
+ pagination={false}
+ size="small"
+ loading={detailLoading}
+ scroll={{ x: tableScrollX(routeColumns) }}
+ />
+ </>
+ );
+ };
+
// ─── Modal: detail tab ────────────────────────────────────────
const renderDetailTab = (topic: Topic) => {
const cluster = CLUSTER_NAME_MAP[topic.clusterId];
@@ -1163,7 +1438,7 @@ const TopicPage = () => {
title={selectedTopic?.name}
open={detailModalOpen}
onCancel={() => setDetailModalOpen(false)}
- width={800}
+ width={1080}
destroyOnHidden
footer={null}
>
@@ -1180,36 +1455,7 @@ const TopicPage = () => {
<Divider style={{ margin: '20px 0 16px' }} />
{/* Section 2: 路由信息 */}
- <Text strong style={{ fontSize: 14, display: 'block',
marginBottom: 12 }}>
- 路由信息
- </Text>
- {!detailLoading && getRoutes(selectedTopic.name).length === 0
&& (
- <Alert
- type="warning"
- showIcon
- style={{ marginBottom: 12 }}
- message="Broker 上没有该 Topic 的路由"
- description="元数据库中存在这条记录，但 Broker 未返回路由信息，可能尚未在 Broker
上创建或已被删除。可按库中记录的队列数重建。"
- action={
- <Button
- size="small"
- type="primary"
- loading={rebuilding}
- onClick={() => void rebuildTopic(selectedTopic)}
- >
- 在 Broker 上重建
- </Button>
- }
- />
- )}
- <Table<BrokerRoute>
- columns={routeColumns}
- dataSource={getRoutes(selectedTopic.name)}
- rowKey="brokerName"
- pagination={false}
- size="small"
- loading={detailLoading}
- />
+ {renderRouteSection(selectedTopic)}
</>
)}
diff --git a/web/src/services/topicService.test.ts
b/web/src/services/topicService.test.ts
index b6a381e8e..43af8c550 100644
--- a/web/src/services/topicService.test.ts
+++ b/web/src/services/topicService.test.ts
@@ -46,9 +46,13 @@ describe('topic service mock data', () => {
expect(first[0].brokerName).toBe('broker-a-0');
first[0].brokerName = 'mutated-broker';
+ if (first[0].brokerAddrs) first[0].brokerAddrs['0'] = '127.0.0.1:10911';
+ if (first[0].brokerIds) first[0].brokerIds.push(99);
const second = await getTopicRoutes('order-create');
expect(second[0].brokerName).toBe('broker-a-0');
+ expect(second[0].brokerAddrs?.['0']).toBe('10.0.1.10:10911');
+ expect(second[0].brokerIds).toEqual([0, 1]);
expect(second[0]).not.toBe(first[0]);
});
diff --git a/web/src/services/topicService.ts b/web/src/services/topicService.ts
index 5843c2e63..2423dc36c 100644
--- a/web/src/services/topicService.ts
+++ b/web/src/services/topicService.ts
@@ -16,7 +16,12 @@ const EXPORT_PAGE_SIZE = 100;
const MAX_EXPORT_PAGES = 100;
const cloneTopic = (topic: Topic): Topic => ({ ...topic });
-const cloneRoutes = (routes: BrokerRoute[]): BrokerRoute[] =>
routes.map((route) => ({ ...route }));
+const cloneRoutes = (routes: BrokerRoute[]): BrokerRoute[] =>
+ routes.map((route) => ({
+ ...route,
+ brokerAddrs: route.brokerAddrs ? { ...route.brokerAddrs } : undefined,
+ brokerIds: route.brokerIds ? [...route.brokerIds] : undefined,
+ }));
const cloneConsumers = (consumers: ConsumerGroupInfo[]): ConsumerGroupInfo[] =>
consumers.map((consumer) => ({ ...consumer }));
diff --git a/web/src/utils/topicRouteDiagnostics.test.ts
b/web/src/utils/topicRouteDiagnostics.test.ts
new file mode 100644
index 000000000..fd05fdec4
--- /dev/null
+++ b/web/src/utils/topicRouteDiagnostics.test.ts
@@ -0,0 +1,181 @@
+/*
+ * 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.
+ */
+
+import { describe, expect, it } from 'vitest';
+import type { BrokerRoute } from '../api/metadata';
+import { analyzeTopicRoutes } from './topicRouteDiagnostics';
+
+const route = (overrides: Partial<BrokerRoute>): BrokerRoute => ({
+ brokerName: 'broker-a',
+ brokerAddr: '10.0.0.1:10911',
+ masterAddr: '10.0.0.1:10911',
+ brokerAddrs: {
+ '0': '10.0.0.1:10911',
+ '1': '10.0.0.2:10911',
+ },
+ brokerIds: [0, 1],
+ replicaCount: 1,
+ writeQueues: 8,
+ readQueues: 8,
+ perm: 'RW',
+ permCode: 6,
+ readable: true,
+ writable: true,
+ topicSysFlag: 0,
+ ...overrides,
+});
+
+describe('topic route diagnostics', () => {
+ it('summarizes balanced readable and writable routes as healthy', () => {
+ const diagnostics = analyzeTopicRoutes([
+ route({ brokerName: 'broker-a' }),
+ route({
+ brokerName: 'broker-b',
+ brokerAddr: '10.0.1.1:10911',
+ masterAddr: '10.0.1.1:10911',
+ brokerAddrs: {
+ '0': '10.0.1.1:10911',
+ '1': '10.0.1.2:10911',
+ },
+ }),
+ ]);
+
+ expect(diagnostics.status).toBe('healthy');
+ expect(diagnostics.summary).toMatchObject({
+ brokerCount: 2,
+ addressCount: 4,
+ replicaCount: 2,
+ writableBrokerCount: 2,
+ readableBrokerCount: 2,
+ totalWriteQueues: 16,
+ totalReadQueues: 16,
+ });
+ expect(diagnostics.distributions.map((item) =>
item.writeShare)).toEqual([50, 50]);
+ expect(diagnostics.issues).toEqual([]);
+ expect(diagnostics.recommendations).toEqual([]);
+ });
+
+ it('returns a critical diagnostic when the broker route is missing', () => {
+ const diagnostics = analyzeTopicRoutes([]);
+
+ expect(diagnostics.status).toBe('critical');
+ expect(diagnostics.summary.brokerCount).toBe(0);
+ expect(diagnostics.distributions).toEqual([]);
+ expect(diagnostics.issues).toEqual([
+ expect.objectContaining({
+ code: 'NO_ROUTE',
+ severity: 'critical',
+ }),
+ ]);
+ expect(diagnostics.recommendations).toContain(
+ '确认 Topic 已在目标 Broker 上创建；必要时使用“在 Broker 上重建”。',
+ );
+ });
+
+ it('flags route risks from permissions, queues, skew, and stale addresses',
() => {
+ const diagnostics = analyzeTopicRoutes([
+ route({
+ brokerName: 'broker-a',
+ brokerAddr: '',
+ masterAddr: '',
+ brokerAddrs: {
+ '1': '10.0.0.2:10911',
+ },
+ brokerIds: [1],
+ writeQueues: 12,
+ readQueues: 0,
+ perm: 'WO',
+ permCode: 2,
+ readable: false,
+ writable: true,
+ }),
+ route({
+ brokerName: 'broker-b',
+ brokerAddr: '10.0.0.2:10911',
+ masterAddr: '10.0.0.2:10911',
+ brokerAddrs: {
+ '0': '10.0.0.2:10911',
+ },
+ brokerIds: [0],
+ writeQueues: 2,
+ readQueues: 8,
+ }),
+ ]);
+
+ const issueCodes = diagnostics.issues.map((item) => item.code);
+
+ expect(diagnostics.status).toBe('critical');
+ expect(diagnostics.summary.writeSkew).toEqual({ gap: 10, ratio: 1.43 });
+ expect(diagnostics.summary.readSkew).toEqual({ gap: 0, ratio: 0 });
+ expect(issueCodes).toEqual(
+ expect.arrayContaining([
+ 'MISSING_MASTER_ADDRESS',
+ 'READ_QUEUE_UNAVAILABLE',
+ 'PERMISSION_NOT_READABLE',
+ 'READ_WRITE_QUEUE_MISMATCH',
+ 'WRITE_QUEUE_SKEW',
+ 'DUPLICATE_BROKER_ADDRESS',
+ ]),
+ );
+ expect(diagnostics.distributions[0]).toMatchObject({
+ brokerName: 'broker-a',
+ brokerAddr: '10.0.0.2:10911',
+ readable: false,
+ writable: true,
+ status: 'critical',
+ });
+ expect(diagnostics.recommendations).toEqual(
+ expect.arrayContaining([
+ '检查 Broker 是否仍向 NameServer 注册，并确认 master 节点可达。',
+ '对比各 Broker 上的 TopicConfig，统一读写队列数后再观察客户端路由。',
+ '评估是否需要扩容、迁移或重新分配队列，降低单 Broker 负载集中风险。',
+ ]),
+ );
+ });
+
+ it('infers read and write permissions from legacy route payloads', () => {
+ const diagnostics = analyzeTopicRoutes([
+ route({
+ brokerName: 'broker-readonly',
+ readable: undefined,
+ writable: undefined,
+ perm: 'RO',
+ }),
+ route({
+ brokerName: 'broker-writeonly',
+ readable: undefined,
+ writable: undefined,
+ perm: 'WO',
+ }),
+ ]);
+
+ expect(diagnostics.status).toBe('warning');
+ expect(diagnostics.distributions[0]).toMatchObject({
+ brokerName: 'broker-readonly',
+ readable: true,
+ writable: false,
+ });
+ expect(diagnostics.distributions[1]).toMatchObject({
+ brokerName: 'broker-writeonly',
+ readable: false,
+ writable: true,
+ });
+ expect(diagnostics.issues.map((item) => item.code)).toEqual(
+ expect.arrayContaining(['PERMISSION_NOT_WRITABLE',
'PERMISSION_NOT_READABLE']),
+ );
+ });
+});
diff --git a/web/src/utils/topicRouteDiagnostics.ts
b/web/src/utils/topicRouteDiagnostics.ts
new file mode 100644
index 000000000..d273dd15e
--- /dev/null
+++ b/web/src/utils/topicRouteDiagnostics.ts
@@ -0,0 +1,509 @@
+/*
+ * 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.
+ */
+
+import type { BrokerRoute } from '../api/metadata';
+
+export type RouteDiagnosticStatus = 'healthy' | 'warning' | 'critical';
+
+export type RouteIssueCode =
+ | 'NO_ROUTE'
+ | 'MISSING_BROKER_ADDRESS'
+ | 'MISSING_MASTER_ADDRESS'
+ | 'NO_WRITABLE_ROUTE'
+ | 'NO_READABLE_ROUTE'
+ | 'WRITE_QUEUE_UNAVAILABLE'
+ | 'READ_QUEUE_UNAVAILABLE'
+ | 'PERMISSION_NOT_WRITABLE'
+ | 'PERMISSION_NOT_READABLE'
+ | 'READ_WRITE_QUEUE_MISMATCH'
+ | 'WRITE_QUEUE_SKEW'
+ | 'READ_QUEUE_SKEW'
+ | 'SINGLE_BROKER_ROUTE'
+ | 'DUPLICATE_BROKER_ADDRESS';
+
+export interface RouteDiagnosticIssue {
+ id: string;
+ code: RouteIssueCode;
+ severity: Exclude<RouteDiagnosticStatus, 'healthy'>;
+ title: string;
+ description: string;
+ brokerName?: string;
+}
+
+export interface RouteQueueSkew {
+ gap: number;
+ ratio: number;
+}
+
+export interface RouteDistribution {
+ key: string;
+ brokerName: string;
+ brokerAddr: string;
+ masterAddr: string;
+ brokerIds: string[];
+ replicaCount: number;
+ writeQueues: number;
+ readQueues: number;
+ writeShare: number;
+ readShare: number;
+ perm: string;
+ readable: boolean;
+ writable: boolean;
+ topicSysFlag?: number;
+ status: RouteDiagnosticStatus;
+ issues: RouteDiagnosticIssue[];
+}
+
+export interface RouteDiagnosticsSummary {
+ brokerCount: number;
+ addressCount: number;
+ replicaCount: number;
+ writableBrokerCount: number;
+ readableBrokerCount: number;
+ totalWriteQueues: number;
+ totalReadQueues: number;
+ writeSkew: RouteQueueSkew;
+ readSkew: RouteQueueSkew;
+}
+
+export interface TopicRouteDiagnostics {
+ status: RouteDiagnosticStatus;
+ statusText: string;
+ statusColor: 'success' | 'warning' | 'error';
+ summary: RouteDiagnosticsSummary;
+ distributions: RouteDistribution[];
+ issues: RouteDiagnosticIssue[];
+ recommendations: string[];
+}
+
+const STATUS_ORDER: Record<RouteDiagnosticStatus, number> = {
+ healthy: 0,
+ warning: 1,
+ critical: 2,
+};
+
+const STATUS_TEXT: Record<RouteDiagnosticStatus, string> = {
+ healthy: '路由健康',
+ warning: '需要关注',
+ critical: '不可用',
+};
+
+const STATUS_COLOR: Record<RouteDiagnosticStatus, 'success' | 'warning' |
'error'> = {
+ healthy: 'success',
+ warning: 'warning',
+ critical: 'error',
+};
+
+const EMPTY_SKEW: RouteQueueSkew = { gap: 0, ratio: 0 };
+
+const issue = (
+ code: RouteIssueCode,
+ severity: Exclude<RouteDiagnosticStatus, 'healthy'>,
+ title: string,
+ description: string,
+ brokerName?: string,
+): RouteDiagnosticIssue => ({
+ id: brokerName ? `${brokerName}:${code}` : code,
+ code,
+ severity,
+ title,
+ description,
+ brokerName,
+});
+
+const queueCount = (value: number | undefined): number =>
+ Number.isFinite(value) && value && value > 0 ? value : 0;
+
+const queueShare = (value: number, total: number): number =>
+ total <= 0 ? 0 : Math.round((value / total) * 1000) / 10;
+
+const inferReadable = (route: BrokerRoute): boolean => {
+ if (typeof route.readable === 'boolean') return route.readable;
+ return route.perm === 'RW' || route.perm === 'RO';
+};
+
+const inferWritable = (route: BrokerRoute): boolean => {
+ if (typeof route.writable === 'boolean') return route.writable;
+ return route.perm === 'RW' || route.perm === 'WO';
+};
+
+const routeBrokerIds = (route: BrokerRoute): string[] => {
+ if (route.brokerIds && route.brokerIds.length > 0) {
+ return route.brokerIds.map(String).sort((left, right) => Number(left) -
Number(right));
+ }
+ if (route.brokerAddrs) {
+ return Object.keys(route.brokerAddrs).sort((left, right) => Number(left) -
Number(right));
+ }
+ return [];
+};
+
+const routeAddresses = (route: BrokerRoute): string[] =>
+ Object.values(route.brokerAddrs ?? {}).filter((addr) => addr.trim().length >
0);
+
+const preferredBrokerAddr = (route: BrokerRoute): string => {
+ if (route.brokerAddr) return route.brokerAddr;
+ if (route.masterAddr) return route.masterAddr;
+ return routeAddresses(route)[0] ?? '';
+};
+
+const routeReplicaCount = (route: BrokerRoute, brokerIds: string[]): number =>
{
+ if (typeof route.replicaCount === 'number') return Math.max(0,
route.replicaCount);
+ return brokerIds.filter((id) => id !== '0').length;
+};
+
+const calculateSkew = (values: number[]): RouteQueueSkew => {
+ const activeValues = values.filter((value) => value > 0);
+ if (activeValues.length <= 1) return EMPTY_SKEW;
+
+ const max = Math.max(...activeValues);
+ const min = Math.min(...activeValues);
+ const gap = max - min;
+ if (gap === 0) return EMPTY_SKEW;
+
+ const average = activeValues.reduce((sum, value) => sum + value, 0) /
activeValues.length;
+ return { gap, ratio: average === 0 ? 0 : Math.round((gap / average) * 100) /
100 };
+};
+
+const maxStatus = (issues: RouteDiagnosticIssue[]): RouteDiagnosticStatus =>
+ issues.reduce<RouteDiagnosticStatus>(
+ (status, current) =>
+ STATUS_ORDER[current.severity] > STATUS_ORDER[status] ? current.severity
: status,
+ 'healthy',
+ );
+
+const hasSkew = (skew: RouteQueueSkew): boolean => skew.gap > 0 && skew.ratio
>= 0.5;
+
+const collectAddressDuplicates = (routes: BrokerRoute[]): Set<string> => {
+ const firstBrokerByAddr = new Map<string, string>();
+ const duplicates = new Set<string>();
+
+ routes.forEach((route) => {
+ const brokerName = route.brokerName || 'unknown';
+ routeAddresses(route).forEach((addr) => {
+ const firstBroker = firstBrokerByAddr.get(addr);
+ if (firstBroker && firstBroker !== brokerName) {
+ duplicates.add(addr);
+ } else {
+ firstBrokerByAddr.set(addr, brokerName);
+ }
+ });
+ });
+
+ return duplicates;
+};
+
+const distributionIssues = (
+ route: BrokerRoute,
+ writeSkew: RouteQueueSkew,
+ readSkew: RouteQueueSkew,
+ duplicateAddresses: Set<string>,
+): RouteDiagnosticIssue[] => {
+ const brokerName = route.brokerName || 'unknown';
+ const writeQueues = queueCount(route.writeQueues);
+ const readQueues = queueCount(route.readQueues);
+ const readable = inferReadable(route);
+ const writable = inferWritable(route);
+ const brokerAddr = preferredBrokerAddr(route);
+ const masterAddr = route.masterAddr ?? route.brokerAddrs?.['0'] ?? '';
+ const issues: RouteDiagnosticIssue[] = [];
+
+ if (!brokerAddr) {
+ issues.push(
+ issue(
+ 'MISSING_BROKER_ADDRESS',
+ 'critical',
+ 'Broker 地址缺失',
+ 'NameServer 返回了队列元数据，但没有返回可用于定位 Broker 的地址。',
+ brokerName,
+ ),
+ );
+ }
+
+ if (route.brokerAddrs && Object.keys(route.brokerAddrs).length > 0 &&
!masterAddr) {
+ issues.push(
+ issue(
+ 'MISSING_MASTER_ADDRESS',
+ 'warning',
+ 'Master 地址缺失',
+ '该 Broker 只返回了非 master 地址，Topic 写入链路需要确认 master 是否在线。',
+ brokerName,
+ ),
+ );
+ }
+
+ if (writeQueues === 0) {
+ issues.push(
+ issue(
+ 'WRITE_QUEUE_UNAVAILABLE',
+ 'critical',
+ '写队列不可用',
+ '该 Broker 没有可写队列，生产者不会把消息写到这个 Broker。',
+ brokerName,
+ ),
+ );
+ }
+
+ if (readQueues === 0) {
+ issues.push(
+ issue(
+ 'READ_QUEUE_UNAVAILABLE',
+ 'critical',
+ '读队列不可用',
+ '该 Broker 没有可读队列，消费者不会从这个 Broker 拉取消息。',
+ brokerName,
+ ),
+ );
+ }
+
+ if (!writable) {
+ issues.push(
+ issue(
+ 'PERMISSION_NOT_WRITABLE',
+ 'warning',
+ '权限不允许写入',
+ 'Topic 权限缺少写权限，生产者发送可能失败或被路由到其他 Broker。',
+ brokerName,
+ ),
+ );
+ }
+
+ if (!readable) {
+ issues.push(
+ issue(
+ 'PERMISSION_NOT_READABLE',
+ 'warning',
+ '权限不允许读取',
+ 'Topic 权限缺少读权限，消费者订阅后可能无法正常消费。',
+ brokerName,
+ ),
+ );
+ }
+
+ if (writeQueues !== readQueues) {
+ issues.push(
+ issue(
+ 'READ_WRITE_QUEUE_MISMATCH',
+ 'warning',
+ '读写队列不一致',
+ '该 Broker 的读队列数和写队列数不同，扩缩容或迁移后需要确认配置是否符合预期。',
+ brokerName,
+ ),
+ );
+ }
+
+ if (hasSkew(writeSkew)) {
+ issues.push(
+ issue(
+ 'WRITE_QUEUE_SKEW',
+ 'warning',
+ '写队列分布不均',
+ '不同 Broker 的写队列数差距较大，生产流量可能无法均匀分摊。',
+ brokerName,
+ ),
+ );
+ }
+
+ if (hasSkew(readSkew)) {
+ issues.push(
+ issue(
+ 'READ_QUEUE_SKEW',
+ 'warning',
+ '读队列分布不均',
+ '不同 Broker 的读队列数差距较大，消费者负载可能无法均匀分摊。',
+ brokerName,
+ ),
+ );
+ }
+
+ if (routeAddresses(route).some((addr) => duplicateAddresses.has(addr))) {
+ issues.push(
+ issue(
+ 'DUPLICATE_BROKER_ADDRESS',
+ 'warning',
+ 'Broker 地址重复',
+ '多个 BrokerName 返回了相同地址，请确认 NameServer 注册信息是否过期。',
+ brokerName,
+ ),
+ );
+ }
+
+ return issues;
+};
+
+const buildRecommendations = (issues: RouteDiagnosticIssue[]): string[] => {
+ const actions: string[] = [];
+ const codes = new Set(issues.map((item) => item.code));
+
+ if (codes.has('NO_ROUTE')) {
+ actions.push('确认 Topic 已在目标 Broker 上创建；必要时使用“在 Broker 上重建”。');
+ }
+ if (codes.has('MISSING_BROKER_ADDRESS') ||
codes.has('MISSING_MASTER_ADDRESS')) {
+ actions.push('检查 Broker 是否仍向 NameServer 注册，并确认 master 节点可达。');
+ }
+ if (codes.has('NO_WRITABLE_ROUTE') || codes.has('PERMISSION_NOT_WRITABLE')) {
+ actions.push('确认 Topic 权限包含写权限，避免生产者发送失败。');
+ }
+ if (codes.has('NO_READABLE_ROUTE') || codes.has('PERMISSION_NOT_READABLE')) {
+ actions.push('确认 Topic 权限包含读权限，避免消费者订阅后无可读队列。');
+ }
+ if (
+ codes.has('WRITE_QUEUE_UNAVAILABLE') ||
+ codes.has('READ_QUEUE_UNAVAILABLE') ||
+ codes.has('READ_WRITE_QUEUE_MISMATCH')
+ ) {
+ actions.push('对比各 Broker 上的 TopicConfig，统一读写队列数后再观察客户端路由。');
+ }
+ if (codes.has('WRITE_QUEUE_SKEW') || codes.has('READ_QUEUE_SKEW')) {
+ actions.push('评估是否需要扩容、迁移或重新分配队列，降低单 Broker 负载集中风险。');
+ }
+ if (codes.has('SINGLE_BROKER_ROUTE')) {
+ actions.push('确认该 Topic 是否预期只部署在单 Broker；生产业务建议准备冗余路由。');
+ }
+ if (codes.has('DUPLICATE_BROKER_ADDRESS')) {
+ actions.push('清理过期 Broker 注册信息，避免客户端拿到重复或错误地址。');
+ }
+
+ return actions;
+};
+
+export const analyzeTopicRoutes = (routes: BrokerRoute[]):
TopicRouteDiagnostics => {
+ if (routes.length === 0) {
+ const issues = [
+ issue(
+ 'NO_ROUTE',
+ 'critical',
+ 'Broker 上没有 Topic 路由',
+ '元数据中存在 Topic 记录，但当前实例没有返回任何 Broker 路由。',
+ ),
+ ];
+ return {
+ status: 'critical',
+ statusText: STATUS_TEXT.critical,
+ statusColor: STATUS_COLOR.critical,
+ summary: {
+ brokerCount: 0,
+ addressCount: 0,
+ replicaCount: 0,
+ writableBrokerCount: 0,
+ readableBrokerCount: 0,
+ totalWriteQueues: 0,
+ totalReadQueues: 0,
+ writeSkew: EMPTY_SKEW,
+ readSkew: EMPTY_SKEW,
+ },
+ distributions: [],
+ issues,
+ recommendations: buildRecommendations(issues),
+ };
+ }
+
+ const writeCounts = routes.map((route) => queueCount(route.writeQueues));
+ const readCounts = routes.map((route) => queueCount(route.readQueues));
+ const totalWriteQueues = writeCounts.reduce((sum, value) => sum + value, 0);
+ const totalReadQueues = readCounts.reduce((sum, value) => sum + value, 0);
+ const writeSkew = calculateSkew(writeCounts);
+ const readSkew = calculateSkew(readCounts);
+ const duplicateAddresses = collectAddressDuplicates(routes);
+
+ const distributions = routes.map<RouteDistribution>((route, index) => {
+ const brokerIds = routeBrokerIds(route);
+ const routeIssues = distributionIssues(route, writeSkew, readSkew,
duplicateAddresses);
+
+ return {
+ key: `${route.brokerName || 'broker'}-${index}`,
+ brokerName: route.brokerName || '-',
+ brokerAddr: preferredBrokerAddr(route) || '-',
+ masterAddr: route.masterAddr ?? route.brokerAddrs?.['0'] ?? '',
+ brokerIds,
+ replicaCount: routeReplicaCount(route, brokerIds),
+ writeQueues: queueCount(route.writeQueues),
+ readQueues: queueCount(route.readQueues),
+ writeShare: queueShare(queueCount(route.writeQueues), totalWriteQueues),
+ readShare: queueShare(queueCount(route.readQueues), totalReadQueues),
+ perm: route.perm,
+ readable: inferReadable(route),
+ writable: inferWritable(route),
+ topicSysFlag: route.topicSysFlag,
+ status: maxStatus(routeIssues),
+ issues: routeIssues,
+ };
+ });
+
+ const issues = distributions.flatMap((distribution) => distribution.issues);
+ const writableBrokerCount = distributions.filter(
+ (distribution) => distribution.writable && distribution.writeQueues > 0,
+ ).length;
+ const readableBrokerCount = distributions.filter(
+ (distribution) => distribution.readable && distribution.readQueues > 0,
+ ).length;
+
+ if (writableBrokerCount === 0) {
+ issues.push(
+ issue(
+ 'NO_WRITABLE_ROUTE',
+ 'critical',
+ '没有可写路由',
+ '所有 Broker 都缺少写权限或写队列，生产者无法向该 Topic 发送消息。',
+ ),
+ );
+ }
+
+ if (readableBrokerCount === 0) {
+ issues.push(
+ issue(
+ 'NO_READABLE_ROUTE',
+ 'critical',
+ '没有可读路由',
+ '所有 Broker 都缺少读权限或读队列，消费者无法从该 Topic 拉取消息。',
+ ),
+ );
+ }
+
+ if (routes.length === 1) {
+ issues.push(
+ issue(
+ 'SINGLE_BROKER_ROUTE',
+ 'warning',
+ '单 Broker 路由',
+ '该 Topic 只返回一个 Broker 路由，生产业务需要确认是否符合容灾预期。',
+ ),
+ );
+ }
+
+ const status = maxStatus(issues);
+ const addressCount = new Set(routes.flatMap((route) =>
routeAddresses(route))).size;
+
+ return {
+ status,
+ statusText: STATUS_TEXT[status],
+ statusColor: STATUS_COLOR[status],
+ summary: {
+ brokerCount: routes.length,
+ addressCount,
+ replicaCount: distributions.reduce((sum, item) => sum +
item.replicaCount, 0),
+ writableBrokerCount,
+ readableBrokerCount,
+ totalWriteQueues,
+ totalReadQueues,
+ writeSkew,
+ readSkew,
+ },
+ distributions,
+ issues,
+ recommendations: buildRecommendations(issues),
+ };
+};
