Partitioned numMessagesInQueue should count it's incoming messages (#78)

b33b2a19 · sschepens · Matteo Merli · 2221aee4 · b33b2a19
隐藏空白更改
内联并排

Showing with 1 addition and 1 deletion

pulsar-client/src/main/java/com/yahoo/pulsar/client/impl/PartitionedConsumerImpl.java ...com/yahoo/pulsar/client/impl/PartitionedConsumerImpl.java +1 -1

未找到文件。
--- a/pulsar-client/src/main/java/com/yahoo/pulsar/client/impl/PartitionedConsumerImpl.java
+++ b/pulsar-client/src/main/java/com/yahoo/pulsar/client/impl/PartitionedConsumerImpl.java
@@ -405,7 +405,7 @@ public class PartitionedConsumerImpl extends ConsumerBase {

    @Override
    public int numMessagesInQueue() {
-        return consumers.stream().mapToInt(ConsumerImpl::numMessagesInQueue).sum();
+        return incomingMessages.size() + consumers.stream().mapToInt(ConsumerImpl::numMessagesInQueue).sum();
    }

    @Override