Introduce topic reader in client API #371

merlimat · 2017-04-20T22:56:07Z

Motivation

As explained in #355, introduce Reader as a new way to receive data in client API.

Modifications

This PR is based on top of #366. In addition to the non-durable cursors, it adds the Reader client API and support in wire protocol to signal the broker where to start reading from.

Please take a look. I'll add unit tests later on this same PR.

merlimat · 2017-04-21T17:41:14Z

@rdhabalia Updated with tests

rdhabalia · 2017-04-21T17:44:54Z

I will review it soon.

rdhabalia · 2017-04-24T18:31:39Z

pulsar-broker/src/main/java/com/yahoo/pulsar/broker/service/ServerCnx.java

@@ -332,6 +329,12 @@ protected void handleSubscribe(final CommandSubscribe subscribe) {
        final long consumerId = subscribe.getConsumerId();
        final SubType subType = subscribe.getSubType();
        final String consumerName = subscribe.getConsumerName();
+        final boolean isDurable = subscribe.getDurable();


Old client don't pass this flag so, broker can consider that subscriber as non-durable. so, should we check subscribe.hasDurable()?

The flag is optional in the proto definition but it defaults to true, so there's no need to check for hasDurable()

I see, default is true

rdhabalia · 2017-04-24T18:32:18Z

pulsar-broker/src/main/java/com/yahoo/pulsar/broker/service/ServerCnx.java

-                                "Consumer is already present on the connection"));
+                        ServerError error = !existingConsumerFuture.isDone() ? ServerError.ServiceNotReady
+                                : getErrorCode(existingConsumerFuture);
+                        ;


rdhabalia · 2017-04-24T21:08:34Z

pulsar-broker/src/main/java/com/yahoo/pulsar/broker/service/persistent/PersistentTopic.java

