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 09def4364 fix(consumer): handle large queue health inventories (#7245)
09def4364 is described below
commit 09def4364965c47c7d4cfb0231a7e89dec579c82
Author: Loyal-Young <[email protected]>
AuthorDate: Mon Oct 26 21:23:08 2026 +0800
fix(consumer): handle large queue health inventories (#7245)
* fix(consumer): handle large queue health inventories
---
web/src/utils/consumerGroupDiagnostics.test.ts | 17 +++++++++++++++++
web/src/utils/consumerGroupDiagnostics.ts | 24 +++++++++++++++++-------
2 files changed, 34 insertions(+), 7 deletions(-)
diff --git a/web/src/utils/consumerGroupDiagnostics.test.ts
b/web/src/utils/consumerGroupDiagnostics.test.ts
index ad491615d..3eaa205fe 100644
--- a/web/src/utils/consumerGroupDiagnostics.test.ts
+++ b/web/src/utils/consumerGroupDiagnostics.test.ts
@@ -89,6 +89,23 @@ describe('consumer group diagnostics', () => {
expect(diagnostics.issues).toEqual([]);
});
+ it('summarizes large queue inventories without spreading them into function
arguments', () => {
+ const progress = Array<QueueProgress>(150_000).fill(queue({ diffTotal: 1
}));
+ progress[0] = queue({ diffTotal: 5 });
+
+ const diagnostics = analyzeConsumerGroupHealth(
+ group({ totalLag: 150_004 }),
+ [subscription()],
+ progress,
+ { now: '2026-08-31T12:01:00Z' },
+ );
+
+ expect(diagnostics.summary.queueCount).toBe(150_000);
+ expect(diagnostics.summary.totalKnownLag).toBe(150_004);
+ expect(diagnostics.summary.maxQueueLag).toBe(5);
+ expect(diagnostics.issues.map((item) =>
item.code)).toContain('QUEUE_LAG_SKEW');
+ });
+
it('does not treat unavailable connection data as zero online clients', ()
=> {
const diagnostics = analyzeConsumerGroupHealth(
group({ onlineInstances: -1, totalLag: 2400, instances: [] }),
diff --git a/web/src/utils/consumerGroupDiagnostics.ts
b/web/src/utils/consumerGroupDiagnostics.ts
index bb0848ecf..c7a3f6883 100644
--- a/web/src/utils/consumerGroupDiagnostics.ts
+++ b/web/src/utils/consumerGroupDiagnostics.ts
@@ -158,11 +158,17 @@ const topicCount = (group: ConsumerGroup, subscriptions:
SubscriptionEntry[]): n
};
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;
+ let positiveCount = 0;
+ let minPositive = Number.POSITIVE_INFINITY;
+ let maxPositive = 0;
+ for (const lag of knownQueueLags) {
+ if (lag <= 0) continue;
+ positiveCount += 1;
+ minPositive = Math.min(minPositive, lag);
+ maxPositive = Math.max(maxPositive, lag);
+ }
+ if (positiveCount <= 1) return 0;
+ return Math.round((maxPositive / minPositive) * 100) / 100;
};
const maxStatus = (issues: ConsumerGroupHealthIssue[]):
ConsumerGroupHealthStatus =>
@@ -427,8 +433,12 @@ export const analyzeConsumerGroupHealth = (
unknownQueueCount,
totalKnownLag,
reportedLag: summaryReportedLag,
- maxQueueLag: knownQueueLags.length > 0 ? Math.max(...knownQueueLags) :
null,
- maxHeartbeatAgeSeconds: heartbeatAges.length > 0 ?
Math.max(...heartbeatAges) : null,
+ maxQueueLag:
+ knownQueueLags.length > 0
+ ? knownQueueLags.reduce((max, lag) => Math.max(max, lag), 0)
+ : null,
+ maxHeartbeatAgeSeconds:
+ heartbeatAges.length > 0 ? heartbeatAges.reduce((max, age) =>
Math.max(max, age), 0) : null,
staleClientCount: issues.filter((item) => item.code ===
'STALE_HEARTBEAT').length,
},
issues,
