Get ConsumerStats using Binary Protocol

[jai1]

Motivation

Sometimes we need topic statics before consuming messages (Common use cases listed below), currently such information is obtained by making admin calls which use HTTP. Admin calls are expensive because each call spawns a new connection and may cause a redirect if the current broker doesn’t own the topic, we need a way to get this information without using HTTP.

Common use cases

• When Pulsar is used as a write-ahead log we decide whether to allow updates to a topic based on the amount of backlog the topic has.
• Consumer may need the number of pending acknowledgements.

Modifications

• Created a new command to get consumer stats - Changed Protobuf and regenerated files.
• Changed the C++ Client to get consumer stats.
• Made changed to Broker to respond to consumer stats request
• Added a caching mechanism to cache the consumer stats for 30 seconds.
• Changed ClientConnection to handle operation timeout

Result

We will be able to get consumer stats without having to use HTTP

[merlimat]

If you break it in 2 lines, then you need to use export PROTOC=... otherwise the variable will not picked up by the next script

[merlimat]

Why not reuse the already existing ConsumerStats class?

[jai1]

We need some subscriber level details as well - backlog, subscription type etc
But I have added a toMap method to convert the Class to a map

ca16032

[jai1]

@rdhabalia, @saandrews and @merlimat Would it be better if I add a class in the protobuf and client side instead of using a map?

[merlimat]

Cannot we use the existing ResultError instead of ResultConsumerStatsError. If I am making a call to get consumer stats, there's no need to re-state the type of the request.

[rdhabalia]

• all the formatting changes are with /pulsar-client-cpp/eclipse-formatter.xml ??

[rdhabalia]

i think cpp/.gitignore already contains: .idea so, we may not need this.

[jai1]

Git ignore will prevent the idea folder to be checked in and this change will prevent the local build to fail

[rdhabalia]

no need of multiple lines :
PersistentTopic topic = (PersistentTopic) getBrokerService().getTopicReference(topicName);

[rdhabalia]

now, cpp library is added, we can also mention : PulsarApi.pb.cc

[rdhabalia]

ByteBuf msg

[rdhabalia]

should write appropriate message such as topic not found

[jai1]

Have written appropriate message in line 244, 240 and 236 (same file)

[saandrews]

Log remaining info as well in all the above error logs? topic name, requested, surname & consumerid?

[rdhabalia]

I mean error log such as: Failed to get consumer-stats response so, it can help while reading log file where exactly it occurred.

[rdhabalia]

same here: error-msg

[jai1]

Have written appropriate message in line 244, 240 and 236 (same file)
This case is for handling some exception

[saandrews]

If this exception is sent to the client, shouldn't it be the one client knows about?

[rdhabalia]

should we rename it as BrokerConsumerStats or ConsumerStats?

[jai1]

There is already a ConsumerStats class in Java client which captures the message rate, acks etc as sent by client - created by Sangming and Sid.

[saandrews]

I would prefer BrokerConsumerStats or ConsumerStatsInfo.

[rdhabalia]

is there any difference here?

[jai1]

yes, the compiler doesn't give ordering guarantees for ++ if used in complex calculations
Try compiling in your mac m/c and you will see the warning.

[rdhabalia]

formatting

[rdhabalia]

formatting

[rdhabalia]

just a thought:

• if client requires subscription level stats then : if consumerId is not provided then should broker return subscription level stats in the map?

[rdhabalia]

for C++ -> git checkout v2.6.0

[rdhabalia]

Do we need topic_name, subscription_name, consumer_id in response? can't we just return request_id as client knows information attached with it?

[jai1]

Yes we can but a little more info might help in debug if we run into some trouble and since this call can be made at most once in 30 seconds - it won't harm the Network BW much.

Still if you feel strongly about it let me know - i will remove the fields

[rdhabalia]

yes, we should not send extra field in response except request_id

[rdhabalia]

I mean error log such as: Failed to get consumer-stats response so, it can help while reading log file where exactly it occurred.

[rdhabalia]

do we really need timer to expiry the request. we already have keep alive time which will break the connection if broker is not responding and it should clear-out pending requests.

