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 f0010063f..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,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", "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
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<K, V> {
     private Optional<ShardKey> 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());
     }
 }