Skip to content

Commit 53f496b

Browse files
astubbsclaude
andcommitted
feat: add max-queued-records-per-shard metric (cherry-pick Confluent confluentinc#905)
Cherry-pick of confluentinc#905 (author: flashmouse). Adds a SHARDS_MAX_SIZE gauge that reports the record count in the most-loaded shard. Useful with KEY ordering to detect hot-key bottlenecks. Also simplifies .keySet().size() to .size(). Upstream PR: confluentinc#905 Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
1 parent 44c72f5 commit 53f496b

2 files changed

Lines changed: 6 additions & 1 deletion

File tree

parallel-consumer-core/src/main/java/io/confluent/parallelconsumer/metrics/PCMetricsDef.java

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -38,6 +38,7 @@ public enum PCMetricsDef {
3838

3939
INCOMPLETE_OFFSETS_TOTAL("incomplete.offsets.total", "Total number of incomplete offsets", PCMetricsSubsystem.SHARD_MANAGER, GAUGE),
4040
SHARDS_SIZE("shards.size", "Number of records queued for processing across all shards", PCMetricsSubsystem.SHARD_MANAGER, GAUGE),
41+
SHARDS_MAX_SIZE("shards.max.size", "The number of queued records in the shards with the most queued records", PCMetricsSubsystem.SHARD_MANAGER, GAUGE),
4142

4243

4344
//TODO: Not implemented yet - add to Metrics.adoc when implemented

parallel-consumer-core/src/main/java/io/confluent/parallelconsumer/state/ShardManager.java

Lines changed: 5 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -83,6 +83,7 @@ public class ShardManager<K, V> {
8383
private Optional<ShardKey> iterationResumePoint = Optional.empty();
8484

8585
private Gauge shardsSizeGauge;
86+
private Gauge shardsMaxSizeGauge;
8687
private Gauge numberOfShardsGauge;
8788

8889
private final PCMetrics pcMetrics;
@@ -302,7 +303,10 @@ private void initMetrics() {
302303
shardsSizeGauge = pcMetrics.gaugeFromMetricDef(PCMetricsDef.SHARDS_SIZE,
303304
this, shardManager -> shardManager.processingShards.values().stream()
304305
.mapToInt(processingShard -> processingShard.getEntries().size()).sum());
306+
shardsMaxSizeGauge = pcMetrics.gaugeFromMetricDef(PCMetricsDef.SHARDS_MAX_SIZE,
307+
this, shardManager -> shardManager.processingShards.values().stream()
308+
.mapToInt(processingShard -> processingShard.getEntries().size()).max().orElse(0));
305309
numberOfShardsGauge = pcMetrics.gaugeFromMetricDef(PCMetricsDef.NUMBER_OF_SHARDS,
306-
this, shardManager -> shardManager.processingShards.keySet().size());
310+
this, shardManager -> shardManager.processingShards.size());
307311
}
308312
}

0 commit comments

Comments
 (0)