Java Client - Support for getting consumer stats from broker #254

jai1 · 2017-02-28T23:12:21Z

Motivation

Java client for Binary Protocol for getting topic information #194

Modifications

A new flow added to Java Client

Result

We can get broker side consumer stats using java client.

merlimat · 2017-03-01T00:59:38Z

pulsar-client-cpp/include/pulsar/Consumer.h

@@ -85,10 +85,10 @@ class BrokerConsumerStats {
    std::string consumerName_;

    /** Number of available message permits for the consumer */
-    int availablePermits_;
+    unsigned long long availablePermits_;


use uint64_t

merlimat · 2017-03-01T00:59:53Z

pulsar-client-cpp/include/pulsar/Consumer.h

-    /// Number of messages in the subscription backlog
-    long msgBacklog_;
+    /** Number of messages in the subscription backlog */
+    unsigned long long msgBacklog_;


merlimat · 2017-03-01T03:02:05Z

@jai1 can you create individual PRs for the unrelated changes?

jai1 · 2017-03-01T06:05:52Z

@merlimat - Opened separate PRs #255 #256 #257 #254

merlimat · 2017-03-01T22:16:55Z

pulsar-client/src/main/java/com/yahoo/pulsar/client/api/Consumer.java

+     *            topic pass only one PartitionIndex in a call.
+     * @return a Completeable future for BrokerConsumerStats
+     */
+    CompletableFuture<BrokerConsumerStats> getBrokerConsumerStatsAsync(int... partitionIndex);


The whole client API is based on abstracting the partitioning layer. This would be mixing the partition across different layers.

@merlimat - does this new design look good to you?

3b76835

saandrews · 2017-03-03T00:01:33Z

pulsar-broker/src/test/java/com/yahoo/pulsar/client/impl/BrokerConsumerStatsTest.java

+                }
+            }
+
+            Thread.sleep(20 * 1000);


Can you reduce the default cache time, so test does not have to wait 20 secs.

rdhabalia · 2017-03-03T23:54:10Z

pulsar-broker/src/test/java/com/yahoo/pulsar/client/impl/BrokerConsumerStatsTest.java

