From d7a31312c444fecefaa8cfba23a86ab799bd70ec Mon Sep 17 00:00:00 2001 From: Jai ASher Date: Tue, 28 Feb 2017 20:17:37 -0800 Subject: [PATCH 1/6] Java Client - Support for getting consumer stats from broker --- .../client/impl/BrokerConsumerStatsTest.java | 170 ++++++++++++++++++ .../com/yahoo/pulsar/client/api/Consumer.java | 10 ++ .../client/api/ConsumerConfiguration.java | 23 +++ .../client/api/PulsarClientException.java | 31 ++++ .../client/impl/BrokerConsumerStats.java | 168 +++++++++++++++++ .../yahoo/pulsar/client/impl/ClientCnx.java | 61 ++++++- .../pulsar/client/impl/ConsumerBase.java | 2 + .../pulsar/client/impl/ConsumerImpl.java | 21 +++ .../client/impl/PartitionedConsumerImpl.java | 11 ++ .../com/yahoo/pulsar/common/api/Commands.java | 18 +- 10 files changed, 509 insertions(+), 6 deletions(-) create mode 100644 pulsar-broker/src/test/java/com/yahoo/pulsar/client/impl/BrokerConsumerStatsTest.java create mode 100644 pulsar-client/src/main/java/com/yahoo/pulsar/client/impl/BrokerConsumerStats.java diff --git a/pulsar-broker/src/test/java/com/yahoo/pulsar/client/impl/BrokerConsumerStatsTest.java b/pulsar-broker/src/test/java/com/yahoo/pulsar/client/impl/BrokerConsumerStatsTest.java new file mode 100644 index 0000000000000..3945d246dafda --- /dev/null +++ b/pulsar-broker/src/test/java/com/yahoo/pulsar/client/impl/BrokerConsumerStatsTest.java @@ -0,0 +1,170 @@ +/** + * Copyright 2016 Yahoo Inc. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package com.yahoo.pulsar.client.impl; + +import java.util.concurrent.TimeUnit; + +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; +import org.testng.Assert; +import org.testng.annotations.AfterClass; +import org.testng.annotations.BeforeClass; +import org.testng.annotations.Test; + +import com.yahoo.pulsar.broker.service.BrokerTestBase; +import com.yahoo.pulsar.client.api.Consumer; +import com.yahoo.pulsar.client.api.ConsumerConfiguration; +import com.yahoo.pulsar.client.api.Message; +import com.yahoo.pulsar.client.api.Producer; +import com.yahoo.pulsar.client.api.ProducerConfiguration; +import com.yahoo.pulsar.client.api.PulsarClientException; +import com.yahoo.pulsar.client.api.SubscriptionType; + +public class BrokerConsumerStatsTest extends BrokerTestBase { + private static final Logger log = LoggerFactory.getLogger(BrokerConsumerStatsTest.class); + private String topicName = "persistent://prop/cluster/ns/topic-"; + + @BeforeClass + @Override + public void setup() throws Exception { + baseSetup(); + } + + @AfterClass + @Override + protected void cleanup() throws Exception { + internalCleanup(); + } + + @Test() + public void testSharedSubscriptionMessageBacklog() throws PulsarClientException { + int totalMessages = 50; + String topicNamePostFix = "testSharedSubscriptionMessageBacklog"; + ConsumerConfiguration conf = new ConsumerConfiguration(); + conf.setSubscriptionType(SubscriptionType.Shared); + Consumer consumer1 = pulsarClient.subscribe(topicName + topicNamePostFix, "my-subscriber-name", conf); + Consumer consumer2 = pulsarClient.subscribe(topicName + topicNamePostFix, "my-subscriber-name", conf); + Consumer consumer3 = pulsarClient.subscribe(topicName + topicNamePostFix, "my-subscriber-name", conf); + + Producer producer = pulsarClient.createProducer(topicName + topicNamePostFix, new ProducerConfiguration()); + for (int i = 0; i < totalMessages; i++) { + String message = "my-message-" + i; + producer.send(message.getBytes()); + } + + int count = 0; + try { + Message msg = null; + while (true) { + msg = consumer1.receive(10, TimeUnit.MILLISECONDS); + if (msg != null) { + consumer1.acknowledge(msg); + count++; + } else { + break; + } + } + + log.debug(consumer1.getBrokerConsumerStatsAsync().get().toString()); + Assert.assertEquals(consumer1.getBrokerConsumerStatsAsync().get().getSubscriptionType(), + SubscriptionType.Shared); + Assert.assertEquals(consumer1.getBrokerConsumerStatsAsync().get().getMsgBacklog(), totalMessages - count); + + while (true) { + msg = consumer2.receive(10, TimeUnit.MILLISECONDS); + if (msg != null) { + consumer2.acknowledge(msg); + count++; + } else { + break; + } + } + + log.debug(consumer2.getBrokerConsumerStatsAsync().get().toString()); + Assert.assertEquals(consumer2.getBrokerConsumerStatsAsync().get().getMsgBacklog(), totalMessages - count); + + while (true) { + msg = consumer3.receive(10, TimeUnit.MILLISECONDS); + if (msg != null) { + consumer3.acknowledge(msg); + count++; + } else { + break; + } + } + + log.debug(consumer3.getBrokerConsumerStatsAsync().get().toString()); + Assert.assertEquals(consumer3.getBrokerConsumerStatsAsync().get().getMsgBacklog(), 0); + } catch (Exception ex) { + Assert.fail("Exception:" + ex); + } finally { + consumer1.close(); + consumer2.close(); + consumer3.close(); + } + } + + @Test + public void testCachingMechanism() throws PulsarClientException { + int totalMessages = 50; + String topicNamePostFix = "testCachingMechanism"; + + ConsumerConfiguration conf = new ConsumerConfiguration(); + conf.setBrokerConsumerStatsCacheTime(3, TimeUnit.SECONDS); + Consumer consumer = pulsarClient.subscribe(topicName + topicNamePostFix, "my-subscriber-name", conf); + + Producer producer = pulsarClient.createProducer(topicName + topicNamePostFix, new ProducerConfiguration()); + for (int i = 0; i < totalMessages; i++) { + String message = "my-message-" + i; + producer.send(message.getBytes()); + } + + try { + BrokerConsumerStats stats = consumer.getBrokerConsumerStatsAsync().get(); + Assert.assertEquals(consumer.getBrokerConsumerStatsAsync().get().getSubscriptionType(), + SubscriptionType.Exclusive); + Assert.assertEquals(consumer.getBrokerConsumerStatsAsync().get().getMsgBacklog(), totalMessages); + Assert.assertEquals(stats, consumer.getBrokerConsumerStatsAsync().get()); + Assert.assertTrue(stats.isValid()); + Message msg = null; + while (true) { + msg = consumer.receive(10, TimeUnit.MILLISECONDS); + if (msg != null) { + consumer.acknowledge(msg); + } else { + break; + } + } + + Thread.sleep(2 * 1000); + // Cached results returned + Assert.assertEquals(consumer.getBrokerConsumerStatsAsync().get().getMsgBacklog(), totalMessages); + Assert.assertTrue(consumer.getBrokerConsumerStatsAsync().get().isValid()); + Assert.assertEquals(stats, consumer.getBrokerConsumerStatsAsync().get()); + Assert.assertTrue(stats.isValid()); + + // Waiting for cache time to expire + Thread.sleep(2 * 1000); + Assert.assertEquals(consumer.getBrokerConsumerStatsAsync().get().getMsgBacklog(), 0); + Assert.assertNotEquals(stats, consumer.getBrokerConsumerStatsAsync().get()); + Assert.assertFalse(stats.isValid()); + } catch (Exception ex) { + Assert.fail("Exception:" + ex); + } finally { + consumer.close(); + } + } +} diff --git a/pulsar-client/src/main/java/com/yahoo/pulsar/client/api/Consumer.java b/pulsar-client/src/main/java/com/yahoo/pulsar/client/api/Consumer.java index 68edcaaa8d2a4..5ecb7737d4540 100644 --- a/pulsar-client/src/main/java/com/yahoo/pulsar/client/api/Consumer.java +++ b/pulsar-client/src/main/java/com/yahoo/pulsar/client/api/Consumer.java @@ -19,6 +19,7 @@ import java.util.concurrent.CompletableFuture; import java.util.concurrent.TimeUnit; +import com.yahoo.pulsar.client.impl.BrokerConsumerStats; import com.yahoo.pulsar.client.impl.ConsumerStats; /** @@ -231,4 +232,13 @@ public interface Consumer extends Closeable { * breaks, the messages are redelivered after reconnect. */ void redeliverUnacknowledgedMessages(); + + /** + * Gets Consumer Stats from broker. The stats are cached for + * {@link ConsumerConfiguration.getBrokerConsumerStatsCacheTimeInMs()} milliseconds. + * + * @return a Completeable future for BrokerConsumerStats + */ + CompletableFuture getBrokerConsumerStatsAsync(); + } diff --git a/pulsar-client/src/main/java/com/yahoo/pulsar/client/api/ConsumerConfiguration.java b/pulsar-client/src/main/java/com/yahoo/pulsar/client/api/ConsumerConfiguration.java index b7cf0bde45cbf..9c6c7a3b61abf 100644 --- a/pulsar-client/src/main/java/com/yahoo/pulsar/client/api/ConsumerConfiguration.java +++ b/pulsar-client/src/main/java/com/yahoo/pulsar/client/api/ConsumerConfiguration.java @@ -48,6 +48,29 @@ public class ConsumerConfiguration implements Serializable { private long ackTimeoutMillis = 0; private int priorityLevel = 0; + + private long brokerConsumerStatsCacheTimeInMs = 30 * 1000; // 30 seconds + + /** + * @return the cache time in milliseconds for broker consumer stats. + */ + public long getBrokerConsumerStatsCacheTimeInMs() { + return brokerConsumerStatsCacheTimeInMs; + } + + /** + * Set cache time for broker consumer stats. + * + * @param brokerConsumerStatsCacheTime + * cache time for broker consumer stats. + * @param timeUnit + * unit in which the brokerConsumerStatsCacheTime is provided. + * @return {@link ConsumerConfiguration} + */ + public ConsumerConfiguration setBrokerConsumerStatsCacheTime(long brokerConsumerStatsCacheTime, TimeUnit timeUnit) { + this.brokerConsumerStatsCacheTimeInMs = timeUnit.toMillis(brokerConsumerStatsCacheTime); + return this; + } /** * @return the configured timeout in milliseconds for unacked messages. diff --git a/pulsar-client/src/main/java/com/yahoo/pulsar/client/api/PulsarClientException.java b/pulsar-client/src/main/java/com/yahoo/pulsar/client/api/PulsarClientException.java index eb0918b4eeca1..93e7e98d5548d 100644 --- a/pulsar-client/src/main/java/com/yahoo/pulsar/client/api/PulsarClientException.java +++ b/pulsar-client/src/main/java/com/yahoo/pulsar/client/api/PulsarClientException.java @@ -24,6 +24,7 @@ */ @SuppressWarnings("serial") public class PulsarClientException extends IOException { + public PulsarClientException(String msg) { super(msg); } @@ -175,4 +176,34 @@ public ChecksumException(String msg) { super(msg); } } + + public static class TopicNotFoundException extends PulsarClientException { + public TopicNotFoundException(String msg) { + super(msg); + } + } + + public static class ConsumerIdNotFoundException extends PulsarClientException { + public ConsumerIdNotFoundException(String msg) { + super(msg); + } + } + + public static class SubscriptionNotFoundException extends PulsarClientException { + public SubscriptionNotFoundException(String msg) { + super(msg); + } + } + + public static class UnknownError extends PulsarClientException { + public UnknownError(String msg) { + super(msg); + } + } + + public static class UnsupportedVersionError extends PulsarClientException { + public UnsupportedVersionError(String msg) { + super(msg); + } + } } \ No newline at end of file diff --git a/pulsar-client/src/main/java/com/yahoo/pulsar/client/impl/BrokerConsumerStats.java b/pulsar-client/src/main/java/com/yahoo/pulsar/client/impl/BrokerConsumerStats.java new file mode 100644 index 0000000000000..813533bc3efb9 --- /dev/null +++ b/pulsar-client/src/main/java/com/yahoo/pulsar/client/impl/BrokerConsumerStats.java @@ -0,0 +1,168 @@ +/** + * Copyright 2016 Yahoo Inc. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package com.yahoo.pulsar.client.impl; + +import com.yahoo.pulsar.client.api.SubscriptionType; +import com.yahoo.pulsar.common.api.proto.PulsarApi.CommandConsumerStatsResponse; + +public class BrokerConsumerStats { + + private final String DELIMITER = ";"; + + /** validTillInMs - Stats will be valid till this time. */ + private long validTillInMs = System.currentTimeMillis(); + + /** Total rate of messages delivered to the consumer. msg/s */ + private double msgRateOut = 0; + + /** Total throughput delivered to the consumer. bytes/s */ + private double msgThroughputOut = 0; + + /** Total rate of messages redelivered by this consumer. msg/s */ + private double msgRateRedeliver = 0; + + /** Name of the consumer */ + private String consumerName = ""; + + /** Number of available message permits for the consumer */ + private long availablePermits = 0; + + /** Number of unacknowledged messages for the consumer */ + private long unackedMessages = 0; + + /** Flag to verify if consumer is blocked due to reaching threshold of unacked messages */ + private boolean blockedConsumerOnUnackedMsgs = false; + + /** Address of this consumer */ + private String address = ""; + + /** Timestamp of connection */ + private String connectedSince = ""; + + /** Whether this subscription is Exclusive or Shared or Failover */ + private SubscriptionType subscriptionType = null; + + /** Total rate of messages expired on this subscription. msg/s */ + private double msgRateExpired = 0; + + /** Number of messages in the subscription backlog */ + private long msgBacklog = 0; + + public BrokerConsumerStats(CommandConsumerStatsResponse response) { + super(); + this.validTillInMs = System.currentTimeMillis(); + this.msgRateOut = response.getMsgRateOut(); + this.msgThroughputOut = response.getMsgThroughputOut(); + this.msgRateRedeliver = response.getMsgRateRedeliver(); + this.consumerName = response.getConsumerName(); + this.availablePermits = response.getAvailablePermits(); + this.unackedMessages = response.getUnackedMessages(); + this.blockedConsumerOnUnackedMsgs = response.getBlockedConsumerOnUnackedMsgs(); + this.address = response.getAddress(); + this.connectedSince = response.getConnectedSince(); + this.subscriptionType = SubscriptionType.valueOf(response.getType()); + this.msgRateExpired = response.getMsgRateExpired(); + this.msgBacklog = response.getMsgBacklog(); + } + + public BrokerConsumerStats() { + } + + public void setCacheTime(long timeInMs) { + validTillInMs = System.currentTimeMillis() + timeInMs; + } + + /** Returns true if the Message is Expired **/ + public synchronized boolean isValid() { + return System.currentTimeMillis() <= validTillInMs; + } + + public double getMsgRateOut() { + return msgRateOut; + } + + public double getMsgThroughputOut() { + return msgThroughputOut; + } + + public double getMsgRateRedeliver() { + return msgRateRedeliver; + } + + public String getConsumerName() { + return consumerName; + } + + public long getAvailablePermits() { + return availablePermits; + } + + public long getUnackedMessages() { + return unackedMessages; + } + + public boolean isBlockedConsumerOnUnackedMsgs() { + return blockedConsumerOnUnackedMsgs; + } + + public String getAddress() { + return address; + } + + public String getConnectedSince() { + return connectedSince; + } + + public SubscriptionType getSubscriptionType() { + return subscriptionType; + } + + public double getMsgRateExpired() { + return msgRateExpired; + } + + public long getMsgBacklog() { + return msgBacklog; + } + + public synchronized void add(BrokerConsumerStats stats) { + this.validTillInMs = (validTillInMs > stats.validTillInMs) ? validTillInMs : stats.validTillInMs; + this.msgRateOut += stats.msgRateOut; + this.msgThroughputOut += stats.msgThroughputOut; + this.msgRateRedeliver += stats.msgRateRedeliver; + this.consumerName += stats.getConsumerName() + DELIMITER; + this.availablePermits += stats.getAvailablePermits(); + this.unackedMessages += stats.unackedMessages; + this.blockedConsumerOnUnackedMsgs |= stats.blockedConsumerOnUnackedMsgs; + this.address += stats.address + DELIMITER; + this.connectedSince += stats.connectedSince + DELIMITER; + if (this.subscriptionType == null) { + this.subscriptionType = stats.getSubscriptionType(); + } + this.msgRateExpired += stats.msgRateExpired; + this.msgBacklog += stats.msgBacklog; + } + + @Override + public String toString() { + return "BrokerConsumerStats [validTillInMs=" + validTillInMs + ", msgRateOut=" + msgRateOut + + ", msgThroughputOut=" + msgThroughputOut + ", msgRateRedeliver=" + msgRateRedeliver + + ", consumerName=" + consumerName + ", availablePermits=" + availablePermits + ", unackedMessages=" + + unackedMessages + ", blockedConsumerOnUnackedMsgs=" + blockedConsumerOnUnackedMsgs + ", address=" + + address + ", connectedSince=" + connectedSince + ", type=" + subscriptionType + ", msgRateExpired=" + + msgRateExpired + ", msgBacklog=" + msgBacklog + "]"; + } +} diff --git a/pulsar-client/src/main/java/com/yahoo/pulsar/client/impl/ClientCnx.java b/pulsar-client/src/main/java/com/yahoo/pulsar/client/impl/ClientCnx.java index 51ac4e08eac25..fb4676f10a725 100644 --- a/pulsar-client/src/main/java/com/yahoo/pulsar/client/impl/ClientCnx.java +++ b/pulsar-client/src/main/java/com/yahoo/pulsar/client/impl/ClientCnx.java @@ -33,6 +33,8 @@ import com.yahoo.pulsar.common.api.proto.PulsarApi.CommandCloseConsumer; import com.yahoo.pulsar.common.api.proto.PulsarApi.CommandCloseProducer; import com.yahoo.pulsar.common.api.proto.PulsarApi.CommandConnected; +import com.yahoo.pulsar.common.api.proto.PulsarApi.CommandConsumerStats; +import com.yahoo.pulsar.common.api.proto.PulsarApi.CommandConsumerStatsResponse; import com.yahoo.pulsar.common.api.proto.PulsarApi.CommandError; import com.yahoo.pulsar.common.api.proto.PulsarApi.CommandLookupTopicResponse; import com.yahoo.pulsar.common.api.proto.PulsarApi.CommandMessage; @@ -56,6 +58,7 @@ public class ClientCnx extends PulsarHandler { private final ConcurrentLongHashMap> pendingRequests = new ConcurrentLongHashMap<>(16, 1); private final ConcurrentLongHashMap> pendingLookupRequests = new ConcurrentLongHashMap<>(16, 1); + private final ConcurrentLongHashMap> pendingConsumerStatsRequests = new ConcurrentLongHashMap<>(16, 1); private final ConcurrentLongHashMap producers = new ConcurrentLongHashMap<>(16, 1); private final ConcurrentLongHashMap consumers = new ConcurrentLongHashMap<>(16, 1); @@ -110,7 +113,8 @@ public void channelInactive(ChannelHandlerContext ctx) throws Exception { // Fail out all the pending ops pendingRequests.forEach((key, future) -> future.completeExceptionally(e)); pendingLookupRequests.forEach((key, future) -> getAndRemovePendingLookupRequest(key).completeExceptionally(e)); - + pendingConsumerStatsRequests.forEach((key, future) -> future.completeExceptionally(e)); + pendingConsumerStatsRequests.clear(); // Notify all attached producers/consumers so they have a chance to reconnect producers.forEach((id, producer) -> producer.connectionClosed(this)); consumers.forEach((id, consumer) -> consumer.connectionClosed(this)); @@ -228,6 +232,26 @@ protected void handleLookupResponse(CommandLookupTopicResponse lookupResult) { log.warn("{} Received unknown request id from server: {}", ctx.channel(), lookupResult.getRequestId()); } } + + @Override + protected void handleConsumerStatsResponse(CommandConsumerStatsResponse response) { + if (log.isDebugEnabled()) { + log.debug("Received Consumer Stats response for request id: {}", response.getRequestId()); + } + long requestId = response.getRequestId(); + CompletableFuture future = pendingConsumerStatsRequests.remove(requestId); + + if (future != null) { + if (response.hasErrorCode()) { + future.completeExceptionally(getPulsarClientException(response.getErrorCode(), + response.hasErrorMessage() ? response.getErrorMessage() : null)); + } else { + future.complete(new BrokerConsumerStats(response)); + } + } else { + log.warn("{} Received unknown request id from server: {}", ctx.channel(), requestId); + } + } @Override protected void handlePartitionResponse(CommandPartitionedTopicMetadataResponse lookupResult) { @@ -339,6 +363,7 @@ CompletableFuture newLookup(ByteBuf request, long requestId) { log.warn("{} Failed to send request {} to broker: {}", ctx.channel(), requestId, writeFuture.cause().getMessage()); future.completeExceptionally(writeFuture.cause()); + getAndRemovePendingLookupRequest(requestId); } }); } else { @@ -350,7 +375,22 @@ CompletableFuture newLookup(ByteBuf request, long requestId) { } return future; } - + + CompletableFuture newConsumerStats(String topicName, String subscriptionName, long consumerId, long requestId) { + CompletableFuture future = new CompletableFuture<>(); + ByteBuf request = Commands.newConsumerStats(topicName, subscriptionName, consumerId, requestId); + ctx.writeAndFlush(request).addListener(writeFuture -> { + if (!writeFuture.isSuccess()) { + log.warn("{} Failed to send request {} to broker: {}", ctx.channel(), requestId, + writeFuture.cause().getMessage()); + future.completeExceptionally(writeFuture.cause()); + } else { + pendingConsumerStatsRequests.put(requestId, future); + } + }); + return future; + } + Promise newPromise() { return ctx.newPromise(); } @@ -420,9 +460,22 @@ private PulsarClientException getPulsarClientException(ServerError error, String case ProducerBlockedQuotaExceededException: return new PulsarClientException.ProducerBlockedQuotaExceededException(errorMsg); case UnknownError: - default: - return new PulsarClientException(errorMsg); + return new PulsarClientException.UnknownError(errorMsg); + case TopicNotFound: + return new PulsarClientException.UnknownError(errorMsg); + case ConsumerNotFound: + return new PulsarClientException.UnknownError(errorMsg); + case SubscriptionNotFound: + return new PulsarClientException.UnknownError(errorMsg); + case ChecksumError: + return new PulsarClientException.ChecksumException(errorMsg); + case UnsupportedVersionError: + return new PulsarClientException.UnsupportedVersionError(errorMsg); } + // NOTE : Do not add default case in the switch above. In future if we get new cases for + // ServerError and miss them in the switch above we will get a warning. + // Adding return here to make the compiler happy. + return new PulsarClientException(errorMsg); } private static final Logger log = LoggerFactory.getLogger(ClientCnx.class); diff --git a/pulsar-client/src/main/java/com/yahoo/pulsar/client/impl/ConsumerBase.java b/pulsar-client/src/main/java/com/yahoo/pulsar/client/impl/ConsumerBase.java index 050dada7dd599..47c84988a2699 100644 --- a/pulsar-client/src/main/java/com/yahoo/pulsar/client/impl/ConsumerBase.java +++ b/pulsar-client/src/main/java/com/yahoo/pulsar/client/impl/ConsumerBase.java @@ -318,4 +318,6 @@ public String getSubscription() { * breaks, the messages are redelivered after reconnect. */ protected abstract void redeliverUnacknowledgedMessages(Set messageIds); + + public abstract CompletableFuture getBrokerConsumerStatsAsync(); } diff --git a/pulsar-client/src/main/java/com/yahoo/pulsar/client/impl/ConsumerImpl.java b/pulsar-client/src/main/java/com/yahoo/pulsar/client/impl/ConsumerImpl.java index fb160131d61c2..1cce20a3d68d4 100644 --- a/pulsar-client/src/main/java/com/yahoo/pulsar/client/impl/ConsumerImpl.java +++ b/pulsar-client/src/main/java/com/yahoo/pulsar/client/impl/ConsumerImpl.java @@ -92,6 +92,7 @@ public class ConsumerImpl extends ConsumerBase { private final ConsumerStats stats; private final int priorityLevel; + private volatile BrokerConsumerStats brokerConsumerStats = new BrokerConsumerStats(); ConsumerImpl(PulsarClientImpl client, String topic, String subscription, ConsumerConfiguration conf, ExecutorService listenerExecutor, CompletableFuture subscribeFuture) { @@ -1015,4 +1016,24 @@ public ConsumerStats getStats() { private static final Logger log = LoggerFactory.getLogger(ConsumerImpl.class); + @Override + public CompletableFuture getBrokerConsumerStatsAsync() { + if (getState() != State.Ready || !isConnected()) { + return FutureUtil.failedFuture(new PulsarClientException.NotConnectedException()); + } + + if (brokerConsumerStats.isValid()) { + CompletableFuture future = new CompletableFuture(); + future.complete(brokerConsumerStats); + return future; + } + long requestId = client.newRequestId(); + return cnx().newConsumerStats(topic, subscription, consumerId, requestId) + .thenApplyAsync(brokerConsumerStats -> { + brokerConsumerStats.setCacheTime(conf.getBrokerConsumerStatsCacheTimeInMs()); + this.brokerConsumerStats = brokerConsumerStats; + return brokerConsumerStats; + }); + } + } diff --git 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 index 5ce7ba662d96d..43b3629753fa0 100644 --- 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 @@ -17,6 +17,7 @@ import static com.google.common.base.Preconditions.checkArgument; +import java.util.ArrayList; import java.util.HashSet; import java.util.List; import java.util.Set; @@ -449,4 +450,14 @@ public synchronized ConsumerStats getStats() { } private static final Logger log = LoggerFactory.getLogger(PartitionedConsumerImpl.class); + + @Override + public CompletableFuture getBrokerConsumerStatsAsync() { + BrokerConsumerStats brokerConsumerStats = new BrokerConsumerStats(); + List> futures = Lists.newArrayList(); + for (Consumer c : consumers) { + futures.add(c.getBrokerConsumerStatsAsync().thenAcceptAsync(stats -> {brokerConsumerStats.add(stats);})); + } + return FutureUtil.waitForAll(futures).thenApplyAsync( r -> {return brokerConsumerStats;}); + } } diff --git a/pulsar-common/src/main/java/com/yahoo/pulsar/common/api/Commands.java b/pulsar-common/src/main/java/com/yahoo/pulsar/common/api/Commands.java index ec6e16d093f26..6b07d4962c7ba 100644 --- a/pulsar-common/src/main/java/com/yahoo/pulsar/common/api/Commands.java +++ b/pulsar-common/src/main/java/com/yahoo/pulsar/common/api/Commands.java @@ -33,6 +33,7 @@ import com.yahoo.pulsar.common.api.proto.PulsarApi.CommandCloseProducer; import com.yahoo.pulsar.common.api.proto.PulsarApi.CommandConnect; import com.yahoo.pulsar.common.api.proto.PulsarApi.CommandConnected; +import com.yahoo.pulsar.common.api.proto.PulsarApi.CommandConsumerStats; import com.yahoo.pulsar.common.api.proto.PulsarApi.CommandConsumerStatsResponse; import com.yahoo.pulsar.common.api.proto.PulsarApi.CommandError; import com.yahoo.pulsar.common.api.proto.PulsarApi.CommandFlow; @@ -52,14 +53,13 @@ import com.yahoo.pulsar.common.api.proto.PulsarApi.CommandSendReceipt; import com.yahoo.pulsar.common.api.proto.PulsarApi.CommandSubscribe; import com.yahoo.pulsar.common.api.proto.PulsarApi.CommandSubscribe.SubType; -import com.yahoo.pulsar.common.policies.data.ConsumerStats; import com.yahoo.pulsar.common.api.proto.PulsarApi.CommandSuccess; import com.yahoo.pulsar.common.api.proto.PulsarApi.CommandUnsubscribe; -import com.yahoo.pulsar.common.api.proto.PulsarApi.KeyValue; import com.yahoo.pulsar.common.api.proto.PulsarApi.MessageIdData; import com.yahoo.pulsar.common.api.proto.PulsarApi.MessageMetadata; import com.yahoo.pulsar.common.api.proto.PulsarApi.ProtocolVersion; import com.yahoo.pulsar.common.api.proto.PulsarApi.ServerError; +import com.yahoo.pulsar.common.policies.data.ConsumerStats; import com.yahoo.pulsar.common.util.protobuf.ByteBufCodedInputStream; import com.yahoo.pulsar.common.util.protobuf.ByteBufCodedOutputStream; @@ -530,6 +530,20 @@ public static ByteBuf newConsumerStatsResponse(CommandConsumerStatsResponse.Buil builder.recycle(); return res; } + + public static ByteBuf newConsumerStats(String topicName, String subscriptionName, long consumerId, long requestId) { + CommandConsumerStats.Builder consumerStatsBuilder = CommandConsumerStats.newBuilder(); + consumerStatsBuilder.setTopicName(topicName); + consumerStatsBuilder.setSubscriptionName(subscriptionName); + consumerStatsBuilder.setConsumerId(consumerId); + consumerStatsBuilder.setRequestId(requestId); + + CommandConsumerStats consumerStats = consumerStatsBuilder.build(); + ByteBuf res = serializeWithSize(BaseCommand.newBuilder().setType(Type.CONSUMER_STATS).setConsumerStats(consumerStatsBuilder)); + consumerStats.recycle(); + consumerStatsBuilder.recycle(); + return res; + } private final static ByteBuf cmdPing = serializeWithSize( BaseCommand.newBuilder().setType(Type.PING).setPing(CommandPing.getDefaultInstance())); From da3ad89ca66f38df9e92ceec53fbc5c0903abc40 Mon Sep 17 00:00:00 2001 From: Jai ASher Date: Fri, 17 Mar 2017 10:55:55 -0700 Subject: [PATCH 2/6] Implemented Rajans suggestion to create single interface and split the impl --- .../BrokerConsumerStatsTest.java | 65 +++++++++- .../client/api/BrokerConsumerStats.java | 84 +++++++++++++ .../com/yahoo/pulsar/client/api/Consumer.java | 7 +- ...tats.java => BrokerConsumerStatsImpl.java} | 9 +- .../yahoo/pulsar/client/impl/ClientCnx.java | 10 +- .../pulsar/client/impl/ConsumerImpl.java | 8 +- .../PartitionedBrokerConsumerStatsImpl.java | 118 ++++++++++++++++++ .../client/impl/PartitionedConsumerImpl.java | 19 ++- 8 files changed, 295 insertions(+), 25 deletions(-) rename pulsar-broker/src/test/java/com/yahoo/pulsar/client/{impl => api}/BrokerConsumerStatsTest.java (69%) create mode 100644 pulsar-client/src/main/java/com/yahoo/pulsar/client/api/BrokerConsumerStats.java rename pulsar-client/src/main/java/com/yahoo/pulsar/client/impl/{BrokerConsumerStats.java => BrokerConsumerStatsImpl.java} (94%) create mode 100644 pulsar-client/src/main/java/com/yahoo/pulsar/client/impl/PartitionedBrokerConsumerStatsImpl.java diff --git a/pulsar-broker/src/test/java/com/yahoo/pulsar/client/impl/BrokerConsumerStatsTest.java b/pulsar-broker/src/test/java/com/yahoo/pulsar/client/api/BrokerConsumerStatsTest.java similarity index 69% rename from pulsar-broker/src/test/java/com/yahoo/pulsar/client/impl/BrokerConsumerStatsTest.java rename to pulsar-broker/src/test/java/com/yahoo/pulsar/client/api/BrokerConsumerStatsTest.java index 3945d246dafda..fc9677fda3b82 100644 --- a/pulsar-broker/src/test/java/com/yahoo/pulsar/client/impl/BrokerConsumerStatsTest.java +++ b/pulsar-broker/src/test/java/com/yahoo/pulsar/client/api/BrokerConsumerStatsTest.java @@ -13,7 +13,7 @@ * See the License for the specific language governing permissions and * limitations under the License. */ -package com.yahoo.pulsar.client.impl; +package com.yahoo.pulsar.client.api; import java.util.concurrent.TimeUnit; @@ -25,13 +25,16 @@ import org.testng.annotations.Test; import com.yahoo.pulsar.broker.service.BrokerTestBase; +import com.yahoo.pulsar.client.admin.PulsarAdminException; import com.yahoo.pulsar.client.api.Consumer; import com.yahoo.pulsar.client.api.ConsumerConfiguration; import com.yahoo.pulsar.client.api.Message; import com.yahoo.pulsar.client.api.Producer; import com.yahoo.pulsar.client.api.ProducerConfiguration; +import com.yahoo.pulsar.client.api.ProducerConfiguration.MessageRoutingMode; import com.yahoo.pulsar.client.api.PulsarClientException; import com.yahoo.pulsar.client.api.SubscriptionType; +import com.yahoo.pulsar.client.impl.PartitionedBrokerConsumerStatsImpl; public class BrokerConsumerStatsTest extends BrokerTestBase { private static final Logger log = LoggerFactory.getLogger(BrokerConsumerStatsTest.class); @@ -53,12 +56,15 @@ protected void cleanup() throws Exception { public void testSharedSubscriptionMessageBacklog() throws PulsarClientException { int totalMessages = 50; String topicNamePostFix = "testSharedSubscriptionMessageBacklog"; + + // 1. Create consumers ConsumerConfiguration conf = new ConsumerConfiguration(); conf.setSubscriptionType(SubscriptionType.Shared); Consumer consumer1 = pulsarClient.subscribe(topicName + topicNamePostFix, "my-subscriber-name", conf); Consumer consumer2 = pulsarClient.subscribe(topicName + topicNamePostFix, "my-subscriber-name", conf); Consumer consumer3 = pulsarClient.subscribe(topicName + topicNamePostFix, "my-subscriber-name", conf); + // 2. Create producers and produce messages Producer producer = pulsarClient.createProducer(topicName + topicNamePostFix, new ProducerConfiguration()); for (int i = 0; i < totalMessages; i++) { String message = "my-message-" + i; @@ -67,6 +73,7 @@ public void testSharedSubscriptionMessageBacklog() throws PulsarClientException int count = 0; try { + // 3. Consumer all messages routed to consumer1 Message msg = null; while (true) { msg = consumer1.receive(10, TimeUnit.MILLISECONDS); @@ -78,11 +85,13 @@ public void testSharedSubscriptionMessageBacklog() throws PulsarClientException } } + // 4. Check subscription type log.debug(consumer1.getBrokerConsumerStatsAsync().get().toString()); Assert.assertEquals(consumer1.getBrokerConsumerStatsAsync().get().getSubscriptionType(), SubscriptionType.Shared); Assert.assertEquals(consumer1.getBrokerConsumerStatsAsync().get().getMsgBacklog(), totalMessages - count); + // 5. Consumer all messages routed to consumer2 while (true) { msg = consumer2.receive(10, TimeUnit.MILLISECONDS); if (msg != null) { @@ -93,9 +102,11 @@ public void testSharedSubscriptionMessageBacklog() throws PulsarClientException } } + // 6. Check consumer2 backlog log.debug(consumer2.getBrokerConsumerStatsAsync().get().toString()); Assert.assertEquals(consumer2.getBrokerConsumerStatsAsync().get().getMsgBacklog(), totalMessages - count); + // 7. Consume all messages routed to consumer2 while (true) { msg = consumer3.receive(10, TimeUnit.MILLISECONDS); if (msg != null) { @@ -106,6 +117,7 @@ public void testSharedSubscriptionMessageBacklog() throws PulsarClientException } } + // 6. Check consumer3 backlog log.debug(consumer3.getBrokerConsumerStatsAsync().get().toString()); Assert.assertEquals(consumer3.getBrokerConsumerStatsAsync().get().getMsgBacklog(), 0); } catch (Exception ex) { @@ -122,10 +134,12 @@ public void testCachingMechanism() throws PulsarClientException { int totalMessages = 50; String topicNamePostFix = "testCachingMechanism"; + // 1. Create consumer ConsumerConfiguration conf = new ConsumerConfiguration(); conf.setBrokerConsumerStatsCacheTime(3, TimeUnit.SECONDS); Consumer consumer = pulsarClient.subscribe(topicName + topicNamePostFix, "my-subscriber-name", conf); + // 2. Create producer and produce messages Producer producer = pulsarClient.createProducer(topicName + topicNamePostFix, new ProducerConfiguration()); for (int i = 0; i < totalMessages; i++) { String message = "my-message-" + i; @@ -133,12 +147,15 @@ public void testCachingMechanism() throws PulsarClientException { } try { + // 3. Get stats and validate BrokerConsumerStats stats = consumer.getBrokerConsumerStatsAsync().get(); Assert.assertEquals(consumer.getBrokerConsumerStatsAsync().get().getSubscriptionType(), SubscriptionType.Exclusive); Assert.assertEquals(consumer.getBrokerConsumerStatsAsync().get().getMsgBacklog(), totalMessages); Assert.assertEquals(stats, consumer.getBrokerConsumerStatsAsync().get()); Assert.assertTrue(stats.isValid()); + + // 4. Consume all messages Message msg = null; while (true) { msg = consumer.receive(10, TimeUnit.MILLISECONDS); @@ -150,13 +167,13 @@ public void testCachingMechanism() throws PulsarClientException { } Thread.sleep(2 * 1000); - // Cached results returned + // 5. Cached results returned Assert.assertEquals(consumer.getBrokerConsumerStatsAsync().get().getMsgBacklog(), totalMessages); Assert.assertTrue(consumer.getBrokerConsumerStatsAsync().get().isValid()); Assert.assertEquals(stats, consumer.getBrokerConsumerStatsAsync().get()); Assert.assertTrue(stats.isValid()); - // Waiting for cache time to expire + // 6. Waiting for cache time to expire Thread.sleep(2 * 1000); Assert.assertEquals(consumer.getBrokerConsumerStatsAsync().get().getMsgBacklog(), 0); Assert.assertNotEquals(stats, consumer.getBrokerConsumerStatsAsync().get()); @@ -167,4 +184,46 @@ public void testCachingMechanism() throws PulsarClientException { consumer.close(); } } + + @Test + public void testPartitionedTopicsStats() throws PulsarClientException, PulsarAdminException { + int totalMessages = 60; + String topicNamePostFix = "testPartitionedTopicsStats"; + int numberOfPartitions = 3; + String topicName = this.topicName + topicNamePostFix; + // 1. Create partitioned topic + admin.persistentTopics().createPartitionedTopic(topicName, numberOfPartitions); + + // 2. Create consumer + Consumer consumer = pulsarClient.subscribe(topicName, "my-subscription"); + + // 3. Create producer and produce messages + ProducerConfiguration conf = new ProducerConfiguration(); + conf.setMessageRoutingMode(MessageRoutingMode.RoundRobinPartition); + Producer producer = pulsarClient.createProducer(topicName, conf); + for (int i = 0; i < totalMessages; i++) { + String message = "my-message-" + i; + producer.send(message.getBytes()); + } + + // 4. Get stats + try { + BrokerConsumerStats stats = consumer.getBrokerConsumerStatsAsync().get(); + Assert.assertEquals(stats.getSubscriptionType(), SubscriptionType.Exclusive); + Assert.assertEquals(stats.getMsgBacklog(), totalMessages); + + // 5. check if stats are instance of PartitionedBrokerConsumerStatsImpl + Assert.assertTrue(stats instanceof PartitionedBrokerConsumerStatsImpl); + Assert.assertEquals(((PartitionedBrokerConsumerStatsImpl) stats).get(0).getMsgBacklog(), totalMessages / 3); + Assert.assertEquals(((PartitionedBrokerConsumerStatsImpl) stats).get(1).getMsgBacklog(), totalMessages / 3); + Assert.assertEquals(((PartitionedBrokerConsumerStatsImpl) stats).get(2).getMsgBacklog(), totalMessages / 3); + + Thread.sleep(2 * 1000); + Assert.assertTrue(consumer.getBrokerConsumerStatsAsync().get().isValid()); + } catch (Exception ex) { + Assert.fail("Exception:" + ex); + } finally { + consumer.close(); + } + } } diff --git a/pulsar-client/src/main/java/com/yahoo/pulsar/client/api/BrokerConsumerStats.java b/pulsar-client/src/main/java/com/yahoo/pulsar/client/api/BrokerConsumerStats.java new file mode 100644 index 0000000000000..9710a1d2741b5 --- /dev/null +++ b/pulsar-client/src/main/java/com/yahoo/pulsar/client/api/BrokerConsumerStats.java @@ -0,0 +1,84 @@ +/** + * Copyright 2016 Yahoo Inc. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package com.yahoo.pulsar.client.api; + +public interface BrokerConsumerStats { + /* + * @return - Whether the stats are valid. Call the {@link Consumer.getBrokerConsumerStatsAsync()} again if the + * function returns false. + */ + public boolean isValid(); + + /* + * @return - Rate at which the messages are delivered to the consumer. msg/s + */ + public double getMsgRateOut(); + + /* + * @return - Throughput at which the messages are delivered to the consumer. bytes/s + */ + public double getMsgThroughputOut(); + + /* + * @return - Rate at which the messages are redelivered by this consumer. msg/s + */ + public double getMsgRateRedeliver(); + + /* + * @return - Name of the consumer. + */ + public String getConsumerName(); + + /* + * @return - Number of available message permits for the consumer + */ + public long getAvailablePermits(); + + /* + * @return - Number of unacknowledged messages for the consumer + */ + public long getUnackedMessages(); + + /* + * @return - Flag to verify if consumer is blocked due to reaching threshold of unacked messages + */ + public boolean isBlockedConsumerOnUnackedMsgs(); + + /* + * @return - Address of this consumer + */ + public String getAddress(); + + /* + * @return - Timestamp of the connection + */ + public String getConnectedSince(); + + /* + * @return - Whether this subscription is Exclusive or Shared or Failover + */ + public SubscriptionType getSubscriptionType(); + + /* + * @return - Rate at which the messages are delivered to the consumer. msg/s + */ + public double getMsgRateExpired(); + + /* + * @return - Number of messages in the subscription backlog + */ + public long getMsgBacklog(); +} diff --git a/pulsar-client/src/main/java/com/yahoo/pulsar/client/api/Consumer.java b/pulsar-client/src/main/java/com/yahoo/pulsar/client/api/Consumer.java index 5ecb7737d4540..c2a653dcfffa4 100644 --- a/pulsar-client/src/main/java/com/yahoo/pulsar/client/api/Consumer.java +++ b/pulsar-client/src/main/java/com/yahoo/pulsar/client/api/Consumer.java @@ -19,7 +19,7 @@ import java.util.concurrent.CompletableFuture; import java.util.concurrent.TimeUnit; -import com.yahoo.pulsar.client.impl.BrokerConsumerStats; +import com.yahoo.pulsar.client.impl.BrokerConsumerStatsImpl; import com.yahoo.pulsar.client.impl.ConsumerStats; /** @@ -232,12 +232,13 @@ public interface Consumer extends Closeable { * breaks, the messages are redelivered after reconnect. */ void redeliverUnacknowledgedMessages(); - + /** * Gets Consumer Stats from broker. The stats are cached for * {@link ConsumerConfiguration.getBrokerConsumerStatsCacheTimeInMs()} milliseconds. * - * @return a Completeable future for BrokerConsumerStats + * @return A Completeable future for BrokerConsumerStats. Cast the BrokerConsumerStats to + * PartitionedBrokerConsumerStats in order to get Consumer Stats per partition. */ CompletableFuture getBrokerConsumerStatsAsync(); diff --git a/pulsar-client/src/main/java/com/yahoo/pulsar/client/impl/BrokerConsumerStats.java b/pulsar-client/src/main/java/com/yahoo/pulsar/client/impl/BrokerConsumerStatsImpl.java similarity index 94% rename from pulsar-client/src/main/java/com/yahoo/pulsar/client/impl/BrokerConsumerStats.java rename to pulsar-client/src/main/java/com/yahoo/pulsar/client/impl/BrokerConsumerStatsImpl.java index 813533bc3efb9..2fbb95074aad3 100644 --- a/pulsar-client/src/main/java/com/yahoo/pulsar/client/impl/BrokerConsumerStats.java +++ b/pulsar-client/src/main/java/com/yahoo/pulsar/client/impl/BrokerConsumerStatsImpl.java @@ -15,10 +15,11 @@ */ package com.yahoo.pulsar.client.impl; +import com.yahoo.pulsar.client.api.BrokerConsumerStats; import com.yahoo.pulsar.client.api.SubscriptionType; import com.yahoo.pulsar.common.api.proto.PulsarApi.CommandConsumerStatsResponse; -public class BrokerConsumerStats { +public class BrokerConsumerStatsImpl implements BrokerConsumerStats { private final String DELIMITER = ";"; @@ -61,7 +62,7 @@ public class BrokerConsumerStats { /** Number of messages in the subscription backlog */ private long msgBacklog = 0; - public BrokerConsumerStats(CommandConsumerStatsResponse response) { + public BrokerConsumerStatsImpl(CommandConsumerStatsResponse response) { super(); this.validTillInMs = System.currentTimeMillis(); this.msgRateOut = response.getMsgRateOut(); @@ -78,7 +79,7 @@ public BrokerConsumerStats(CommandConsumerStatsResponse response) { this.msgBacklog = response.getMsgBacklog(); } - public BrokerConsumerStats() { + public BrokerConsumerStatsImpl() { } public void setCacheTime(long timeInMs) { @@ -138,7 +139,7 @@ public long getMsgBacklog() { return msgBacklog; } - public synchronized void add(BrokerConsumerStats stats) { + public synchronized void add(BrokerConsumerStatsImpl stats) { this.validTillInMs = (validTillInMs > stats.validTillInMs) ? validTillInMs : stats.validTillInMs; this.msgRateOut += stats.msgRateOut; this.msgThroughputOut += stats.msgThroughputOut; diff --git a/pulsar-client/src/main/java/com/yahoo/pulsar/client/impl/ClientCnx.java b/pulsar-client/src/main/java/com/yahoo/pulsar/client/impl/ClientCnx.java index 295954b099901..60398f4453cb2 100644 --- a/pulsar-client/src/main/java/com/yahoo/pulsar/client/impl/ClientCnx.java +++ b/pulsar-client/src/main/java/com/yahoo/pulsar/client/impl/ClientCnx.java @@ -60,7 +60,7 @@ public class ClientCnx extends PulsarHandler { private final ConcurrentLongHashMap> pendingRequests = new ConcurrentLongHashMap<>(16, 1); private final ConcurrentLongHashMap> pendingLookupRequests = new ConcurrentLongHashMap<>(16, 1); - private final ConcurrentLongHashMap> pendingConsumerStatsRequests = new ConcurrentLongHashMap<>(16, 1); + private final ConcurrentLongHashMap> pendingConsumerStatsRequests = new ConcurrentLongHashMap<>(16, 1); private final ConcurrentLongHashMap producers = new ConcurrentLongHashMap<>(16, 1); private final ConcurrentLongHashMap consumers = new ConcurrentLongHashMap<>(16, 1); @@ -251,14 +251,14 @@ protected void handleConsumerStatsResponse(CommandConsumerStatsResponse response log.debug("Received Consumer Stats response for request id: {}", response.getRequestId()); } long requestId = response.getRequestId(); - CompletableFuture future = pendingConsumerStatsRequests.remove(requestId); + CompletableFuture future = pendingConsumerStatsRequests.remove(requestId); if (future != null) { if (response.hasErrorCode()) { future.completeExceptionally(getPulsarClientException(response.getErrorCode(), response.hasErrorMessage() ? response.getErrorMessage() : null)); } else { - future.complete(new BrokerConsumerStats(response)); + future.complete(new BrokerConsumerStatsImpl(response)); } } else { log.warn("{} Received unknown request id from server: {}", ctx.channel(), requestId); @@ -390,8 +390,8 @@ CompletableFuture newLookup(ByteBuf request, long requestId) { return future; } - CompletableFuture newConsumerStats(String topicName, String subscriptionName, long consumerId, long requestId) { - CompletableFuture future = new CompletableFuture<>(); + CompletableFuture newConsumerStats(String topicName, String subscriptionName, long consumerId, long requestId) { + CompletableFuture future = new CompletableFuture<>(); ByteBuf request = Commands.newConsumerStats(topicName, subscriptionName, consumerId, requestId); ctx.writeAndFlush(request).addListener(writeFuture -> { if (!writeFuture.isSuccess()) { diff --git a/pulsar-client/src/main/java/com/yahoo/pulsar/client/impl/ConsumerImpl.java b/pulsar-client/src/main/java/com/yahoo/pulsar/client/impl/ConsumerImpl.java index 1cce20a3d68d4..fa856c434066c 100644 --- a/pulsar-client/src/main/java/com/yahoo/pulsar/client/impl/ConsumerImpl.java +++ b/pulsar-client/src/main/java/com/yahoo/pulsar/client/impl/ConsumerImpl.java @@ -37,6 +37,7 @@ import org.slf4j.Logger; import org.slf4j.LoggerFactory; +import com.yahoo.pulsar.client.api.BrokerConsumerStats; import com.yahoo.pulsar.client.api.Consumer; import com.yahoo.pulsar.client.api.ConsumerConfiguration; import com.yahoo.pulsar.client.api.Message; @@ -92,7 +93,7 @@ public class ConsumerImpl extends ConsumerBase { private final ConsumerStats stats; private final int priorityLevel; - private volatile BrokerConsumerStats brokerConsumerStats = new BrokerConsumerStats(); + private volatile BrokerConsumerStatsImpl brokerConsumerStats = new BrokerConsumerStatsImpl(); ConsumerImpl(PulsarClientImpl client, String topic, String subscription, ConsumerConfiguration conf, ExecutorService listenerExecutor, CompletableFuture subscribeFuture) { @@ -1023,10 +1024,9 @@ public CompletableFuture getBrokerConsumerStatsAsync() { } if (brokerConsumerStats.isValid()) { - CompletableFuture future = new CompletableFuture(); - future.complete(brokerConsumerStats); - return future; + return CompletableFuture.completedFuture(brokerConsumerStats); } + long requestId = client.newRequestId(); return cnx().newConsumerStats(topic, subscription, consumerId, requestId) .thenApplyAsync(brokerConsumerStats -> { diff --git a/pulsar-client/src/main/java/com/yahoo/pulsar/client/impl/PartitionedBrokerConsumerStatsImpl.java b/pulsar-client/src/main/java/com/yahoo/pulsar/client/impl/PartitionedBrokerConsumerStatsImpl.java new file mode 100644 index 0000000000000..2f453dd7eb9b0 --- /dev/null +++ b/pulsar-client/src/main/java/com/yahoo/pulsar/client/impl/PartitionedBrokerConsumerStatsImpl.java @@ -0,0 +1,118 @@ +/** + * Copyright 2016 Yahoo Inc. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package com.yahoo.pulsar.client.impl; + +import java.util.List; + +import com.google.common.collect.Lists; +import com.yahoo.pulsar.client.api.BrokerConsumerStats; +import com.yahoo.pulsar.client.api.PulsarClientException; +import com.yahoo.pulsar.client.api.PulsarClientException.InvalidConfigurationException; +import com.yahoo.pulsar.client.api.SubscriptionType; + +public class PartitionedBrokerConsumerStatsImpl implements BrokerConsumerStats { + + private String DELIMITER = ";"; + + List brokerConsumerStatsList; + + PartitionedBrokerConsumerStatsImpl() { + brokerConsumerStatsList = Lists.newArrayList(); + } + + public synchronized boolean isValid() { + return brokerConsumerStatsList.stream().reduce(true, (accumulated, stats) -> accumulated & stats.isValid(), + (accumulated1, accumulated2) -> accumulated1 & accumulated2); + } + + public double getMsgRateOut() { + return brokerConsumerStatsList.stream().mapToDouble(stats -> stats.getMsgRateOut()).sum(); + } + + public double getMsgThroughputOut() { + return brokerConsumerStatsList.stream().mapToDouble(stats -> stats.getMsgThroughputOut()).sum(); + } + + public double getMsgRateRedeliver() { + return brokerConsumerStatsList.stream().mapToDouble(stats -> stats.getMsgRateRedeliver()).sum(); + } + + public String getConsumerName() { + return brokerConsumerStatsList.stream().reduce("", + (accumulator, stats) -> stats.getConsumerName() + DELIMITER + accumulator, + (acc1, acc2) -> acc1 + DELIMITER + acc2); + } + + public long getAvailablePermits() { + return brokerConsumerStatsList.stream().mapToLong(stats -> stats.getAvailablePermits()).sum(); + } + + public long getUnackedMessages() { + return brokerConsumerStatsList.stream().mapToLong(stats -> stats.getUnackedMessages()).sum(); + } + + public boolean isBlockedConsumerOnUnackedMsgs() { + return brokerConsumerStatsList.stream().reduce(true, + (accumulated, stats) -> accumulated & stats.isBlockedConsumerOnUnackedMsgs(), + (accumulated1, accumulated2) -> accumulated1 & accumulated2); + } + + public String getAddress() { + return brokerConsumerStatsList.stream().reduce("", + (accumulator, stats) -> stats.getAddress() + DELIMITER + accumulator, + (acc1, acc2) -> acc1 + DELIMITER + acc2); + } + + public String getConnectedSince() { + return brokerConsumerStatsList.stream().reduce("", + (accumulator, stats) -> stats.getConnectedSince() + DELIMITER + accumulator, + (acc1, acc2) -> acc1 + DELIMITER + acc2); + } + + public SubscriptionType getSubscriptionType() { + if (brokerConsumerStatsList.size() > 0) { + return brokerConsumerStatsList.get(0).getSubscriptionType(); + } + return SubscriptionType.Exclusive; + } + + public double getMsgRateExpired() { + return brokerConsumerStatsList.stream().mapToDouble(stats -> stats.getMsgRateExpired()).sum(); + } + + public long getMsgBacklog() { + return brokerConsumerStatsList.stream().mapToLong(stats -> stats.getMsgBacklog()).sum(); + } + + public synchronized void add(BrokerConsumerStats stats) { + brokerConsumerStatsList.add(stats); + } + + public synchronized BrokerConsumerStats get(int partitionIndex) throws InvalidConfigurationException { + int size = brokerConsumerStatsList.size(); + if (partitionIndex < 0 || partitionIndex >= size) { + throw new PulsarClientException.InvalidConfigurationException("partitionIndex [" + partitionIndex + + "] needs to be positive and less than brokerConsumerStatsList.size() [" + size + "]"); + } + return brokerConsumerStatsList.get(partitionIndex); + } + + @Override + public String toString() { + return "PartitionedBrokerConsumerStatsImpl [" + brokerConsumerStatsList.stream().reduce("", + (acc, stats) -> stats + "," + acc, (acc1, acc2) -> acc1 + "," + acc2) + "]"; + } +} diff --git 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 index 43b3629753fa0..5cdcdcfb94bf1 100644 --- 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 @@ -34,6 +34,7 @@ import org.slf4j.LoggerFactory; import com.google.common.collect.Lists; +import com.yahoo.pulsar.client.api.BrokerConsumerStats; import com.yahoo.pulsar.client.api.Consumer; import com.yahoo.pulsar.client.api.ConsumerConfiguration; import com.yahoo.pulsar.client.api.Message; @@ -129,7 +130,7 @@ private void receiveMessageFromConsumer(ConsumerImpl consumer) { if (incomingMessages.size() >= maxReceiverQueueSize || (incomingMessages.size() > sharedQueueResumeThreshold && !pausedConsumers.isEmpty())) { - // mark this consumer to be resumed later: if No more space left in shared queue, + // mark this consumer to be resumed later: if No more space left in shared queue, // or if any consumer is already paused (to create fair chance for already paused consumers) pausedConsumers.add(consumer); } else { @@ -329,7 +330,8 @@ void messageReceived(Message message) { lock.readLock().lock(); try { if (log.isDebugEnabled()) { - log.debug("[{}][{}] Received message from partitioned-consumer {}", topic, subscription, message.getMessageId()); + log.debug("[{}][{}] Received message from partitioned-consumer {}", topic, subscription, + message.getMessageId()); } // if asyncReceive is waiting : return message to callback without adding to incomingMessages queue if (!pendingReceives.isEmpty()) { @@ -360,7 +362,8 @@ void messageReceived(Message message) { try { if (log.isDebugEnabled()) { - log.debug("[{}][{}] Calling message listener for message {}", topic, subscription, message.getMessageId()); + log.debug("[{}][{}] Calling message listener for message {}", topic, subscription, + message.getMessageId()); } listener.received(PartitionedConsumerImpl.this, msg); } catch (Throwable t) { @@ -453,11 +456,15 @@ public synchronized ConsumerStats getStats() { @Override public CompletableFuture getBrokerConsumerStatsAsync() { - BrokerConsumerStats brokerConsumerStats = new BrokerConsumerStats(); + PartitionedBrokerConsumerStatsImpl brokerConsumerStats = new PartitionedBrokerConsumerStatsImpl(); List> futures = Lists.newArrayList(); for (Consumer c : consumers) { - futures.add(c.getBrokerConsumerStatsAsync().thenAcceptAsync(stats -> {brokerConsumerStats.add(stats);})); + futures.add(c.getBrokerConsumerStatsAsync().thenAccept(stats -> { + brokerConsumerStats.add(stats); + })); } - return FutureUtil.waitForAll(futures).thenApplyAsync( r -> {return brokerConsumerStats;}); + return FutureUtil.waitForAll(futures).thenApply(r -> { + return brokerConsumerStats; + }); } } From 4ce6a443bfec543974aa755d23a79ba526c4526a Mon Sep 17 00:00:00 2001 From: Jai ASher Date: Mon, 20 Mar 2017 18:55:42 -0700 Subject: [PATCH 3/6] Some exception handling --- .../PartitionedBrokerConsumerStatsImpl.java | 16 ++++++++++++---- .../client/impl/PartitionedConsumerImpl.java | 19 ++++++++++++++----- 2 files changed, 26 insertions(+), 9 deletions(-) diff --git a/pulsar-client/src/main/java/com/yahoo/pulsar/client/impl/PartitionedBrokerConsumerStatsImpl.java b/pulsar-client/src/main/java/com/yahoo/pulsar/client/impl/PartitionedBrokerConsumerStatsImpl.java index 2f453dd7eb9b0..24df757698894 100644 --- a/pulsar-client/src/main/java/com/yahoo/pulsar/client/impl/PartitionedBrokerConsumerStatsImpl.java +++ b/pulsar-client/src/main/java/com/yahoo/pulsar/client/impl/PartitionedBrokerConsumerStatsImpl.java @@ -29,8 +29,8 @@ public class PartitionedBrokerConsumerStatsImpl implements BrokerConsumerStats { List brokerConsumerStatsList; - PartitionedBrokerConsumerStatsImpl() { - brokerConsumerStatsList = Lists.newArrayList(); + PartitionedBrokerConsumerStatsImpl(int initialArraySize) { + brokerConsumerStatsList = Lists.newArrayListWithCapacity(initialArraySize); } public synchronized boolean isValid() { @@ -65,6 +65,10 @@ public long getUnackedMessages() { } public boolean isBlockedConsumerOnUnackedMsgs() { + if (brokerConsumerStatsList.size() == 0) { + return false; + } + return brokerConsumerStatsList.stream().reduce(true, (accumulated, stats) -> accumulated & stats.isBlockedConsumerOnUnackedMsgs(), (accumulated1, accumulated2) -> accumulated1 & accumulated2); @@ -97,8 +101,12 @@ public long getMsgBacklog() { return brokerConsumerStatsList.stream().mapToLong(stats -> stats.getMsgBacklog()).sum(); } - public synchronized void add(BrokerConsumerStats stats) { - brokerConsumerStatsList.add(stats); + public synchronized void add(int index, BrokerConsumerStats stats) { + brokerConsumerStatsList.add(index, stats); + } + + public synchronized void clear() { + brokerConsumerStatsList.clear(); } public synchronized BrokerConsumerStats get(int partitionIndex) throws InvalidConfigurationException { diff --git 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 index 5cdcdcfb94bf1..1344342dda77a 100644 --- 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 @@ -456,15 +456,24 @@ public synchronized ConsumerStats getStats() { @Override public CompletableFuture getBrokerConsumerStatsAsync() { - PartitionedBrokerConsumerStatsImpl brokerConsumerStats = new PartitionedBrokerConsumerStatsImpl(); + PartitionedBrokerConsumerStatsImpl brokerConsumerStats = new PartitionedBrokerConsumerStatsImpl(consumers.size()); List> futures = Lists.newArrayList(); - for (Consumer c : consumers) { - futures.add(c.getBrokerConsumerStatsAsync().thenAccept(stats -> { - brokerConsumerStats.add(stats); + for (int i = 0; i < consumers.size(); i++) { + final int index = i; // need effectively final variable to use in thenAccept(...) + futures.add(consumers.get(index).getBrokerConsumerStatsAsync() + .thenAccept(stats -> { + brokerConsumerStats.add(index, stats); })); } - return FutureUtil.waitForAll(futures).thenApply(r -> { + + CompletableFuture future = FutureUtil.waitForAll(futures).thenApply(r -> { return brokerConsumerStats; }); + + if (future.isCompletedExceptionally()) { + brokerConsumerStats.clear(); + } + + return future; } } From 9bddb6681b0f0030e1684a8c660b0417b86e8a34 Mon Sep 17 00:00:00 2001 From: Jai ASher Date: Tue, 21 Mar 2017 21:56:33 -0700 Subject: [PATCH 4/6] Added check to see if the Partitioned topic is ready --- .../yahoo/pulsar/client/impl/PartitionedConsumerImpl.java | 7 ++++--- 1 file changed, 4 insertions(+), 3 deletions(-) diff --git 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 index 1344342dda77a..4c304a9599c7f 100644 --- 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 @@ -17,7 +17,6 @@ import static com.google.common.base.Preconditions.checkArgument; -import java.util.ArrayList; import java.util.HashSet; import java.util.List; import java.util.Set; @@ -266,7 +265,6 @@ public CompletableFuture unsubscribeAsync() { @Override public CompletableFuture closeAsync() { - if (getState() == State.Closing || getState() == State.Closed) { return CompletableFuture.completedFuture(null); } @@ -455,7 +453,10 @@ public synchronized ConsumerStats getStats() { private static final Logger log = LoggerFactory.getLogger(PartitionedConsumerImpl.class); @Override - public CompletableFuture getBrokerConsumerStatsAsync() { + public synchronized CompletableFuture getBrokerConsumerStatsAsync() { + if (getState() != State.Ready) { + return FutureUtil.failedFuture(new PulsarClientException.NotConnectedException()); + } PartitionedBrokerConsumerStatsImpl brokerConsumerStats = new PartitionedBrokerConsumerStatsImpl(consumers.size()); List> futures = Lists.newArrayList(); for (int i = 0; i < consumers.size(); i++) { From 8fd385e5ee22bc63a8dc823952a212ed35d06000 Mon Sep 17 00:00:00 2001 From: Jai ASher Date: Tue, 28 Mar 2017 18:19:03 -0700 Subject: [PATCH 5/6] Corrected handling of futures --- .../client/api/BrokerConsumerStatsTest.java | 35 +++++++------------ .../client/impl/PartitionedConsumerImpl.java | 16 ++++----- 2 files changed, 20 insertions(+), 31 deletions(-) diff --git a/pulsar-broker/src/test/java/com/yahoo/pulsar/client/api/BrokerConsumerStatsTest.java b/pulsar-broker/src/test/java/com/yahoo/pulsar/client/api/BrokerConsumerStatsTest.java index fc9677fda3b82..81e3737c830b3 100644 --- a/pulsar-broker/src/test/java/com/yahoo/pulsar/client/api/BrokerConsumerStatsTest.java +++ b/pulsar-broker/src/test/java/com/yahoo/pulsar/client/api/BrokerConsumerStatsTest.java @@ -75,15 +75,12 @@ public void testSharedSubscriptionMessageBacklog() throws PulsarClientException try { // 3. Consumer all messages routed to consumer1 Message msg = null; - while (true) { - msg = consumer1.receive(10, TimeUnit.MILLISECONDS); + do { + msg = consumer1.receive(1, TimeUnit.SECONDS); if (msg != null) { consumer1.acknowledge(msg); - count++; - } else { - break; } - } + } while(msg != null); // 4. Check subscription type log.debug(consumer1.getBrokerConsumerStatsAsync().get().toString()); @@ -92,30 +89,24 @@ public void testSharedSubscriptionMessageBacklog() throws PulsarClientException Assert.assertEquals(consumer1.getBrokerConsumerStatsAsync().get().getMsgBacklog(), totalMessages - count); // 5. Consumer all messages routed to consumer2 - while (true) { - msg = consumer2.receive(10, TimeUnit.MILLISECONDS); + do { + msg = consumer2.receive(1, TimeUnit.SECONDS); if (msg != null) { consumer2.acknowledge(msg); - count++; - } else { - break; } - } + } while(msg != null); // 6. Check consumer2 backlog log.debug(consumer2.getBrokerConsumerStatsAsync().get().toString()); Assert.assertEquals(consumer2.getBrokerConsumerStatsAsync().get().getMsgBacklog(), totalMessages - count); // 7. Consume all messages routed to consumer2 - while (true) { - msg = consumer3.receive(10, TimeUnit.MILLISECONDS); + do { + msg = consumer3.receive(1, TimeUnit.SECONDS); if (msg != null) { consumer3.acknowledge(msg); - count++; - } else { - break; } - } + } while(msg != null); // 6. Check consumer3 backlog log.debug(consumer3.getBrokerConsumerStatsAsync().get().toString()); @@ -157,14 +148,12 @@ public void testCachingMechanism() throws PulsarClientException { // 4. Consume all messages Message msg = null; - while (true) { - msg = consumer.receive(10, TimeUnit.MILLISECONDS); + do { + msg = consumer.receive(1, TimeUnit.SECONDS); if (msg != null) { consumer.acknowledge(msg); - } else { - break; } - } + } while(msg != null); Thread.sleep(2 * 1000); // 5. Cached results returned diff --git 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 index 4c304a9599c7f..2ecd564b7f3cf 100644 --- 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 @@ -466,15 +466,15 @@ public synchronized CompletableFuture getBrokerConsumerStat brokerConsumerStats.add(index, stats); })); } - - CompletableFuture future = FutureUtil.waitForAll(futures).thenApply(r -> { - return brokerConsumerStats; - }); - - if (future.isCompletedExceptionally()) { + final CompletableFuture future = new CompletableFuture(); + FutureUtil.waitForAll(futures).thenApply(r -> { + future.complete(brokerConsumerStats); + return (BrokerConsumerStats) brokerConsumerStats; + }).exceptionally(ex -> { brokerConsumerStats.clear(); - } - + future.completeExceptionally(ex); + return null; + }); return future; } } From de29999437c4f3d43da840cb1739cdbf278e2d5d Mon Sep 17 00:00:00 2001 From: Jai ASher Date: Wed, 29 Mar 2017 00:38:45 -0700 Subject: [PATCH 6/6] Fixed some test cases --- .../pulsar/client/api/BrokerConsumerStatsTest.java | 13 +++++++------ .../com/yahoo/pulsar/client/impl/ConsumerImpl.java | 2 +- 2 files changed, 8 insertions(+), 7 deletions(-) diff --git a/pulsar-broker/src/test/java/com/yahoo/pulsar/client/api/BrokerConsumerStatsTest.java b/pulsar-broker/src/test/java/com/yahoo/pulsar/client/api/BrokerConsumerStatsTest.java index 81e3737c830b3..d5fb66c236833 100644 --- a/pulsar-broker/src/test/java/com/yahoo/pulsar/client/api/BrokerConsumerStatsTest.java +++ b/pulsar-broker/src/test/java/com/yahoo/pulsar/client/api/BrokerConsumerStatsTest.java @@ -79,6 +79,7 @@ public void testSharedSubscriptionMessageBacklog() throws PulsarClientException msg = consumer1.receive(1, TimeUnit.SECONDS); if (msg != null) { consumer1.acknowledge(msg); + count++; } } while(msg != null); @@ -93,6 +94,7 @@ public void testSharedSubscriptionMessageBacklog() throws PulsarClientException msg = consumer2.receive(1, TimeUnit.SECONDS); if (msg != null) { consumer2.acknowledge(msg); + count++; } } while(msg != null); @@ -105,6 +107,7 @@ public void testSharedSubscriptionMessageBacklog() throws PulsarClientException msg = consumer3.receive(1, TimeUnit.SECONDS); if (msg != null) { consumer3.acknowledge(msg); + count++; } } while(msg != null); @@ -127,7 +130,7 @@ public void testCachingMechanism() throws PulsarClientException { // 1. Create consumer ConsumerConfiguration conf = new ConsumerConfiguration(); - conf.setBrokerConsumerStatsCacheTime(3, TimeUnit.SECONDS); + conf.setBrokerConsumerStatsCacheTime(6, TimeUnit.SECONDS); Consumer consumer = pulsarClient.subscribe(topicName + topicNamePostFix, "my-subscriber-name", conf); // 2. Create producer and produce messages @@ -155,7 +158,6 @@ public void testCachingMechanism() throws PulsarClientException { } } while(msg != null); - Thread.sleep(2 * 1000); // 5. Cached results returned Assert.assertEquals(consumer.getBrokerConsumerStatsAsync().get().getMsgBacklog(), totalMessages); Assert.assertTrue(consumer.getBrokerConsumerStatsAsync().get().isValid()); @@ -163,7 +165,7 @@ public void testCachingMechanism() throws PulsarClientException { Assert.assertTrue(stats.isValid()); // 6. Waiting for cache time to expire - Thread.sleep(2 * 1000); + Thread.sleep(8 * 1000); Assert.assertEquals(consumer.getBrokerConsumerStatsAsync().get().getMsgBacklog(), 0); Assert.assertNotEquals(stats, consumer.getBrokerConsumerStatsAsync().get()); Assert.assertFalse(stats.isValid()); @@ -200,15 +202,14 @@ public void testPartitionedTopicsStats() throws PulsarClientException, PulsarAdm BrokerConsumerStats stats = consumer.getBrokerConsumerStatsAsync().get(); Assert.assertEquals(stats.getSubscriptionType(), SubscriptionType.Exclusive); Assert.assertEquals(stats.getMsgBacklog(), totalMessages); + Thread.sleep(4 * 1000); + Assert.assertTrue(consumer.getBrokerConsumerStatsAsync().get().isValid()); // 5. check if stats are instance of PartitionedBrokerConsumerStatsImpl Assert.assertTrue(stats instanceof PartitionedBrokerConsumerStatsImpl); Assert.assertEquals(((PartitionedBrokerConsumerStatsImpl) stats).get(0).getMsgBacklog(), totalMessages / 3); Assert.assertEquals(((PartitionedBrokerConsumerStatsImpl) stats).get(1).getMsgBacklog(), totalMessages / 3); Assert.assertEquals(((PartitionedBrokerConsumerStatsImpl) stats).get(2).getMsgBacklog(), totalMessages / 3); - - Thread.sleep(2 * 1000); - Assert.assertTrue(consumer.getBrokerConsumerStatsAsync().get().isValid()); } catch (Exception ex) { Assert.fail("Exception:" + ex); } finally { diff --git a/pulsar-client/src/main/java/com/yahoo/pulsar/client/impl/ConsumerImpl.java b/pulsar-client/src/main/java/com/yahoo/pulsar/client/impl/ConsumerImpl.java index fa856c434066c..b64055e971372 100644 --- a/pulsar-client/src/main/java/com/yahoo/pulsar/client/impl/ConsumerImpl.java +++ b/pulsar-client/src/main/java/com/yahoo/pulsar/client/impl/ConsumerImpl.java @@ -1029,7 +1029,7 @@ public CompletableFuture getBrokerConsumerStatsAsync() { long requestId = client.newRequestId(); return cnx().newConsumerStats(topic, subscription, consumerId, requestId) - .thenApplyAsync(brokerConsumerStats -> { + .thenApply(brokerConsumerStats -> { brokerConsumerStats.setCacheTime(conf.getBrokerConsumerStatsCacheTimeInMs()); this.brokerConsumerStats = brokerConsumerStats; return brokerConsumerStats;