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 af4c6a59e feat(consumer): add Consumer Group health diagnostics (#2881)
af4c6a59e is described below
commit af4c6a59e2eee8f57ac33edac9bf89be7c941c23
Author: coder999o <[email protected]>
AuthorDate: Wed Sep 2 16:29:12 2026 +0800
feat(consumer): add Consumer Group health diagnostics (#2881)
---
.../pages/instance/__tests__/ConsumerPage.test.tsx | 67 ++++
web/src/pages/instance/consumer.tsx | 223 ++++++++++-
web/src/utils/consumerGroupDiagnostics.test.ts | 157 ++++++++
web/src/utils/consumerGroupDiagnostics.ts | 423 +++++++++++++++++++++
4 files changed, 866 insertions(+), 4 deletions(-)
diff --git a/web/src/pages/instance/__tests__/ConsumerPage.test.tsx
b/web/src/pages/instance/__tests__/ConsumerPage.test.tsx
index 12186cc05..761ea2858 100644
--- a/web/src/pages/instance/__tests__/ConsumerPage.test.tsx
+++ b/web/src/pages/instance/__tests__/ConsumerPage.test.tsx
@@ -482,6 +482,73 @@ describe('Consumer page', () => {
await waitFor(() =>
expect(screen.getAllByText('remote-topic').length).toBeGreaterThan(0));
});
+ it('shows group health diagnostics from subscriptions, progress and
clients', async () => {
+ const riskyGroup: ConsumerGroup = {
+ ...group,
+ totalLag: 1_200,
+ delaySeconds: 720,
+ onlineInstances: 2,
+ instances: [
+ {
+ clientId: '[email protected]',
+ protocol: 'GRPC',
+ address: '10.0.0.1:49152',
+ subscribedTopics: ['remote-topic'],
+ lastHeartbeat: '2026-07-23T00:00:00Z',
+ topicLag: { 'remote-topic': 10 },
+ },
+ {
+ clientId: '[email protected]',
+ protocol: 'REMOTING',
+ address: '10.0.0.2:49152',
+ subscribedTopics: [],
+ lastHeartbeat: '2026-07-23T00:00:00Z',
+ topicLag: {},
+ },
+ ],
+ };
+
vi.mocked(consumerService.listConsumerGroupPage).mockResolvedValue(groupPage([riskyGroup]));
+ vi.mocked(consumerService.getConsumerSubscriptions).mockResolvedValue([
+ {
+ topic: 'remote-topic',
+ expression: 'tagA',
+ type: 'NORMAL',
+ filterMode: 'Tag 过滤',
+ consistency: '不一致',
+ },
+ ]);
+ vi.mocked(consumerService.getConsumerProgress).mockResolvedValue([
+ {
+ topic: 'remote-topic',
+ broker: 'broker-a',
+ queueId: 0,
+ brokerOffset: 100,
+ consumerOffset: 90,
+ diffTotal: 10,
+ },
+ {
+ topic: 'remote-topic',
+ broker: 'broker-b',
+ queueId: 1,
+ brokerOffset: 1_200,
+ consumerOffset: 100,
+ diffTotal: 1_100,
+ },
+ ]);
+ const user = userEvent.setup();
+ renderWithProviders(<ConsumerPage />);
+
+ await user.click(await screen.findByRole('button', { name: /详情/ }));
+ await user.click(await screen.findByRole('tab', { name: /健康诊断/ }));
+ const panel = await screen.findByRole('tabpanel', { name: /健康诊断/ });
+
+ await waitFor(() =>
expect(within(panel).getAllByText('消费风险').length).toBeGreaterThan(0));
+ expect(within(panel).getByText('订阅表达式不一致')).toBeInTheDocument();
+ expect(within(panel).getByText('Queue 堆积分布严重倾斜')).toBeInTheDocument();
+ expect(within(panel).getAllByText('客户端心跳过期').length).toBeGreaterThan(0);
+ expect(within(panel).getByText('处理建议')).toBeInTheDocument();
+ });
+
it('filters queue progress to the topic of the clicked distribution button',
async () => {
vi.mocked(consumerService.getConsumerSubscriptions).mockResolvedValue([
{
diff --git a/web/src/pages/instance/consumer.tsx
b/web/src/pages/instance/consumer.tsx
index e286603ce..7ddf7da24 100644
--- a/web/src/pages/instance/consumer.tsx
+++ b/web/src/pages/instance/consumer.tsx
@@ -41,6 +41,7 @@ import {
DatePicker,
Tooltip,
Spin,
+ Progress,
message,
} from 'antd';
import {
@@ -103,6 +104,11 @@ import {
import { downloadCsv } from '../../utils/download';
import { formatLag, isLagAvailable, lagSortValue } from
'../../utils/consumerLag';
import { tableScrollX } from '../../utils/table';
+import {
+ analyzeConsumerGroupHealth,
+ type ConsumerGroupHealthIssue,
+ type ConsumerGroupHealthStatus,
+} from '../../utils/consumerGroupDiagnostics';
const { Text } = Typography;
@@ -188,6 +194,24 @@ const resetPreviewRiskLabel = (riskLevel: string) => {
return '正常';
};
+const healthStatusTagColor = (status: ConsumerGroupHealthStatus) => {
+ if (status === 'critical') return 'red';
+ if (status === 'warning') return 'orange';
+ return 'green';
+};
+
+const issueSeverityTagColor = (severity: ConsumerGroupHealthIssue['severity'])
=> {
+ if (severity === 'critical') return 'red';
+ if (severity === 'warning') return 'orange';
+ return 'blue';
+};
+
+const issueSeverityLabel = (severity: ConsumerGroupHealthIssue['severity']) =>
{
+ if (severity === 'critical') return '风险';
+ if (severity === 'warning') return '关注';
+ return '提示';
+};
+
const resetPreviewQueueMessage = (queue: ResetConsumerOffsetQueuePreview) => {
const messages: string[] = [];
if (queue.riskLevel === 'ERROR') {
@@ -477,6 +501,10 @@ const ConsumerPageContent = ({
if (key === 'settings' && selectedGroup && settingsGroup?.name !==
selectedGroup.name) {
void loadGroupSettings(selectedGroup);
}
+ if (key === 'health' && selectedGroup) {
+ void loadSubscriptions(selectedGroup.name);
+ void loadProgress(selectedGroup.name);
+ }
};
const saveSettings = async () => {
@@ -533,9 +561,10 @@ const ConsumerPageContent = ({
}
return Array.from(topics).map((topic) => ({ label: topic, value: topic }));
}, [resetDiagnosticKey, resetGroup, subscriptionsByGroup]);
- const selectedSubscriptions = selectedGroup
- ? (subscriptionsByGroup[selectedDiagnosticKey] ?? [])
- : [];
+ const selectedSubscriptions = useMemo(
+ () => (selectedGroupName ? (subscriptionsByGroup[selectedDiagnosticKey] ??
[]) : []),
+ [selectedDiagnosticKey, selectedGroupName, subscriptionsByGroup],
+ );
const inconsistentSubscriptions =
selectedSubscriptions.filter(isInconsistentSubscription);
const unknownSubscriptions = selectedSubscriptions.filter(
(subscription) =>
@@ -570,6 +599,13 @@ const ConsumerPageContent = ({
(sum, q) => sum + (isLagAvailable(q.diffTotal) ? q.diffTotal : 0),
0,
);
+ const selectedGroupHealth = useMemo(
+ () =>
+ selectedGroup
+ ? analyzeConsumerGroupHealth(selectedGroup, selectedSubscriptions,
selectedProgress)
+ : null,
+ [selectedGroup, selectedProgress, selectedSubscriptions],
+ );
const handlePreviewResetOffset = async () => {
if (!resetGroup || !resetTopic) {
@@ -1095,6 +1131,36 @@ const ConsumerPageContent = ({
},
];
+ const healthIssueColumns: ColumnsType<ConsumerGroupHealthIssue> = [
+ {
+ title: '级别',
+ dataIndex: 'severity',
+ key: 'severity',
+ width: 84,
+ render: (severity: ConsumerGroupHealthIssue['severity']) => (
+ <Tag
color={issueSeverityTagColor(severity)}>{issueSeverityLabel(severity)}</Tag>
+ ),
+ },
+ {
+ title: '诊断项',
+ dataIndex: 'title',
+ key: 'title',
+ width: 180,
+ render: (title: string, record) => (
+ <Space direction="vertical" size={0}>
+ <Text strong>{title}</Text>
+ {record.subject && <Text type="secondary">{record.subject}</Text>}
+ </Space>
+ ),
+ },
+ {
+ title: '说明',
+ dataIndex: 'description',
+ key: 'description',
+ render: (description: string) => <Text>{description}</Text>,
+ },
+ ];
+
/* ═══════════════════════════════════════════
Modal: Queue Progress Tab
═══════════════════════════════════════════ */
@@ -1487,7 +1553,7 @@ const ConsumerPageContent = ({
setSettingsLoading(false);
settingsForm.resetFields();
}}
- width={detailTab === 'progress' ? 1080 : 800}
+ width={detailTab === 'progress' || detailTab === 'health' ? 1080 : 800}
destroyOnHidden
footer={null}
>
@@ -1713,6 +1779,155 @@ const ConsumerPageContent = ({
</div>
),
},
+ /* ─── 健康诊断 Tab ─── */
+ {
+ key: 'health',
+ label: (
+ <Space size={4}>
+ <Info size={14} />
+ <span>健康诊断</span>
+ </Space>
+ ),
+ children: selectedGroupHealth && (
+ <Space direction="vertical" size={16} style={{ width: '100%'
}}>
+ <Flex justify="space-between" align="center" gap={12} wrap>
+ <Space direction="vertical" size={2}>
+ <Space>
+ <Tag
color={healthStatusTagColor(selectedGroupHealth.status)}>
+ {selectedGroupHealth.statusText}
+ </Tag>
+ <Text type="secondary">
+ 汇总订阅、队列进度和在线客户端，定位消费风险。
+ </Text>
+ </Space>
+ <Text type="secondary">诊断结果随详情弹窗每 2 秒自动刷新。</Text>
+ </Space>
+ <Button
+ size="small"
+ icon={<ArrowsClockwise size={14} />}
+
loading={subscriptionLoadingByGroup[selectedDiagnosticKey]}
+ onClick={() => {
+ void loadSubscriptions(selectedGroup.name, true);
+ void loadProgress(selectedGroup.name, true);
+ }}
+ >
+ 重新诊断
+ </Button>
+ </Flex>
+
+ {subscriptionErrorByGroup[selectedDiagnosticKey] && (
+ <Alert
+ type="warning"
+ showIcon
+ message="订阅一致性检查失败，诊断仍使用当前可用的进度和客户端数据。"
+ />
+ )}
+
+ <Row gutter={16}>
+ <Col span={6}>
+ <Card size="small" style={{ borderRadius: 8 }}>
+ <Statistic
+ title="健康分"
+ value={selectedGroupHealth.summary.healthScore}
+ suffix="/ 100"
+ valueStyle={{
+ color:
+ selectedGroupHealth.status === 'critical'
+ ? '#ff4d4f'
+ : selectedGroupHealth.status === 'warning'
+ ? '#faad14'
+ : '#52c41a',
+ }}
+ />
+ <Progress
+ percent={selectedGroupHealth.summary.healthScore}
+ showInfo={false}
+ status={
+ selectedGroupHealth.status === 'critical'
+ ? 'exception'
+ : selectedGroupHealth.status === 'warning'
+ ? 'active'
+ : 'success'
+ }
+ />
+ </Card>
+ </Col>
+ <Col span={6}>
+ <Card size="small" style={{ borderRadius: 8 }}>
+ <Statistic
+ title="已知堆积"
+ value={selectedGroupHealth.summary.totalKnownLag}
+ valueStyle={{
+ color:
lagColor(selectedGroupHealth.summary.totalKnownLag),
+ }}
+ />
+ <Text type="secondary">
+ 报告堆积：
+ {selectedGroupHealth.summary.reportedLag === null
+ ? UNAVAILABLE_LAG_LABEL
+ :
selectedGroupHealth.summary.reportedLag.toLocaleString()}
+ </Text>
+ </Card>
+ </Col>
+ <Col span={6}>
+ <Card size="small" style={{ borderRadius: 8 }}>
+ <Statistic
+ title="Queue 覆盖"
+ value={selectedGroupHealth.summary.queueCount}
+
suffix={`/${selectedGroupHealth.summary.subscribedTopicCount} Topic`}
+ />
+ <Text type="secondary">
+ {selectedGroupHealth.summary.unknownQueueCount > 0
+ ?
`${selectedGroupHealth.summary.unknownQueueCount} 个 Queue 堆积不可用`
+ : 'Queue 堆积均可计算'}
+ </Text>
+ </Card>
+ </Col>
+ <Col span={6}>
+ <Card size="small" style={{ borderRadius: 8 }}>
+ <Statistic
+ title="客户端"
+ value={selectedGroupHealth.summary.onlineInstances}
+ />
+ <Text type="secondary">
+ {selectedGroupHealth.summary.staleClientCount > 0
+ ?
`${selectedGroupHealth.summary.staleClientCount} 个心跳过期`
+ : '心跳状态正常'}
+ </Text>
+ </Card>
+ </Col>
+ </Row>
+
+ {selectedGroupHealth.issues.length > 0 ? (
+ <Table
+ columns={healthIssueColumns}
+ dataSource={selectedGroupHealth.issues}
+ rowKey="id"
+ pagination={false}
+ size="small"
+ scroll={{ x: tableScrollX(healthIssueColumns) }}
+ />
+ ) : (
+ <Alert type="success" showIcon message="未发现消费组健康风险" />
+ )}
+
+ {selectedGroupHealth.recommendations.length > 0 && (
+ <Alert
+ type="info"
+ showIcon
+ message="处理建议"
+ description={
+ <Space direction="vertical" size={4}>
+
{selectedGroupHealth.recommendations.map((recommendation) => (
+ <Text
key={recommendation}>{recommendation}</Text>
+ ))}
+ </Space>
+ }
+ />
+ )}
+ </Space>
+ ),
+ },
/* ─── 消费进度 Tab ─── */
{
key: 'progress',
diff --git a/web/src/utils/consumerGroupDiagnostics.test.ts
b/web/src/utils/consumerGroupDiagnostics.test.ts
new file mode 100644
index 000000000..b6a09bfcc
--- /dev/null
+++ b/web/src/utils/consumerGroupDiagnostics.test.ts
@@ -0,0 +1,157 @@
+/*
+ * 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 { ConsumerGroup, QueueProgress, SubscriptionEntry } from
'../api/metadata';
+import { analyzeConsumerGroupHealth } from './consumerGroupDiagnostics';
+
+const group = (overrides: Partial<ConsumerGroup> = {}): ConsumerGroup => ({
+ name: 'orders-cg',
+ namespace: 'default',
+ clusterId: 'cluster-a',
+ instanceId: 'instance-a',
+ subscriptionMode: 'Push',
+ consumeType: 'CLUSTERING',
+ onlineInstances: 1,
+ totalLag: 12,
+ subscribedTopics: ['orders'],
+ subscriptionDataType: 'NORMAL',
+ retryMaxTimes: 16,
+ gmtCreate: '2026-08-31T10:00:00Z',
+ gmtModified: '2026-08-31T10:00:00Z',
+ delaySeconds: 10,
+ instances: [
+ {
+ clientId: '[email protected]',
+ protocol: 'GRPC',
+ address: '10.0.0.1:49152',
+ subscribedTopics: ['orders'],
+ lastHeartbeat: '2026-08-31T12:00:00Z',
+ topicLag: { orders: 12 },
+ },
+ ],
+ ...overrides,
+});
+
+const subscription = (overrides: Partial<SubscriptionEntry> = {}):
SubscriptionEntry => ({
+ topic: 'orders',
+ expression: '*',
+ type: 'NORMAL',
+ filterMode: '全量',
+ consistency: '一致',
+ ...overrides,
+});
+
+const queue = (overrides: Partial<QueueProgress> = {}): QueueProgress => ({
+ topic: 'orders',
+ broker: 'broker-a',
+ queueId: 0,
+ brokerOffset: 120,
+ consumerOffset: 114,
+ diffTotal: 6,
+ ...overrides,
+});
+
+describe('consumer group diagnostics', () => {
+ it('summarizes an active group with balanced queues as healthy', () => {
+ const diagnostics = analyzeConsumerGroupHealth(
+ group(),
+ [subscription()],
+ [queue(), queue({ broker: 'broker-b', queueId: 1 })],
+ { now: '2026-08-31T12:01:00Z' },
+ );
+
+ expect(diagnostics.status).toBe('healthy');
+ expect(diagnostics.summary).toMatchObject({
+ healthScore: 100,
+ onlineInstances: 1,
+ subscribedTopicCount: 1,
+ queueCount: 2,
+ lagQueueCount: 2,
+ totalKnownLag: 12,
+ reportedLag: 12,
+ staleClientCount: 0,
+ });
+ expect(diagnostics.issues).toEqual([]);
+ });
+
+ it('flags critical subscription, queue and runtime risks', () => {
+ const diagnostics = analyzeConsumerGroupHealth(
+ group({
+ onlineInstances: 0,
+ totalLag: 2_400,
+ delaySeconds: 1_900,
+ instances: [],
+ }),
+ [subscription({ consistency: '不一致', expression: 'tagA' })],
+ [queue({ diffTotal: 10 }), queue({ broker: 'broker-b', queueId: 1,
diffTotal: 1_100 })],
+ { now: '2026-08-31T12:01:00Z' },
+ );
+
+ expect(diagnostics.status).toBe('critical');
+ expect(diagnostics.summary.healthScore).toBeLessThan(50);
+ expect(diagnostics.issues.map((item) => item.code)).toEqual(
+ expect.arrayContaining([
+ 'SUBSCRIPTION_INCONSISTENT',
+ 'QUEUE_LAG_SKEW',
+ 'HIGH_GROUP_LAG',
+ 'NO_ACTIVE_CLIENTS_WITH_LAG',
+ 'HIGH_CONSUME_DELAY',
+ ]),
+ );
+ expect(diagnostics.recommendations).toEqual(
+ expect.arrayContaining([
+ '先确认消费者进程、Proxy/Broker 网络连通性和客户端心跳是否恢复。',
+ '检查热点 Queue 的分配、消费者线程池和单分区顺序消费阻塞情况。',
+ ]),
+ );
+ });
+
+ it('keeps warnings for unknown lag, unknown subscriptions and stale
clients', () => {
+ const diagnostics = analyzeConsumerGroupHealth(
+ group({
+ totalLag: -1,
+ delaySeconds: 360,
+ instances: [
+ {
+ clientId: '[email protected]',
+ protocol: 'GRPC',
+ address: '10.0.0.1:49152',
+ subscribedTopics: ['orders'],
+ lastHeartbeat: '2026-08-31T11:50:00Z',
+ topicLag: {},
+ },
+ ],
+ }),
+ [subscription({ consistency: 'unknown' })],
+ [queue({ diffTotal: -1 })],
+ { now: '2026-08-31T12:01:00Z' },
+ );
+
+ expect(diagnostics.status).toBe('warning');
+ expect(diagnostics.summary).toMatchObject({
+ reportedLag: 0,
+ unknownQueueCount: 1,
+ maxQueueLag: null,
+ maxHeartbeatAgeSeconds: 660,
+ staleClientCount: 1,
+ });
+ expect(diagnostics.issues.map((item) => item.code)).toEqual(
+ expect.arrayContaining(['SUBSCRIPTION_UNKNOWN', 'UNKNOWN_QUEUE_LAG',
'STALE_HEARTBEAT']),
+ );
+ });
+});
diff --git a/web/src/utils/consumerGroupDiagnostics.ts
b/web/src/utils/consumerGroupDiagnostics.ts
new file mode 100644
index 000000000..4e1d8c52a
--- /dev/null
+++ b/web/src/utils/consumerGroupDiagnostics.ts
@@ -0,0 +1,423 @@
+/*
+ * 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 { ConsumerGroup, QueueProgress, SubscriptionEntry } from
'../api/metadata';
+import { isLagAvailable } from './consumerLag';
+
+export type ConsumerGroupHealthStatus = 'healthy' | 'warning' | 'critical';
+
+export type ConsumerGroupHealthIssueCode =
+ | 'NO_ACTIVE_CLIENTS_WITH_LAG'
+ | 'NO_SUBSCRIPTION_DATA'
+ | 'SUBSCRIPTION_INCONSISTENT'
+ | 'SUBSCRIPTION_UNKNOWN'
+ | 'QUEUE_LAG_SKEW'
+ | 'HIGH_GROUP_LAG'
+ | 'HIGH_CONSUME_DELAY'
+ | 'UNKNOWN_QUEUE_LAG'
+ | 'STALE_HEARTBEAT';
+
+export interface ConsumerGroupHealthIssue {
+ id: string;
+ code: ConsumerGroupHealthIssueCode;
+ severity: Exclude<ConsumerGroupHealthStatus, 'healthy'>;
+ title: string;
+ description: string;
+ subject?: string;
+}
+
+export interface ConsumerGroupHealthSummary {
+ healthScore: number;
+ onlineInstances: number;
+ subscribedTopicCount: number;
+ queueCount: number;
+ lagQueueCount: number;
+ unknownQueueCount: number;
+ totalKnownLag: number;
+ reportedLag: number | null;
+ maxQueueLag: number | null;
+ maxHeartbeatAgeSeconds: number | null;
+ staleClientCount: number;
+}
+
+export interface ConsumerGroupHealthDiagnostics {
+ status: ConsumerGroupHealthStatus;
+ statusText: string;
+ statusColor: 'success' | 'warning' | 'error';
+ summary: ConsumerGroupHealthSummary;
+ issues: ConsumerGroupHealthIssue[];
+ recommendations: string[];
+}
+
+export interface ConsumerGroupHealthOptions {
+ now?: Date | string | number;
+ staleHeartbeatSeconds?: number;
+ highLagThreshold?: number;
+ criticalLagThreshold?: number;
+ highDelaySeconds?: number;
+ criticalDelaySeconds?: number;
+ skewWarningRatio?: number;
+ skewCriticalRatio?: number;
+}
+
+const STATUS_ORDER: Record<ConsumerGroupHealthStatus, number> = {
+ healthy: 0,
+ warning: 1,
+ critical: 2,
+};
+
+const STATUS_TEXT: Record<ConsumerGroupHealthStatus, string> = {
+ healthy: '消费组健康',
+ warning: '需要关注',
+ critical: '消费风险',
+};
+
+const STATUS_COLOR: Record<ConsumerGroupHealthStatus, 'success' | 'warning' |
'error'> = {
+ healthy: 'success',
+ warning: 'warning',
+ critical: 'error',
+};
+
+const DEFAULT_STALE_HEARTBEAT_SECONDS = 300;
+const DEFAULT_HIGH_LAG_THRESHOLD = 1_000;
+const DEFAULT_CRITICAL_LAG_THRESHOLD = 10_000;
+const DEFAULT_HIGH_DELAY_SECONDS = 300;
+const DEFAULT_CRITICAL_DELAY_SECONDS = 1_800;
+const DEFAULT_SKEW_WARNING_RATIO = 2;
+const DEFAULT_SKEW_CRITICAL_RATIO = 5;
+
+const issue = (
+ code: ConsumerGroupHealthIssueCode,
+ severity: Exclude<ConsumerGroupHealthStatus, 'healthy'>,
+ title: string,
+ description: string,
+ subject?: string,
+): ConsumerGroupHealthIssue => ({
+ id: [code, subject].filter(Boolean).join(':'),
+ code,
+ severity,
+ title,
+ description,
+ subject,
+});
+
+const normalizeText = (value?: string | null): string => value?.trim() ?? '';
+
+const normalizeKey = (value?: string | null): string =>
normalizeText(value).toLowerCase();
+
+const isConsistentSubscription = (subscription: SubscriptionEntry): boolean =>
+ ['consistent', '一致'].includes(normalizeKey(subscription.consistency));
+
+const isInconsistentSubscription = (subscription: SubscriptionEntry): boolean
=>
+ ['inconsistent', '不一致'].includes(normalizeKey(subscription.consistency));
+
+const parseTimestamp = (value: Date | string | number | undefined): number |
null => {
+ if (value === undefined || value === null || value === '') return null;
+ const timestamp = value instanceof Date ? value.getTime() : new
Date(value).getTime();
+ return Number.isFinite(timestamp) ? timestamp : null;
+};
+
+const heartbeatAgeSeconds = (lastHeartbeat: string | undefined, now: number):
number | null => {
+ const timestamp = parseTimestamp(lastHeartbeat);
+ if (timestamp === null) return null;
+ return Math.max(0, Math.floor((now - timestamp) / 1000));
+};
+
+const knownLag = (progress: QueueProgress[]): number =>
+ progress.reduce((sum, queue) => sum + (isLagAvailable(queue.diffTotal) ?
queue.diffTotal : 0), 0);
+
+const reportedLag = (group: ConsumerGroup, fallback: number): number | null =>
+ isLagAvailable(group.totalLag) ? group.totalLag : fallback;
+
+const topicCount = (group: ConsumerGroup, subscriptions: SubscriptionEntry[]):
number => {
+ const topics = new Set<string>();
+ for (const topic of group.subscribedTopics ?? []) {
+ const normalized = normalizeText(topic);
+ if (normalized) topics.add(normalized);
+ }
+ for (const subscription of subscriptions) {
+ const normalized = normalizeText(subscription.topic);
+ if (normalized) topics.add(normalized);
+ }
+ return topics.size;
+};
+
+const lagSkewRatio = (knownQueueLags: number[]): number => {
+ const positive = knownQueueLags.filter((lag) => lag > 0);
+ if (positive.length <= 1) return 0;
+ const min = Math.min(...positive);
+ if (min === 0) return 0;
+ return Math.round((Math.max(...positive) / min) * 100) / 100;
+};
+
+const maxStatus = (issues: ConsumerGroupHealthIssue[]):
ConsumerGroupHealthStatus =>
+ issues.reduce<ConsumerGroupHealthStatus>(
+ (status, current) =>
+ STATUS_ORDER[current.severity] > STATUS_ORDER[status] ? current.severity
: status,
+ 'healthy',
+ );
+
+const healthScore = (issues: ConsumerGroupHealthIssue[]): number => {
+ const penalties = issues.reduce(
+ (sum, current) => sum + (current.severity === 'critical' ? 25 : 12),
+ 0,
+ );
+ return Math.max(0, Math.min(100, Math.round(100 - penalties)));
+};
+
+const subscriptionIssues = (subscriptions: SubscriptionEntry[]):
ConsumerGroupHealthIssue[] => {
+ if (subscriptions.length === 0) {
+ return [
+ issue(
+ 'NO_SUBSCRIPTION_DATA',
+ 'warning',
+ '暂无订阅明细',
+ '无法从当前结果判断客户端订阅表达式是否一致。',
+ ),
+ ];
+ }
+
+ const issues: ConsumerGroupHealthIssue[] = [];
+ for (const subscription of subscriptions) {
+ if (isInconsistentSubscription(subscription)) {
+ issues.push(
+ issue(
+ 'SUBSCRIPTION_INCONSISTENT',
+ 'critical',
+ '订阅表达式不一致',
+ `${subscription.topic} 的订阅表达式在客户端之间不一致，可能导致消息遗漏或重复消费。`,
+ subscription.topic,
+ ),
+ );
+ } else if (!isConsistentSubscription(subscription)) {
+ issues.push(
+ issue(
+ 'SUBSCRIPTION_UNKNOWN',
+ 'warning',
+ '订阅一致性未知',
+ `${subscription.topic} 的一致性状态未知，建议重新检查客户端订阅。`,
+ subscription.topic,
+ ),
+ );
+ }
+ }
+ return issues;
+};
+
+const progressIssues = (
+ progress: QueueProgress[],
+ knownQueueLags: number[],
+ unknownQueueCount: number,
+ options: Required<
+ Pick<
+ ConsumerGroupHealthOptions,
+ 'skewWarningRatio' | 'skewCriticalRatio' | 'highLagThreshold' |
'criticalLagThreshold'
+ >
+ >,
+): ConsumerGroupHealthIssue[] => {
+ if (progress.length === 0) return [];
+
+ const issues: ConsumerGroupHealthIssue[] = [];
+ const skew = lagSkewRatio(knownQueueLags);
+ const totalLag = knownQueueLags.reduce((sum, lag) => sum + lag, 0);
+
+ if (unknownQueueCount > 0) {
+ issues.push(
+ issue(
+ 'UNKNOWN_QUEUE_LAG',
+ 'warning',
+ '部分 Queue 堆积不可用',
+ `${unknownQueueCount} 个 Queue 无法计算堆积，当前总堆积只包含可用数据。`,
+ ),
+ );
+ }
+ if (skew >= options.skewCriticalRatio) {
+ issues.push(
+ issue(
+ 'QUEUE_LAG_SKEW',
+ 'critical',
+ 'Queue 堆积分布严重倾斜',
+ `最大/最小 Queue 堆积约为 ${skew}:1，可能存在单队列热点或消费者分配不均。`,
+ ),
+ );
+ } else if (skew >= options.skewWarningRatio) {
+ issues.push(
+ issue(
+ 'QUEUE_LAG_SKEW',
+ 'warning',
+ 'Queue 堆积分布不均',
+ `最大/最小 Queue 堆积约为 ${skew}:1，建议观察是否持续扩大。`,
+ ),
+ );
+ }
+ if (totalLag >= options.criticalLagThreshold) {
+ issues.push(
+ issue(
+ 'HIGH_GROUP_LAG',
+ 'critical',
+ 'Group 总堆积过高',
+ `Group 当前已知堆积达到 ${totalLag.toLocaleString()} 条。`,
+ ),
+ );
+ } else if (totalLag >= options.highLagThreshold) {
+ issues.push(
+ issue(
+ 'HIGH_GROUP_LAG',
+ 'warning',
+ 'Group 总堆积偏高',
+ `Group 当前已知堆积达到 ${totalLag.toLocaleString()} 条。`,
+ ),
+ );
+ }
+ return issues;
+};
+
+const runtimeIssues = (
+ group: ConsumerGroup,
+ lag: number | null,
+ now: number,
+ options: Required<
+ Pick<
+ ConsumerGroupHealthOptions,
+ 'staleHeartbeatSeconds' | 'highDelaySeconds' | 'criticalDelaySeconds'
+ >
+ >,
+): ConsumerGroupHealthIssue[] => {
+ const issues: ConsumerGroupHealthIssue[] = [];
+ if ((group.onlineInstances ?? 0) === 0 && (lag ?? 0) > 0) {
+ issues.push(
+ issue(
+ 'NO_ACTIVE_CLIENTS_WITH_LAG',
+ 'critical',
+ '有堆积但无在线客户端',
+ '消费组存在未消费消息，但当前没有在线客户端处理这些消息。',
+ ),
+ );
+ }
+
+ for (const client of group.instances ?? []) {
+ const age = heartbeatAgeSeconds(client.lastHeartbeat, now);
+ if (age !== null && age > options.staleHeartbeatSeconds) {
+ issues.push(
+ issue(
+ 'STALE_HEARTBEAT',
+ 'warning',
+ '客户端心跳过期',
+ `${client.clientId} 的最后心跳已超过 ${options.staleHeartbeatSeconds} 秒。`,
+ client.clientId,
+ ),
+ );
+ }
+ }
+
+ const delaySeconds = Number.isFinite(group.delaySeconds) ?
group.delaySeconds : 0;
+ if (delaySeconds >= options.criticalDelaySeconds) {
+ issues.push(
+ issue(
+ 'HIGH_CONSUME_DELAY',
+ 'critical',
+ '消费延迟过高',
+ `Group 当前消费延迟约 ${delaySeconds.toLocaleString()} 秒，业务可能已经感知延迟。`,
+ ),
+ );
+ } else if (delaySeconds >= options.highDelaySeconds) {
+ issues.push(
+ issue(
+ 'HIGH_CONSUME_DELAY',
+ 'warning',
+ '消费延迟偏高',
+ `Group 当前消费延迟约 ${delaySeconds.toLocaleString()} 秒，建议继续观察趋势。`,
+ ),
+ );
+ }
+ return issues;
+};
+
+const recommendations = (issues: ConsumerGroupHealthIssue[]): string[] => {
+ const codes = new Set(issues.map((item) => item.code));
+ const result: string[] = [];
+ if (codes.has('NO_ACTIVE_CLIENTS_WITH_LAG') || codes.has('STALE_HEARTBEAT'))
{
+ result.push('先确认消费者进程、Proxy/Broker 网络连通性和客户端心跳是否恢复。');
+ }
+ if (codes.has('SUBSCRIPTION_INCONSISTENT') ||
codes.has('SUBSCRIPTION_UNKNOWN')) {
+ result.push('统一同一 Group 内所有客户端的订阅表达式，避免灰度期间同时运行不同过滤条件。');
+ }
+ if (codes.has('QUEUE_LAG_SKEW')) {
+ result.push('检查热点 Queue 的分配、消费者线程池和单分区顺序消费阻塞情况。');
+ }
+ if (codes.has('HIGH_GROUP_LAG') || codes.has('HIGH_CONSUME_DELAY')) {
+ result.push('结合消费 TPS、业务耗时和重试堆积判断是否需要扩容消费者或限流生产端。');
+ }
+ if (codes.has('UNKNOWN_QUEUE_LAG')) {
+ result.push('当堆积不可用时，优先确认 Proxy 指标采集和 Broker offset 查询权限。');
+ }
+ return result;
+};
+
+export const analyzeConsumerGroupHealth = (
+ group: ConsumerGroup,
+ subscriptions: SubscriptionEntry[],
+ progress: QueueProgress[],
+ options: ConsumerGroupHealthOptions = {},
+): ConsumerGroupHealthDiagnostics => {
+ const normalizedOptions = {
+ staleHeartbeatSeconds: options.staleHeartbeatSeconds ??
DEFAULT_STALE_HEARTBEAT_SECONDS,
+ highLagThreshold: options.highLagThreshold ?? DEFAULT_HIGH_LAG_THRESHOLD,
+ criticalLagThreshold: options.criticalLagThreshold ??
DEFAULT_CRITICAL_LAG_THRESHOLD,
+ highDelaySeconds: options.highDelaySeconds ?? DEFAULT_HIGH_DELAY_SECONDS,
+ criticalDelaySeconds: options.criticalDelaySeconds ??
DEFAULT_CRITICAL_DELAY_SECONDS,
+ skewWarningRatio: options.skewWarningRatio ?? DEFAULT_SKEW_WARNING_RATIO,
+ skewCriticalRatio: options.skewCriticalRatio ??
DEFAULT_SKEW_CRITICAL_RATIO,
+ };
+ const now = parseTimestamp(options.now ?? Date.now()) ?? Date.now();
+ const knownQueueLags = progress
+ .map((queue) => queue.diffTotal)
+ .filter((lag): lag is number => isLagAvailable(lag));
+ const totalKnownLag = knownLag(progress);
+ const summaryReportedLag = reportedLag(group, totalKnownLag);
+ const unknownQueueCount = progress.length - knownQueueLags.length;
+ const heartbeatAges = (group.instances ?? [])
+ .map((client) => heartbeatAgeSeconds(client.lastHeartbeat, now))
+ .filter((age): age is number => age !== null);
+ const issues = [
+ ...subscriptionIssues(subscriptions),
+ ...progressIssues(progress, knownQueueLags, unknownQueueCount,
normalizedOptions),
+ ...runtimeIssues(group, summaryReportedLag, now, normalizedOptions),
+ ];
+ const status = maxStatus(issues);
+
+ return {
+ status,
+ statusText: STATUS_TEXT[status],
+ statusColor: STATUS_COLOR[status],
+ summary: {
+ healthScore: healthScore(issues),
+ onlineInstances: group.onlineInstances ?? 0,
+ subscribedTopicCount: topicCount(group, subscriptions),
+ queueCount: progress.length,
+ lagQueueCount: knownQueueLags.filter((lag) => lag > 0).length,
+ unknownQueueCount,
+ totalKnownLag,
+ reportedLag: summaryReportedLag,
+ maxQueueLag: knownQueueLags.length > 0 ? Math.max(...knownQueueLags) :
null,
+ maxHeartbeatAgeSeconds: heartbeatAges.length > 0 ?
Math.max(...heartbeatAges) : null,
+ staleClientCount: issues.filter((item) => item.code ===
'STALE_HEARTBEAT').length,
+ },
+ issues,
+ recommendations: recommendations(issues),
+ };
+};
