[fix][client] Fix getPendingQueueSize for PartitionedTopicProducerStatsRecorderImpl: avoid NPE and implement aggregation (apache#24830)

Co-authored-by: Zixuan Liu <nodeces@gmail.com>
(cherry picked from commit a43762b)

Author: lhotari

pulsar-client/src/main/java/org/apache/pulsar/client/impl/PartitionedTopicProducerStatsRecorderImpl.java

@@ -35,6 +35,7 @@ public class PartitionedTopicProducerStatsRecorderImpl extends ProducerStatsReco
     private final DoubleAdder sendMsgsRateAggregate;
     private final DoubleAdder sendBytesRateAggregate;
     private int partitions = 0;
+    private int pendingQueueSize;
 
     public PartitionedTopicProducerStatsRecorderImpl() {
         super();
@@ -46,6 +47,7 @@ public PartitionedTopicProducerStatsRecorderImpl() {
     void reset() {
         super.reset();
         partitions = 0;
+        pendingQueueSize = 0;
     }
 
     void updateCumulativeStats(String partition, ProducerStats stats) {
@@ -58,6 +60,7 @@ void updateCumulativeStats(String partition, ProducerStats stats) {
         sendMsgsRateAggregate.add(stats.getSendMsgsRate());
         sendBytesRateAggregate.add(stats.getSendBytesRate());
         partitions++;
+        pendingQueueSize += stats.getPendingQueueSize();
     }
 
     @Override
@@ -75,5 +78,10 @@ public Map<String, ProducerStats> getPartitionStats() {
         return partitionStats;
     }
 
+    @Override
+    public int getPendingQueueSize() {
+        return pendingQueueSize;
+    }
+
     private static final Logger log = LoggerFactory.getLogger(PartitionedTopicProducerStatsRecorderImpl.class);
 }

pulsar-client/src/main/java/org/apache/pulsar/client/impl/ProducerStatsRecorderImpl.java

@@ -332,7 +332,7 @@ public double getSendLatencyMillisMax() {
 
     @Override
     public int getPendingQueueSize() {
-        return producer.getPendingQueueSize();
+        return producer != null ? producer.getPendingQueueSize() : 0;
     }
 
     public void cancelStatsTimeout() {

pulsar-client/src/test/java/org/apache/pulsar/client/impl/ProducerStatsRecorderImplTest.java

@@ -89,4 +89,29 @@ public void testPartitionTopicAggegationStats() {
         assertTrue(recorder2.getSendBytesRate() > 0);
         assertTrue(recorder2.getSendMsgsRate() > 0);
     }
+
+    @Test
+    public void testPartitionedTopicProducerStatsPendingQueueSizeDoesntNPE() {
+        PartitionedTopicProducerStatsRecorderImpl recorder = new PartitionedTopicProducerStatsRecorderImpl();
+        assertEquals(recorder.getPendingQueueSize(), 0);
+    }
+
+    @Test
+    public void testProducerStatsPendingQueueSizeDoesntNPE() {
+        ProducerStatsRecorderImpl recorder = new ProducerStatsRecorderImpl();
+        assertEquals(recorder.getPendingQueueSize(), 0);
+    }
+
+    @Test
+    public void testPartitionedTopicProducerStatsPendingQueueSizeAggregated() {
+        PartitionedTopicProducerStatsRecorderImpl recorder = new PartitionedTopicProducerStatsRecorderImpl();
+
+        ProducerStatsRecorderImpl individualStats = spy(new ProducerStatsRecorderImpl());
+        when(individualStats.getPendingQueueSize()).thenReturn(1);
+        recorder.updateCumulativeStats("1", individualStats);
+        recorder.updateCumulativeStats("2", individualStats);
+        recorder.updateCumulativeStats("3", individualStats);
+
+        assertEquals(recorder.getPendingQueueSize(), 3);
+    }
 }