+                Consumer consumer = new Consumer(subscription, subType, consumerId, priorityLevel, consumerName,
+                        maxUnackedMessages, cnx, cnx.getRole());
+                subscription.addConsumer(consumer);
+                if (!cnx.isActive()) {


instead can we avoid creation of Consumer if cnx is not active.

if (!cnx.isActive()) { future.completeExceptionally(new BrokerServiceException("Connection was closed while the opening the cursor ")); return; }

This part of the code is exactly the same as before, I just refactored the way to get the subscriptionFuture. For the specific concern, the problem is that the connection can flip to closed even after we have checked the isActive(), so we need to double check anyway the connection status after we have created the consumer, which internally register itself to get notified when the connection is gone.

rdhabalia

LGTM .. just few minor comments

rdhabalia · 2017-04-24T23:32:24Z

pulsar-broker/src/main/java/com/yahoo/pulsar/broker/service/ServerCnx.java

@@ -332,6 +329,12 @@ protected void handleSubscribe(final CommandSubscribe subscribe) {
        final long consumerId = subscribe.getConsumerId();
        final SubType subType = subscribe.getSubType();
        final String consumerName = subscribe.getConsumerName();
+        final boolean isDurable = subscribe.getDurable();


I see, default is true

rdhabalia · 2017-04-24T23:47:42Z

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

@@ -0,0 +1,82 @@
+package com.yahoo.pulsar.client.api;


rdhabalia · 2017-04-25T00:04:10Z

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

+     * @param msg
+     *            the message object
+     */
+    void received(Message msg);


I think it's better if api has reader => void received(Reader reader, Message msg) similar like consumerListener => void received(Consumer consumer, Message msg) because it helps to perform logging
or any additional operation on reader
e.g. => use to log : reader.getTopic()

Good point. I'll make the change

rdhabalia · 2017-04-25T00:05:17Z

pulsar-common/src/main/proto/PulsarApi.proto

+
+	// Signal wether the subscription should be backed by a
+	// durable cursor or not
+	optional bool  durable            = 8 [default = true];


extra space

rdhabalia · 2017-04-25T00:32:12Z

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

+        incomingMessages.drainTo(currentMessageQueue);
+        if (!currentMessageQueue.isEmpty()) {
+            return (MessageIdImpl) currentMessageQueue.get(0).getMessageId();
+        } else if (lastDequeuedMessage != null) {


I think every time when reader read msg from queue we update lastDequeuedMessage value. So, it means if lastDequeuedMessage is not null then we can use this position to start with. So, should we change the sequence:

if (lastDequeuedMessage != null) {

if (!currentMessageQueue.isEmpty()) { => then only we initialize currentMessageQueue and drain incomingMessages to get first position.
?

Actually, this code is wrong... I had changed the logic for startMessageId from referring to the "first message to be returned" to the "last message that has been consumed"... but that wasn't reflected here.

@rdhabalia Apart from the wrong logic here (delete pos vs read pos). The tricky problem is to tell the broker the exact message from where we want to resume reading after reconnection.

If we do have messages in the queue, we cannot rely on the lastDequeuedMessage, because another thread might be dequeueing at the same time. So we could see msg1 there, while the application already has dequeued msg2. At that point the msg2 will get duplicated after reconnection.

By atomically draining the queue on the side, we can establish which was the last message "seen" by the application. Checking the queue size and draining must be done atomically.

merlimat

Many good point. will fix that

merlimat · 2017-04-25T04:18:39Z

pulsar-broker/src/main/java/com/yahoo/pulsar/broker/service/persistent/PersistentTopic.java

+                Consumer consumer = new Consumer(subscription, subType, consumerId, priorityLevel, consumerName,
+                        maxUnackedMessages, cnx, cnx.getRole());
+                subscription.addConsumer(consumer);
+                if (!cnx.isActive()) {


This part of the code is exactly the same as before, I just refactored the way to get the subscriptionFuture. For the specific concern, the problem is that the connection can flip to closed even after we have checked the isActive(), so we need to double check anyway the connection status after we have created the consumer, which internally register itself to get notified when the connection is gone.

merlimat · 2017-04-25T04:18:51Z

pulsar-broker/src/main/java/com/yahoo/pulsar/broker/service/ServerCnx.java

-                                "Consumer is already present on the connection"));
+                        ServerError error = !existingConsumerFuture.isDone() ? ServerError.ServiceNotReady
+                                : getErrorCode(existingConsumerFuture);
+                        ;


merlimat · 2017-04-25T04:19:28Z

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

@@ -0,0 +1,82 @@
+package com.yahoo.pulsar.client.api;


merlimat · 2017-04-25T04:20:05Z

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

+     * @param msg
+     *            the message object
+     */
+    void received(Message msg);


Good point. I'll make the change

merlimat · 2017-04-25T04:40:33Z

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

+        incomingMessages.drainTo(currentMessageQueue);
+        if (!currentMessageQueue.isEmpty()) {
+            return (MessageIdImpl) currentMessageQueue.get(0).getMessageId();
+        } else if (lastDequeuedMessage != null) {


Actually, this code is wrong... I had changed the logic for startMessageId from referring to the "first message to be returned" to the "last message that has been consumed"... but that wasn't reflected here.

merlimat · 2017-04-25T19:18:14Z

@rdhabalia Addressed comments

rdhabalia · 2017-04-25T20:27:47Z

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

+        MessageIdImpl startMessageId;
+        synchronized (this) {
+            currentSize = incomingMessages.size();
+            startMessageId = clearReceiverQueue();


should we just clear incomingMessages.clear() if subscription has Durable mode?

I don't expect this to have a practical impact on the consumer performance, since it only happens on reconnections.

rdhabalia · 2017-04-25T20:35:07Z

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

+        } else if (lastDequeuedMessage != null) {
+            // If the queue was empty we need to restart from the message just after the last one that has been dequeued
+            // in the past
+            return lastDequeuedMessage;


If I understood it correctly then : when client tries to receive message from the queue, consumer calls messageProcessed which always updates lastDequeuedMessage, it means we know which msg, client has seen last. In that case, should we give first preference to lastDequeuedMessage and we should get it from incomingMessages only if lastDequeuedMessage is null?
so, we don't have to drain incomingMessages and can directly return lastDequeuedMessage?

Not exactly, if the queue was already empty, then we can rely on the lastDequeuedMessage, but if the queue has messages, it means the application is actually getting those messages and then updating the lastDequeuedMessage, though not atomically.

By draining the queue, we can establish with 100% accuracy what would be the next message , the one that the application has surely not got yet. With that, we can just take 1 step back and get the last-dequeued message id.

Not exactly, if the queue was already empty, then we can rely on the lastDequeuedMessage, but if the queue has messages, it means the application is actually getting those messages and then updating the lastDequeuedMessage

I couldn't understand actual reason so, sorry but let's say queue received msgs 1,2,3,4. and Reader read msgs 1 & 2. So, as soon as reader reads msg=2, lastDequeuedMessage gets updated with value=2 and queue is left with 3,4. Now, reader lost connection and reconnected so, now according to logic => incomingMessages drains 3,4 to currentMessageQueue and previousMessage=3-1=2. So, is there any issue if reader starts with lastDequeuedMessage=2 or just takes incomingMessages.poll(0, TimeUnit.MILLISECONDS)-1 and then incomingMessages.clear()

So, is there any issue if reader starts with lastDequeuedMessage=2 or just takes incomingMessages.poll(0, TimeUnit.MILLISECONDS)-1 and then incomingMessages.clear()

The goal here is to have exact precision on which message to ask the broker after the reconnection. When we drain the queue, we can determine with (with no race conditions) which is the 1st message that was in the queue.

If we just rely on the lastDequeuedMessage, the dequeing and the update of the lastDequeuedMessage are not synchronized and thus that could lead to some re-deliveries.

rdhabalia

👍

merlimat · 2017-05-05T23:06:22Z

@rdhabalia Can take a 2nd look at #366 as well ?

…pache#371) 1. remove log if the active transaction map is empty. 2. return the `highWaterMark` if there is no active transaction.

merlimat added the type/feature The PR added a new feature or issue requested a new feature label Apr 20, 2017

merlimat added this to the 1.18 milestone Apr 20, 2017

merlimat self-assigned this Apr 20, 2017

merlimat requested review from rdhabalia, jai1 and saandrews April 20, 2017 22:56

NonDurable cursor for managed ledger

98e687e

merlimat force-pushed the topic-reader branch from 854419b to 3f965b5 Compare April 21, 2017 17:40

Introduce topic reader in client API

fd3291e

merlimat force-pushed the topic-reader branch from 3f965b5 to fd3291e Compare April 21, 2017 17:46

rdhabalia reviewed Apr 24, 2017

View reviewed changes

rdhabalia reviewed Apr 25, 2017

View reviewed changes

merlimat commented Apr 25, 2017

View reviewed changes

Addressed comments

144bd81

merlimat force-pushed the topic-reader branch from 300c420 to 144bd81 Compare April 25, 2017 19:18

rdhabalia reviewed Apr 25, 2017

View reviewed changes

rdhabalia approved these changes May 5, 2017

View reviewed changes

merlimat merged commit 7badf1a into apache:master May 6, 2017

merlimat deleted the topic-reader branch May 6, 2017 00:22

merlimat mentioned this pull request May 6, 2017

Introduce topic reader #355

Closed

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Introduce topic reader in client API #371

Introduce topic reader in client API #371

merlimat commented Apr 20, 2017

merlimat commented Apr 21, 2017

rdhabalia commented Apr 21, 2017

rdhabalia Apr 24, 2017

merlimat Apr 24, 2017

rdhabalia Apr 24, 2017

rdhabalia Apr 24, 2017

merlimat Apr 25, 2017

rdhabalia Apr 24, 2017

merlimat Apr 25, 2017

rdhabalia left a comment

rdhabalia Apr 24, 2017

rdhabalia Apr 24, 2017

merlimat Apr 25, 2017

rdhabalia Apr 25, 2017

merlimat Apr 25, 2017

rdhabalia Apr 25, 2017

rdhabalia Apr 25, 2017

merlimat Apr 25, 2017

merlimat Apr 25, 2017

merlimat left a comment

merlimat Apr 25, 2017

merlimat Apr 25, 2017

merlimat Apr 25, 2017

merlimat Apr 25, 2017

merlimat Apr 25, 2017

merlimat commented Apr 25, 2017

rdhabalia Apr 25, 2017 •

edited

Loading

merlimat May 5, 2017

rdhabalia Apr 25, 2017

merlimat Apr 25, 2017

rdhabalia Apr 26, 2017

merlimat May 5, 2017

rdhabalia left a comment

merlimat commented May 5, 2017

Introduce topic reader in client API #371

Introduce topic reader in client API #371

Conversation

merlimat commented Apr 20, 2017

Motivation

Modifications

merlimat commented Apr 21, 2017

rdhabalia commented Apr 21, 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

rdhabalia left a 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

Choose a reason for hiding this comment

Choose a reason for hiding this comment

merlimat left a 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

merlimat commented Apr 25, 2017

rdhabalia Apr 25, 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

rdhabalia left a comment

Choose a reason for hiding this comment

merlimat commented May 5, 2017

rdhabalia Apr 25, 2017 •

edited

Loading