fix: Return cached pausedPartitionSet

CHANGELOG.adoc

@@ -14,6 +14,12 @@ ifndef::github_name[]
 toc::[]
 endif::[]
 
+== 0.5.2.7
+
+=== Fixes
+
+* fix: Return cached pausedPartitionSet (#618)
+
 == 0.5.2.6
 === Improvements
 

parallel-consumer-core/src/main/java/io/confluent/parallelconsumer/internal/BrokerPollSystem.java

@@ -89,7 +89,7 @@ public BrokerPollSystem(ConsumerManager<K, V> consumerMgr, WorkManager<K, V> wm,
     private void initMetrics() {
         statusGauge = PCMetrics.getInstance().gaugeFromMetricDef(PCMetricsDef.PC_POLLER_STATUS, this, poller -> poller.runState.getValue());
         numPausedPartitionsGauge = PCMetrics.getInstance().gaugeFromMetricDef(PCMetricsDef.NUM_PAUSED_PARTITIONS,
-                this.consumerManager, consumerManager -> consumerManager.paused().size());
+                this.consumerManager, ConsumerManager::getPausedPartitionSize);
     }
 
     public void start(String managedExecutorService) {

parallel-consumer-core/src/main/java/io/confluent/parallelconsumer/internal/ConsumerManager.java

@@ -1,8 +1,9 @@
 package io.confluent.parallelconsumer.internal;
 
 /*-
- * Copyright (C) 2020-2022 Confluent, Inc.
+ * Copyright (C) 2020-2023 Confluent, Inc.
  */
+
 import lombok.RequiredArgsConstructor;
 import lombok.extern.slf4j.Slf4j;
 import org.apache.kafka.clients.consumer.*;
@@ -26,6 +27,7 @@ public class ConsumerManager<K, V> {
 
     private final AtomicBoolean pollingBroker = new AtomicBoolean(false);
 
+
     /**
      * Since Kakfa 2.7, multi-threaded access to consumer group metadata was blocked, so before and after polling, save
      * a copy of the metadata.
@@ -34,6 +36,8 @@ public class ConsumerManager<K, V> {
      */
     private ConsumerGroupMetadata metaCache;
 
+    private volatile int pausedPartitionSizeCache = 0;
+
     private int erroneousWakups = 0;
     private int correctPollWakeups = 0;
     private int noWakeups = 0;
@@ -49,11 +53,11 @@ ConsumerRecords<K, V> poll(Duration requestedLongPollTimeout) {
                 commitRequested = false;
             }
             pollingBroker.set(true);
-            updateMetadataCache();
+            updateCache();
             log.debug("Poll starting with timeout: {}", timeoutToUse);
             records = consumer.poll(timeoutToUse);
             log.debug("Poll completed normally (after timeout of {}) and returned {}...", timeoutToUse, records.count());
-            updateMetadataCache();
+            updateCache();
         } catch (WakeupException w) {
             correctPollWakeups++;
             log.debug("Awoken from broker poll");
@@ -65,8 +69,9 @@ ConsumerRecords<K, V> poll(Duration requestedLongPollTimeout) {
         return records;
     }
 
-    protected void updateMetadataCache() {
+    protected void updateCache() {
         metaCache = consumer.groupMetadata();
+        pausedPartitionSizeCache = consumer.paused().size();
     }
 
     /**
@@ -133,6 +138,10 @@ public Set<TopicPartition> paused() {
         return consumer.paused();
     }
 
+    public int getPausedPartitionSize() {
+        return pausedPartitionSizeCache;
+    }
+
     public void resume(final Set<TopicPartition> pausedTopics) {
         consumer.resume(pausedTopics);
     }

parallel-consumer-core/src/test/java/io/confluent/parallelconsumer/internal/PCModuleTestEnv.java

@@ -101,7 +101,7 @@ protected ConsumerManager<String, String> consumerManager() {
         ConsumerManager<String, String> consumerManager = super.consumerManager();
 
         // force update to set cache, otherwise maybe never called (fake consumer)
-        consumerManager.updateMetadataCache();
+        consumerManager.updateCache();
 
         return consumerManager;
     }