+
+public class BrokerConsumerStatsTest extends BrokerTestBase {
+    private static final Logger log = LoggerFactory.getLogger(BrokerConsumerStatsTest.class);
+    private final int totalMessages = 50;


can we move it to the test-method variable

rdhabalia · 2017-03-03T23:54:19Z

pulsar-broker/src/test/java/com/yahoo/pulsar/client/impl/BrokerConsumerStatsTest.java

+public class BrokerConsumerStatsTest extends BrokerTestBase {
+    private static final Logger log = LoggerFactory.getLogger(BrokerConsumerStatsTest.class);
+    private final int totalMessages = 50;
+    private String topicName = "persistent://prop/cluster/ns/topic-";


same here, test-method variable.

rdhabalia · 2017-03-03T23:57:34Z

pulsar-broker/src/test/java/com/yahoo/pulsar/client/impl/BrokerConsumerStatsTest.java

+
+            log.debug(consumer1.getBrokerConsumerStatsAsync().get().toString());
+            Assert.assertEquals(consumer1.getBrokerConsumerStatsAsync().get().getMsgBacklog(), totalMessages - count);
+


here, Consumer's default availablePermit is 1000. So, broker can dispatch all 50 messages at this point to consumer so, isn't backlog become 0 in that case?

Since it's shared subscription we send 20 messages per consumer in round robin fashion

rdhabalia · 2017-03-03T23:59:58Z

pulsar-client/src/main/java/com/yahoo/pulsar/client/api/Consumer.java

+     * used to check if the stats are still valid.
+     * 
+     * @return a Completeable future for BrokerConsumerStats
+     */


instead can we just say that: stats get refreshed at every X seconds?

rdhabalia · 2017-03-04T00:01:34Z

pulsar-client/src/main/java/com/yahoo/pulsar/client/api/PulsarClientException.java

+        }
+    }
+
+    public static class UnknownError extends PulsarClientException {


I think for unknownError we just throw PulsarClientException. So, we may not need unknownError

As per convention:-
Every error in ServerError is mapped to a unique exception in Java and a unique Result in C++ via clientCnx.getPulsarClientException() (Java) and clientConnection.getResult(serverError)

Since UnknownError is a ServerError type I mapped it to a separate ClientException

rdhabalia · 2017-03-04T00:56:52Z

pulsar-client/src/main/java/com/yahoo/pulsar/client/impl/ClientCnx.java

+        if (future != null) {
+            if (response.hasErrorCode()) {
+                future.completeExceptionally(getPulsarClientException(response.getErrorCode(),
+                        response.hasErrorMessage() ? response.getErrorMessage() : ""));


should we pass null as errorMsg if not present?

rdhabalia · 2017-03-04T00:57:39Z

pulsar-client/src/main/java/com/yahoo/pulsar/client/impl/ClientCnx.java

-
+
+    CompletableFuture<BrokerConsumerStats> newConsumerStats(String topicName, String subscriptionName, long consumerId, long requestId) {
+        CompletableFuture<BrokerConsumerStats> future = new CompletableFuture<BrokerConsumerStats>();


new CompletableFuture<>()

rdhabalia · 2017-03-04T01:03:49Z

pulsar-client/src/main/java/com/yahoo/pulsar/client/impl/ConsumerImpl.java

@@ -92,6 +92,7 @@

    private final ConsumerStats stats;
    private final int priorityLevel;
+    private volatile BrokerConsumerStats brokerConsumerStats = new BrokerConsumerStats();


should we need volatile for stats?

Yes, don't want us to change the brokerStats while user is reading it.

yes, I see we update brokerConsumerStats object.

rdhabalia · 2017-03-04T01:04:47Z

pulsar-client/src/main/java/com/yahoo/pulsar/client/impl/ConsumerImpl.java

+    @Override
+    public CompletableFuture<BrokerConsumerStats> getBrokerConsumerStatsAsync() {
+        if (getState() != State.Ready || !isConnected()) {
+            return FutureUtil.failedFuture(new PulsarClientException.NotConnectedException());


PulsarClientException.NotConnectedException(ERROR_MSG) ?

PulsarClientException.NotConnectedException(ERROR_MSG) no such constructor exists
I can add this constructor but I don't think the ERROR_MSG will be much different from the default message.

rdhabalia · 2017-03-04T01:10:14Z

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

+    @Override
+    public CompletableFuture<BrokerConsumerStats> getBrokerConsumerStatsAsync() {
+        BrokerConsumerStats brokerConsumerStats = new BrokerConsumerStats();
+        List<CompletableFuture<Void>> futures = new ArrayList<CompletableFuture<Void>>();


Lists.newArrayList()??

jai1 · 2017-03-07T00:04:14Z

@rdhabalia @merlimat - kindly review this.

jai1 · 2017-03-14T16:24:43Z

@merlimat - Please review this when you get time

…rStatsJavaClient

rdhabalia · 2017-03-15T19:16:16Z

pulsar-client/src/main/java/com/yahoo/pulsar/client/impl/ConsumerImpl.java

+        if (brokerConsumerStats.isValid()) {
+            CompletableFuture<BrokerConsumerStats> future = new CompletableFuture<BrokerConsumerStats>();
+            future.complete(brokerConsumerStats);
+            return future;


return CompletableFuture.completedFuture(brokerConsumerStats) ?

rdhabalia · 2017-03-15T21:15:03Z

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

+        BrokerConsumerStats brokerConsumerStats = new BrokerConsumerStats();
+        List<CompletableFuture<Void>> futures = Lists.newArrayList();
+        for (Consumer c : consumers) {
+            futures.add(c.getBrokerConsumerStatsAsync().thenAcceptAsync(stats -> {brokerConsumerStats.add(stats);}));


not sure if we want to delegate to another thread using thenApplyAsync has any advantage here over continue with the same thread using thenApply.

rdhabalia · 2017-03-15T23:52:30Z

pulsar-client/src/main/java/com/yahoo/pulsar/client/impl/ConsumerImpl.java

@@ -92,6 +92,7 @@

    private final ConsumerStats stats;
    private final int priorityLevel;
+    private volatile BrokerConsumerStats brokerConsumerStats = new BrokerConsumerStats();


yes, I see we update brokerConsumerStats object.

rdhabalia · 2017-03-16T00:04:13Z

pulsar-client/src/main/java/com/yahoo/pulsar/client/impl/BrokerConsumerStats.java

+        return msgBacklog;
+    }
+
+    public synchronized void add(BrokerConsumerStats stats) {


just a thought here for Partitioned consumers: (here we have add(..) for to aggregate partitioned-consumers result. But only problem is we can't get values for individual consumer as all attributes have aggregate value.

interface BaseBrokerConsumerStats { //getters long getAvailablePermits(); boolean isBlockedConsumerOnUnackedMsgs(); }

PartitionedBrokerConsumerStats implements BaseBrokerConsumerStats { //getters: this will return aggregate result long getAvailablePermits(); boolean isBlockedConsumerOnUnackedMsgs(); //getters: long getAvailablePermits(int partition); boolean isBlockedConsumerOnUnackedMsgs(int partition); }

BrokerConsumerStats implements BaseBrokerConsumerStats{ //getters }

and client can cast to PartitionedBrokerConsumerStats if they are really interested per partitions else consumerStats will be abstract from client.

…rStatsJavaClient

…e impl

jai1 · 2017-03-17T17:56:45Z

@rdhabalia - have implemented your suggestion

jai1 · 2017-03-17T17:57:10Z

@saandrews @rdhabalia - need a plus one on this - please

rdhabalia · 2017-03-20T23:51:03Z

@merlimat can you also please review it when you get a chance.

…rStatsJavaClient

saandrews · 2017-04-05T23:36:08Z

pulsar-client/src/main/java/com/yahoo/pulsar/client/api/BrokerConsumerStats.java

@@ -0,0 +1,84 @@
+/**
+ * Copyright 2016 Yahoo Inc.
+ *


Can you add relevant documentation as well?

saandrews · 2017-04-05T23:47:34Z

pulsar-client/src/main/java/com/yahoo/pulsar/client/impl/ClientCnx.java

@@ -352,6 +377,7 @@ protected boolean isHandshakeCompleted() {
                            writeFuture.cause().getMessage());
                    getAndRemovePendingLookupRequest(requestId);
                    future.completeExceptionally(writeFuture.cause());
+                    getAndRemovePendingLookupRequest(requestId);


Why do we remove requestId from pendingLookupRequest twice?

saandrews · 2017-04-05T23:53:37Z

pulsar-client/src/main/java/com/yahoo/pulsar/client/impl/ClientCnx.java

+                        writeFuture.cause().getMessage());
+                future.completeExceptionally(writeFuture.cause());
+            } else {
+                pendingConsumerStatsRequests.put(requestId, future);


We should have a way to limit pending requests.

* Upgrade dlog dependency * Make pulsar functions work after rebase - bump zk version 3.5 - use the same dlog version as bookkeeper - change standalone to use sorted ledger storage manager (this is a temp fix to address the issues after rebase)

* Fix send async comments Signed-off-by: xiaolong.ran <rxl@apache.org> * fix a little Signed-off-by: xiaolong.ran <rxl@apache.org> * fix a little Signed-off-by: xiaolong.ran <rxl@apache.org> * fix comments Signed-off-by: xiaolong.ran <rxl@apache.org>

Master issue: apache#241 Currently if a Kafka Client tries to list all topics, the topic name in response will contain the namespace prefix like "persistent://public/default". However, for topics in the default namespace, they shouldn't contain the prefix because Kafka Client would like to use the short topic name. So this PR will remove the prefix of topics in the default namespace. * Add a method to remove the default namespace prefix * Use short topic name if the topic is in the default namespace * Change variable name to avoid conflict * Fix simpleProduceAndConsumeWithPulsarAuthed test error

jai1 self-assigned this Feb 28, 2017

jai1 added type/bug The PR fixed a bug or issue reported a bug area/client area/test labels Feb 28, 2017

jai1 requested review from merlimat and rdhabalia February 28, 2017 23:39

merlimat reviewed Mar 1, 2017

View reviewed changes

jai1 requested a review from saandrews March 1, 2017 01:48

jai1 force-pushed the ConsumerStatsJavaClient branch from 48ee9f3 to 33e93cc Compare March 1, 2017 04:18

jai1 changed the title ~~Java client, C++ Async call and test cases for getConsumerStats~~ Java client Mar 1, 2017

jai1 changed the title ~~Java client~~ Java client for #194 Mar 1, 2017

merlimat changed the title ~~Java client for #194~~ Java Client - Support for getting consumer stats from broker Mar 1, 2017

merlimat added this to the 1.17 milestone Mar 1, 2017

merlimat removed type/bug The PR fixed a bug or issue reported a bug area/test labels Mar 1, 2017

merlimat reviewed Mar 1, 2017

View reviewed changes

saandrews reviewed Mar 3, 2017

View reviewed changes

rdhabalia reviewed Mar 4, 2017

View reviewed changes

jai1 force-pushed the ConsumerStatsJavaClient branch 4 times, most recently from f09d38a to 7949061 Compare March 6, 2017 23:55

Java Client - Support for getting consumer stats from broker

d7a3131

jai1 force-pushed the ConsumerStatsJavaClient branch from 7949061 to d7a3131 Compare March 7, 2017 00:01

Merge branch 'master' of https://github.com/yahoo/pulsar into Consume…

c2c9c80

…rStatsJavaClient

rdhabalia reviewed Mar 16, 2017

View reviewed changes

Jai ASher added 2 commits March 16, 2017 12:15

Merge branch 'master' of https://github.com/yahoo/pulsar into Consume…

ce5bda2

…rStatsJavaClient

Implemented Rajans suggestion to create single interface and split th…

da3ad89

…e impl

Jai ASher added 3 commits March 20, 2017 18:55

Some exception handling

4ce6a44

Added check to see if the Partitioned topic is ready

9bddb66

Merge branch 'master' of https://github.com/yahoo/pulsar into Consume…

f5769e6

…rStatsJavaClient

jai1 force-pushed the ConsumerStatsJavaClient branch from 2951e10 to f5769e6 Compare March 22, 2017 05:01

Jai ASher added 3 commits March 23, 2017 23:31

Merge branch 'master' of https://github.com/yahoo/pulsar into Consume…

ffadeed

…rStatsJavaClient

Corrected handling of futures

8fd385e

Fixed some test cases

de29999

merlimat modified the milestones: 1.17, 1.18 Mar 31, 2017

saandrews reviewed Apr 5, 2017

View reviewed changes

merlimat modified the milestones: 1.18, 1.19 Jun 14, 2017

jai1 closed this Jun 18, 2018

xiaotongwang1 mentioned this pull request Aug 4, 2021

Pulsar 2.7.0+ KOP 2.7.2.x getPartitionedTopicMetadata timeout #11532

Closed

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Java Client - Support for getting consumer stats from broker #254

Java Client - Support for getting consumer stats from broker #254

jai1 commented Feb 28, 2017 •

edited

Loading

merlimat Mar 1, 2017

merlimat Mar 1, 2017

merlimat commented Mar 1, 2017

jai1 commented Mar 1, 2017

merlimat Mar 1, 2017

jai1 Mar 2, 2017

saandrews Mar 3, 2017

rdhabalia Mar 3, 2017

rdhabalia Mar 3, 2017

rdhabalia Mar 3, 2017

jai1 Mar 6, 2017

rdhabalia Mar 3, 2017

rdhabalia Mar 4, 2017

jai1 Mar 6, 2017 •

edited

Loading

rdhabalia Mar 4, 2017

rdhabalia Mar 4, 2017

rdhabalia Mar 4, 2017

jai1 Mar 6, 2017

rdhabalia Mar 15, 2017

rdhabalia Mar 4, 2017

jai1 Mar 6, 2017

rdhabalia Mar 4, 2017

jai1 commented Mar 7, 2017

jai1 commented Mar 14, 2017

rdhabalia Mar 15, 2017

rdhabalia Mar 15, 2017

rdhabalia Mar 15, 2017

rdhabalia Mar 16, 2017

jai1 commented Mar 17, 2017

jai1 commented Mar 17, 2017

rdhabalia commented Mar 20, 2017

saandrews Apr 5, 2017

saandrews Apr 5, 2017

saandrews Apr 5, 2017


		log.debug(consumer1.getBrokerConsumerStatsAsync().get().toString());
		Assert.assertEquals(consumer1.getBrokerConsumerStatsAsync().get().getMsgBacklog(), totalMessages - count);



		CompletableFuture<BrokerConsumerStats> newConsumerStats(String topicName, String subscriptionName, long consumerId, long requestId) {
		CompletableFuture<BrokerConsumerStats> future = new CompletableFuture<BrokerConsumerStats>();

Java Client - Support for getting consumer stats from broker #254

Java Client - Support for getting consumer stats from broker #254

Conversation

jai1 commented Feb 28, 2017 • edited Loading

Motivation

Modifications

Result

Choose a reason for hiding this comment

Choose a reason for hiding this comment

merlimat commented Mar 1, 2017

jai1 commented Mar 1, 2017

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

jai1 Mar 6, 2017 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

jai1 commented Mar 7, 2017

jai1 commented Mar 14, 2017

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

jai1 commented Mar 17, 2017

jai1 commented Mar 17, 2017

rdhabalia commented Mar 20, 2017

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

jai1 commented Feb 28, 2017 •

edited

Loading

jai1 Mar 6, 2017 •

edited

Loading