Added pause and resume to Java client Consumer

pulsar-client/src/main/java/org/apache/pulsar/client/api/Consumer.java

@@ -294,4 +294,14 @@ public interface Consumer<T> extends Closeable {
      * @return consumer name.
      */
     String getConsumerName();
+
+    /**
+     * Stop requesting new messages from the broker until {@link #resume()} is called.
+     */
+    void pause();
+
+    /**
+     * Resume requesting messages from the broker.
+     */
+    void resume();
 }

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

@@ -144,6 +144,8 @@ public class ConsumerImpl<T> extends ConsumerBase<T> implements ConnectionHandle
 
     private Producer<T> deadLetterProducer;
 
+    protected volatile boolean paused;
+
     enum SubscriptionMode {
         // Make the subscription to be backed by a durable cursor that will retain messages and persist the current
         // position
@@ -1073,7 +1075,7 @@ void increaseAvailablePermits(ClientCnx currentCnx) {
     private void increaseAvailablePermits(ClientCnx currentCnx, int delta) {
         int available = AVAILABLE_PERMITS_UPDATER.addAndGet(this, delta);
 
-        while (available >= receiverQueueRefillThreshold) {
+        while (available >= receiverQueueRefillThreshold && !paused) {
             if (AVAILABLE_PERMITS_UPDATER.compareAndSet(this, available, 0)) {
                 sendFlowPermitsToBroker(currentCnx, available);
                 break;
@@ -1083,6 +1085,19 @@ private void increaseAvailablePermits(ClientCnx currentCnx, int delta) {
         }
     }
 
+    @Override
+    public void pause() {
+        paused = true;
+    }
+
+    @Override
+    public void resume() {
+        if (paused) {
+            paused = false;
+            increaseAvailablePermits(cnx(), 0);
+        }
+    }
+
     private ByteBuf decryptPayloadIfNeeded(MessageIdData messageId, MessageMetadata msgMetadata, ByteBuf payload,
             ClientCnx currentCnx) {
 

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

@@ -853,5 +853,15 @@ public List<ConsumerImpl<T>> getConsumers() {
         return consumers.values().stream().collect(Collectors.toList());
     }
 
+    @Override
+    public void pause() {
+        for (Map.Entry<String, ConsumerImpl<T>> e : consumers.entrySet()) e.getValue().pause();
+    }
+
+    @Override
+    public void resume() {
+        for (Map.Entry<String, ConsumerImpl<T>> e : consumers.entrySet()) e.getValue().resume();
+    }
+
     private static final Logger log = LoggerFactory.getLogger(MultiTopicsConsumerImpl.class);
 }

pulsar-flink/src/test/java/org/apache/flink/streaming/connectors/pulsar/PulsarConsumerSourceTests.java

@@ -504,6 +504,16 @@ public boolean isConnected() {
         public String getConsumerName() {
             return "test-consumer-0";
         }
+
+        @Override
+        public void pause() {
+
+        }
+
+        @Override
+        public void resume() {
+
+        }
     }
 
     private static List<Message> createMessages(int startIndex, int numMessages) {