[rdhabalia]

should we also provide getConsumerStats(BrokerSideConsumerStats& brokerSideConsumerStats) ?

[jai1]

As discussed yesterday - We can the code is not difficult but the interface will become quite ugly since now we need to return a list of stats always.

[rdhabalia]

should we initialize validTill_+X so it can be valid for X time.??

[jai1]

we do that in ClientConnection

                        boost::posix_time::ptime currentTime = now() + milliseconds(consumerStatsTTLMs_);
                        consumerStatsPromise.setValue(BrokerSideConsumerStats(consumerStats, currentTime));

[rdhabalia]

is it a blocking call on future?

[jai1]

yes

[rdhabalia]

then should we make it non blocking call by attaching the listener?

[jai1]

It will block the user thread not out io thread, similar logic is used to producer.send and consumer.receive

Result Producer::send(const Message& msg) {
    Promise<Result, Message> promise;
    sendAsync(msg, WaitForCallbackValue<Message>(promise));

    Message m;
    Result result = promise.getFuture().get(m);
    return result;
}

[rdhabalia]

Do you think Is it worth it to provide callback option also in case of client doesn't want to block such as: sendAsync(msg, callBack)

[rdhabalia]

v7?

[saandrews]

If this exception is sent to the client, shouldn't it be the one client knows about?

[saandrews]

Log remaining info as well in all the above error logs? topic name, requested, surname & consumerid?

[saandrews]

I would prefer BrokerConsumerStats or ConsumerStatsInfo.

[saandrews]

Its not really currentTime

[saandrews]

Copy pasted message? Also we shouldn't have to explicitly say "In getConsumerStats call"

[saandrews]

Remove "In getConsumerStats call -"

[saandrews]

Why INFO log?

[jai1]

getCause().getMessage() is not very informative

[jai1]

In a few test runs I found that tls was automatically enabled - hence initialized the params

[jai1]

datatype compatible with what we send over the wire

[jai1]

both mean the same - changed since underscores were ugly

static const SubType Exclusive = CommandSubscribe_SubType_Exclusive;
static const SubType Shared = CommandSubscribe_SubType_Shared;
static const SubType Failover = CommandSubscribe_SubType_Failover;

[rdhabalia]

this will be CompletionException and it will not be helpful, we have to log exception.getCause()

[saandrews]

Log topic and subscription id as well?

[saandrews]

Unlock here.

[saandrews]

Fix log msg.

[saandrews]

Fix msg

[saandrews]

Fix the formatting caused by the editor w.r.t. space and &.

[jai1]

@merlimat - if you find time, can you review this PR.

[merlimat]

Ok, I was about to get to it ;)

…

On Mon, Feb 27, 2017 at 9:52 AM jai1 ***@***.***> wrote: Merged #216 <#216>. — You are receiving this because you were mentioned. Reply to this email directly, view it on GitHub <#216 (comment)>, or mute the thread <https://github.com/notifications/unsubscribe-auth/AAD0JFYQ-dU-Mz5EZMqX8kadwm8fhT5bks5rgw1PgaJpZM4MDYHI> .

[jai1]

Sorry - was in a hurry to get this out, will plan better next time onwards On Mon, Feb 27, 2017 at 9:54 AM, Matteo Merli <notifications@github.com> wrote:

…

Ok, I was about to get to it ;) On Mon, Feb 27, 2017 at 9:52 AM jai1 ***@***.***> wrote: > Merged #216 <#216>. > > — > You are receiving this because you were mentioned. > Reply to this email directly, view it on GitHub > <#216 (comment)>, or mute the > thread > <https://github.com/notifications/unsubscribe-auth/AAD0JFYQ-dU- Mz5EZMqX8kadwm8fhT5bks5rgw1PgaJpZM4MDYHI> > . > — You are receiving this because you modified the open/close state. Reply to this email directly, view it on GitHub <#216 (comment)>, or mute the thread <https://github.com/notifications/unsubscribe-auth/AIQh_F8HALVIdU1obGNbSU6712P0g9_Bks5rgw3XgaJpZM4MDYHI> .