From dc1f9eb5344bb5db22778adc46d57eea338fc4b8 Mon Sep 17 00:00:00 2001 From: lixy Date: Sun, 4 Jan 2026 16:54:03 +0800 Subject: [PATCH 1/2] add metric for max queued shard --- .../io/confluent/parallelconsumer/metrics/PCMetricsDef.java | 1 + .../io/confluent/parallelconsumer/state/ShardManager.java | 6 +++++- 2 files changed, 6 insertions(+), 1 deletion(-) diff --git a/parallel-consumer-core/src/main/java/io/confluent/parallelconsumer/metrics/PCMetricsDef.java b/parallel-consumer-core/src/main/java/io/confluent/parallelconsumer/metrics/PCMetricsDef.java index f0010063f..6e394c7f1 100644 --- a/parallel-consumer-core/src/main/java/io/confluent/parallelconsumer/metrics/PCMetricsDef.java +++ b/parallel-consumer-core/src/main/java/io/confluent/parallelconsumer/metrics/PCMetricsDef.java @@ -38,6 +38,7 @@ public enum PCMetricsDef { INCOMPLETE_OFFSETS_TOTAL("incomplete.offsets.total", "Total number of incomplete offsets", PCMetricsSubsystem.SHARD_MANAGER, GAUGE), SHARDS_SIZE("shards.size", "Number of records queued for processing across all shards", PCMetricsSubsystem.SHARD_MANAGER, GAUGE), + SHARDS_MAX_SIZE("shards.max.size", "Number of records queued for processing shard holds the max size queue", PCMetricsSubsystem.SHARD_MANAGER, GAUGE), //TODO: Not implemented yet - add to Metrics.adoc when implemented diff --git a/parallel-consumer-core/src/main/java/io/confluent/parallelconsumer/state/ShardManager.java b/parallel-consumer-core/src/main/java/io/confluent/parallelconsumer/state/ShardManager.java index eb9478c61..eb19cef5a 100644 --- a/parallel-consumer-core/src/main/java/io/confluent/parallelconsumer/state/ShardManager.java +++ b/parallel-consumer-core/src/main/java/io/confluent/parallelconsumer/state/ShardManager.java @@ -83,6 +83,7 @@ public class ShardManager { private Optional iterationResumePoint = Optional.empty(); private Gauge shardsSizeGauge; + private Gauge shardsMaxSizeGauge; private Gauge numberOfShardsGauge; private final PCMetrics pcMetrics; @@ -302,7 +303,10 @@ private void initMetrics() { shardsSizeGauge = pcMetrics.gaugeFromMetricDef(PCMetricsDef.SHARDS_SIZE, this, shardManager -> shardManager.processingShards.values().stream() .mapToInt(processingShard -> processingShard.getEntries().size()).sum()); + shardsMaxSizeGauge = pcMetrics.gaugeFromMetricDef(PCMetricsDef.SHARDS_MAX_SIZE, + this, shardManager -> shardManager.processingShards.values().stream() + .mapToInt(processingShard -> processingShard.getEntries().size()).max().orElse(0)); numberOfShardsGauge = pcMetrics.gaugeFromMetricDef(PCMetricsDef.NUMBER_OF_SHARDS, - this, shardManager -> shardManager.processingShards.keySet().size()); + this, shardManager -> shardManager.processingShards.size()); } } From c6d81d2b038fa6371a8c48b990224d7d94c1e1ce Mon Sep 17 00:00:00 2001 From: lixy Date: Sun, 4 Jan 2026 17:09:42 +0800 Subject: [PATCH 2/2] add doc --- CHANGELOG.adoc | 4 ++++ README.adoc | 6 ++++++ .../io/confluent/parallelconsumer/metrics/PCMetricsDef.java | 2 +- 3 files changed, 11 insertions(+), 1 deletion(-) diff --git a/CHANGELOG.adoc b/CHANGELOG.adoc index 543a21f07..f0a4e6b94 100644 --- a/CHANGELOG.adoc +++ b/CHANGELOG.adoc @@ -20,6 +20,10 @@ endif::[] * fix: keep instantiated OffsetMapCodecManager so that metrics will not be recreated every commit (#859) +=== Improvements + +* improvement: add metric to monitor the maximum number of queued records for the shard + == 0.5.3.3 === Fixes diff --git a/README.adoc b/README.adoc index 39f4db1be..e93e2f414 100644 --- a/README.adoc +++ b/README.adoc @@ -1129,6 +1129,12 @@ Gauge `pc.shards.size{subsystem=shardmanager}` Number of records queued for processing across all shards +**Max Shards Size** + +Gauge `pc.shards.max.size{subsystem=shardmanager}` + +The number of queued records in the shards with the most queued records + ==== Work Manager **Inflight Records** diff --git a/parallel-consumer-core/src/main/java/io/confluent/parallelconsumer/metrics/PCMetricsDef.java b/parallel-consumer-core/src/main/java/io/confluent/parallelconsumer/metrics/PCMetricsDef.java index 6e394c7f1..eda6028ab 100644 --- a/parallel-consumer-core/src/main/java/io/confluent/parallelconsumer/metrics/PCMetricsDef.java +++ b/parallel-consumer-core/src/main/java/io/confluent/parallelconsumer/metrics/PCMetricsDef.java @@ -38,7 +38,7 @@ public enum PCMetricsDef { INCOMPLETE_OFFSETS_TOTAL("incomplete.offsets.total", "Total number of incomplete offsets", PCMetricsSubsystem.SHARD_MANAGER, GAUGE), SHARDS_SIZE("shards.size", "Number of records queued for processing across all shards", PCMetricsSubsystem.SHARD_MANAGER, GAUGE), - SHARDS_MAX_SIZE("shards.max.size", "Number of records queued for processing shard holds the max size queue", PCMetricsSubsystem.SHARD_MANAGER, GAUGE), + SHARDS_MAX_SIZE("shards.max.size", "The number of queued records in the shards with the most queued records", PCMetricsSubsystem.SHARD_MANAGER, GAUGE), //TODO: Not implemented yet - add to Metrics.adoc when implemented