jai1 closed pull request #254: Java Client - Support for getting consumer stats from broker URL: https://github.com/apache/incubator-pulsar/pull/254
This is a PR merged from a forked repository. As GitHub hides the original diff on merge, it is displayed below for the sake of provenance: As this is a foreign pull request (from a fork), the diff is supplied below (as it won't show otherwise due to GitHub magic): 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 new file mode 100644 index 0000000000..d5fb66c236 --- /dev/null +++ b/pulsar-broker/src/test/java/com/yahoo/pulsar/client/api/BrokerConsumerStatsTest.java @@ -0,0 +1,219 @@ +/** + * 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; + +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.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); + 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"; + + // 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; + producer.send(message.getBytes()); + } + + int count = 0; + try { + // 3. Consumer all messages routed to consumer1 + Message msg = null; + do { + msg = consumer1.receive(1, TimeUnit.SECONDS); + if (msg != null) { + consumer1.acknowledge(msg); + count++; + } + } while(msg != null); + + // 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 + do { + msg = consumer2.receive(1, TimeUnit.SECONDS); + if (msg != null) { + consumer2.acknowledge(msg); + count++; + } + } 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 + do { + msg = consumer3.receive(1, TimeUnit.SECONDS); + if (msg != null) { + consumer3.acknowledge(msg); + count++; + } + } while(msg != null); + + // 6. Check consumer3 backlog + 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"; + + // 1. Create consumer + ConsumerConfiguration conf = new ConsumerConfiguration(); + conf.setBrokerConsumerStatsCacheTime(6, 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; + producer.send(message.getBytes()); + } + + 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; + do { + msg = consumer.receive(1, TimeUnit.SECONDS); + if (msg != null) { + consumer.acknowledge(msg); + } + } while(msg != null); + + // 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()); + + // 6. Waiting for cache time to expire + Thread.sleep(8 * 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(); + } + } + + @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); + 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); + } 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 0000000000..9710a1d274 --- /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 68edcaaa8d..c2a653dcff 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.BrokerConsumerStatsImpl; import com.yahoo.pulsar.client.impl.ConsumerStats; /** @@ -231,4 +232,14 @@ * 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. Cast the BrokerConsumerStats to + * PartitionedBrokerConsumerStats in order to get Consumer Stats per partition. + */ + CompletableFuture<BrokerConsumerStats> 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 b7cf0bde45..9c6c7a3b61 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 @@ 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 eb0918b4ee..93e7e98d55 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/BrokerConsumerStatsImpl.java b/pulsar-client/src/main/java/com/yahoo/pulsar/client/impl/BrokerConsumerStatsImpl.java new file mode 100644 index 0000000000..2fbb95074a --- /dev/null +++ b/pulsar-client/src/main/java/com/yahoo/pulsar/client/impl/BrokerConsumerStatsImpl.java @@ -0,0 +1,169 @@ +/** + * 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.BrokerConsumerStats; +import com.yahoo.pulsar.client.api.SubscriptionType; +import com.yahoo.pulsar.common.api.proto.PulsarApi.CommandConsumerStatsResponse; + +public class BrokerConsumerStatsImpl implements 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 BrokerConsumerStatsImpl(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 BrokerConsumerStatsImpl() { + } + + 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(BrokerConsumerStatsImpl 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 a498cc0adc..60398f4453 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 @@ -34,6 +34,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; @@ -58,6 +60,7 @@ private final ConcurrentLongHashMap<CompletableFuture<String>> pendingRequests = new ConcurrentLongHashMap<>(16, 1); private final ConcurrentLongHashMap<CompletableFuture<LookupDataResult>> pendingLookupRequests = new ConcurrentLongHashMap<>(16, 1); + private final ConcurrentLongHashMap<CompletableFuture<BrokerConsumerStatsImpl>> pendingConsumerStatsRequests = new ConcurrentLongHashMap<>(16, 1); private final ConcurrentLongHashMap<ProducerImpl> producers = new ConcurrentLongHashMap<>(16, 1); private final ConcurrentLongHashMap<ConsumerImpl> consumers = new ConcurrentLongHashMap<>(16, 1); @@ -119,6 +122,8 @@ public void channelInactive(ChannelHandlerContext ctx) throws Exception { // Fail out all the pending ops pendingRequests.forEach((key, future) -> future.completeExceptionally(e)); + pendingConsumerStatsRequests.forEach((key, future) -> future.completeExceptionally(e)); + pendingConsumerStatsRequests.clear(); pendingLookupRequests.forEach((key, future) -> future.completeExceptionally(e)); // Notify all attached producers/consumers so they have a chance to reconnect @@ -239,6 +244,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<BrokerConsumerStatsImpl> future = pendingConsumerStatsRequests.remove(requestId); + + if (future != null) { + if (response.hasErrorCode()) { + future.completeExceptionally(getPulsarClientException(response.getErrorCode(), + response.hasErrorMessage() ? response.getErrorMessage() : null)); + } else { + future.complete(new BrokerConsumerStatsImpl(response)); + } + } else { + log.warn("{} Received unknown request id from server: {}", ctx.channel(), requestId); + } + } @Override protected void handlePartitionResponse(CommandPartitionedTopicMetadataResponse lookupResult) { @@ -352,6 +377,7 @@ protected boolean isHandshakeCompleted() { writeFuture.cause().getMessage()); getAndRemovePendingLookupRequest(requestId); future.completeExceptionally(writeFuture.cause()); + getAndRemovePendingLookupRequest(requestId); } }); } else { @@ -363,7 +389,22 @@ protected boolean isHandshakeCompleted() { } return future; } - + + CompletableFuture<BrokerConsumerStatsImpl> newConsumerStats(String topicName, String subscriptionName, long consumerId, long requestId) { + CompletableFuture<BrokerConsumerStatsImpl> 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<Void> newPromise() { return ctx.newPromise(); } @@ -463,9 +504,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 050dada7dd..47c84988a2 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<MessageIdImpl> messageIds); + + public abstract CompletableFuture<BrokerConsumerStats> 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 fb160131d6..b64055e971 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,6 +93,7 @@ private final ConsumerStats stats; private final int priorityLevel; + private volatile BrokerConsumerStatsImpl brokerConsumerStats = new BrokerConsumerStatsImpl(); ConsumerImpl(PulsarClientImpl client, String topic, String subscription, ConsumerConfiguration conf, ExecutorService listenerExecutor, CompletableFuture<Consumer> subscribeFuture) { @@ -1015,4 +1017,23 @@ public ConsumerStats getStats() { private static final Logger log = LoggerFactory.getLogger(ConsumerImpl.class); + @Override + public CompletableFuture<BrokerConsumerStats> getBrokerConsumerStatsAsync() { + if (getState() != State.Ready || !isConnected()) { + return FutureUtil.failedFuture(new PulsarClientException.NotConnectedException()); + } + + if (brokerConsumerStats.isValid()) { + return CompletableFuture.completedFuture(brokerConsumerStats); + } + + long requestId = client.newRequestId(); + return cnx().newConsumerStats(topic, subscription, consumerId, requestId) + .thenApply(brokerConsumerStats -> { + brokerConsumerStats.setCacheTime(conf.getBrokerConsumerStatsCacheTimeInMs()); + this.brokerConsumerStats = brokerConsumerStats; + return 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 0000000000..24df757698 --- /dev/null +++ b/pulsar-client/src/main/java/com/yahoo/pulsar/client/impl/PartitionedBrokerConsumerStatsImpl.java @@ -0,0 +1,126 @@ +/** + * 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<BrokerConsumerStats> brokerConsumerStatsList; + + PartitionedBrokerConsumerStatsImpl(int initialArraySize) { + brokerConsumerStatsList = Lists.newArrayListWithCapacity(initialArraySize); + } + + 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() { + if (brokerConsumerStatsList.size() == 0) { + return false; + } + + 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(int index, BrokerConsumerStats stats) { + brokerConsumerStatsList.add(index, stats); + } + + public synchronized void clear() { + brokerConsumerStatsList.clear(); + } + + 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 5ce7ba662d..2ecd564b7f 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 @@ -33,6 +33,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; @@ -128,7 +129,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 { @@ -264,7 +265,6 @@ protected Message internalReceive(int timeout, TimeUnit unit) throws PulsarClien @Override public CompletableFuture<Void> closeAsync() { - if (getState() == State.Closing || getState() == State.Closed) { return CompletableFuture.completedFuture(null); } @@ -328,7 +328,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()) { @@ -359,7 +360,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) { @@ -449,4 +451,30 @@ public synchronized ConsumerStats getStats() { } private static final Logger log = LoggerFactory.getLogger(PartitionedConsumerImpl.class); + + @Override + public synchronized CompletableFuture<BrokerConsumerStats> getBrokerConsumerStatsAsync() { + if (getState() != State.Ready) { + return FutureUtil.failedFuture(new PulsarClientException.NotConnectedException()); + } + PartitionedBrokerConsumerStatsImpl brokerConsumerStats = new PartitionedBrokerConsumerStatsImpl(consumers.size()); + List<CompletableFuture<Void>> futures = Lists.newArrayList(); + 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); + })); + } + final CompletableFuture<BrokerConsumerStats> future = new CompletableFuture<BrokerConsumerStats>(); + FutureUtil.waitForAll(futures).thenApply(r -> { + future.complete(brokerConsumerStats); + return (BrokerConsumerStats) brokerConsumerStats; + }).exceptionally(ex -> { + brokerConsumerStats.clear(); + future.completeExceptionally(ex); + return null; + }); + return future; + } } 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 6ab4a01003..375388d254 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; @@ -531,6 +531,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; ---------------------------------------------------------------- This is an automated message from the Apache Git Service. To respond to the message, please log on GitHub and use the URL above to go to the specific comment. For queries about this service, please contact Infrastructure at: [email protected] With regards, Apache Git Services
