Add blockedConsumerOnUnackedMsgs flag in consumer-stats (#124)

pulsar-broker/src/main/java/com/yahoo/pulsar/broker/service/Consumer.java

@@ -368,6 +368,7 @@ public class Consumer {
     public ConsumerStats getStats() {
         stats.availablePermits = getAvailablePermits();
         stats.unackedMessages = unackedMessages.get();
+        stats.blockedConsumerOnUnackedMsgs = blockedConsumerOnUnackedMsgs;
         return stats;
     }
 

pulsar-broker/src/main/java/com/yahoo/pulsar/broker/service/persistent/PersistentTopic.java

@@ -845,6 +845,7 @@ public class PersistentTopic implements Topic, AddEntryCallback {
                     destStatsStream.writePair("consumerName", consumerStats.consumerName);
                     destStatsStream.writePair("availablePermits", consumerStats.availablePermits);
                     destStatsStream.writePair("unackedMessages", consumerStats.unackedMessages);
+                    destStatsStream.writePair("blockedConsumerOnUnackedMsgs", consumerStats.blockedConsumerOnUnackedMsgs);
                     destStatsStream.writePair("connectedSince", consumerStats.connectedSince);
                     destStatsStream.writePair("msgRateOut", consumerStats.msgRateOut);
                     destStatsStream.writePair("msgThroughputOut", consumerStats.msgThroughputOut);

pulsar-common/src/main/java/com/yahoo/pulsar/common/policies/data/ConsumerStats.java

@@ -37,6 +37,9 @@ public class ConsumerStats {
     
     /** Number of unacknowledged messages for the consumer */
     public int unackedMessages;
+    
+    /** Flag to verify if consumer is blocked due to reaching threshold of unacked messages */
+    public boolean blockedConsumerOnUnackedMsgs;
 
     /** Address of this consumer */
     public String address;
@@ -51,6 +54,7 @@ public class ConsumerStats {
         this.msgRateRedeliver += stats.msgRateRedeliver;
         this.availablePermits += stats.availablePermits;
         this.unackedMessages += stats.unackedMessages;
+        this.blockedConsumerOnUnackedMsgs = stats.blockedConsumerOnUnackedMsgs;
         return this;
     }
 }