Feature - support seek() on Reader #4031

lovelle · 2019-04-12T03:22:23Z

Motivation

Trying to fix #3976

According to what was discussed in pull #3983 it would be an acceptable solution
to add seek() command to Reader in order to reset a non durable cursor after
Reader instance was built.

Modifications

Bugfix reset() by timestamp on a non-durable consumer, previously the
cached cursor was not present, therefore the state set by reset() was missed
resulting in a reset() at the beginning of the cursor instead of a reset()
at the expected position.
Copy seek() commands to Reader interface from Consumer interface.
Fix inconsistency with lastDequeuedMessage field after seek() command was
performed successfully.
Fix consumer discarding messages on receive (after seek() command) due to
messages being present on acknowledge grouping tacker.

Verifying this change

Make sure that the change passes the CI checks.

ConcurrencyPractitioner · 2019-04-12T03:49:42Z

@lovelle It looks good!

ConcurrencyPractitioner · 2019-04-12T14:41:18Z

BTW, I don't know if you have considered other test scenarios. I think your test is only calling a seek() to the beginning of a sequence of messages you sent right (i.e. equivalent to seek(MessageId.EARLIEST)). Have you ever tried seeking for example to the middle of the sequence of messages and saw what happened?

I'm asking this because in my test, only seek(firstPublishTimeStamp) works, otherwise, as you saw in my testoutput, despite the fact that seek appeared to be successful, we still started processing from my-message-0.

lovelle · 2019-04-12T18:21:32Z

@ConcurrencyPractitioner great suggestion! please take a look now, I've just add more tests according to what you said, please let me know what you think! Thanks 👍

Added test seeks in the middle of the sequence of messages using MessageId and works great !

pulsar-broker/src/test/java/org/apache/pulsar/client/api/TopicReaderTest.java

jai1 · 2019-04-12T21:48:46Z

pulsar-broker/src/test/java/org/apache/pulsar/client/api/TopicReaderTest.java

+    @Test
+    public void testReaderIsAbleToSeekWithMessageIdOnMiddleOfTopic() throws Exception {
+        final String topicName = "persistent://my-property/my-ns/ReaderSeekWithMessageIdOnMiddleOfTopic";
+        final int numOfMessage = 100;


can you numOfMessage as an argument to this test and try with edge cases like
numOfMessage = [1, 9, 100]

By this you mean having a generic method doing all the asserts and receiving by arguments all the parameters?

jai1 · 2019-04-12T21:52:03Z

pulsar-client-api/src/main/java/org/apache/pulsar/client/api/Reader.java

+     * the individual partitions.
+     *
+     * @param messageId the message id where to reposition the reader
+     */


shouldn't the operation be renamed as reset instead of seek. I mean I would assume seek to just find the message id but not change the state of subscription.

mm I don't agree here, imho keeping seek() would be easier for the user understanding because of already existing seek on Consumer.

Also, what I think that is great of naming it seek() is that it behaves exactly the same as seeking on a file e.g. Linux which a lot of people already know what is meant for.

jai1 · 2019-04-12T21:58:27Z

...nt/src/main/java/org/apache/pulsar/client/impl/PersistentAcknowledgmentsGroupingTracker.java

+    public void flushAndClean() {
+        flush();
+        lastCumulativeAck = (MessageIdImpl) MessageId.earliest;
+        pendingIndividualAcks.clear();


Doesn't flush empty out pendingIndividualAcks

https://github.com/apache/pulsar/blob/5f157a2a82eff0202a9fc0920bad67f5aabcde87/pulsar-client/src/main/java/org/apache/pulsar/client/impl/PersistentAcknowledgmentsGroupingTracker.java#L170

If yes then I would not create a new function - I would just add lastCumulativeAck = (MessageIdImpl) MessageId.earliest; to the flush() function.

Yes, flush will empty pendingIndividualAcks but that doesn't mean that when you call flush() (do the effective redelivery of the message) also wants to lost track of lastCumulativeAck.

I added flushAndClean() because here we need to start over again and forget every message tracked, the same as when the constructor occurs.

ConcurrencyPractitioner · 2019-04-13T01:19:18Z

Sorry, just one last test case.
Have you considered testing the seek(long startPublishTime) to the middle of a sequence of messages?
I think seek(MessageId id) was expected to work, but what about the other seek method?

If seek(long startPublishTime) works as well, then I'm all for this issue.

ConcurrencyPractitioner · 2019-04-13T15:28:13Z

Oh, so @lovelle
I added the following test in my local repo and ran it:

    @Test
    public void testReaderIsAbleToSeekWithTime() throws Exception {
        final String topic = "persistent://my-property/my-ns/testReaderIsAbleToSeekWithTime";
        Producer<byte[]> producer = pulsarClient.newProducer().topic(topic).create();
        final long timeMs = System.currentTimeMillis();
        log.info("Current system time is: " + timeMs);
        for (int i = 0; i < 10; i++) {
           String message = "my-message-" + i;
           producer.send(message.getBytes("UTF-8"));
           Thread.sleep(1000);
       }

       log.info("Seeking to time: " + (timeMs + 3000));
       Reader<byte[]> reader = pulsarClient.newReader().topic(topic)
               .startMessageId(MessageId.earliest).create();

       reader.seek(timeMs + 3000);

       Message<byte[]> msg = null;
       Set<String> messageSet = Sets.newHashSet();
       for (int i = 3; i < 10; i++) {
           msg = reader.readNext();

           String receivedMessage = new String(msg.getData(), "UTF-8");
           log.info("Received message: [{}]", receivedMessage);
           String expectedMessage = "my-message-" + i;
           testMessageOrderAndDuplicates(messageSet, receivedMessage, expectedMessage);
       }
       reader.close();
       producer.close();
    }

And this was the test output:

[ERROR] testReaderIsAbleToSeekWithTime(org.apache.pulsar.client.api.TopicReaderTest)  Time elapsed: 10.112 s  <<< FAILURE!
java.lang.AssertionError: Received message my-message-0 did not match the expected message my-message-3 expected [my-message-3] but found [my-message-0]

I think seek(long timestamp) in your code is also suffering from the same problems that I am.

lovelle · 2019-04-13T16:45:08Z

@ConcurrencyPractitioner

Yes, you are right, reset on ManagedCursorImpl tells moving the cursor to 3:4 position but after this I still receive [3:0..3:9]
e.g:

13:36:22.428 [bookkeeper-ml-workers-OrderedExecutor-5-0:org.apache.bookkeeper.mledger.impl.ManagedCursorImpl@822] INFO  org.apache.bookkeeper.mledger.impl.ManagedCursorImpl - [my-property/my-ns/persistent/testReaderIsAbleToSeekWithTime] Initiate reset position to 3:4 on cursor reader-f3710e27ca
13:36:22.429 [bookkeeper-ml-workers-OrderedExecutor-5-0:org.apache.bookkeeper.mledger.impl.ManagedCursorImpl$7@861] INFO  org.apache.bookkeeper.mledger.impl.ManagedCursorImpl - [my-property/my-ns/persistent/testReaderIsAbleToSeekWithTime] reset position to 3:4 before current read position 3:10 on cursor reader-f3710e27ca
13:36:25.195 [main:org.apache.pulsar.client.api.TopicReaderTest@653] INFO  org.apache.pulsar.client.api.TopicReaderTest - received msg -> 3:0:-1:0
13:36:25.196 [main:org.apache.pulsar.client.api.TopicReaderTest@653] INFO  org.apache.pulsar.client.api.TopicReaderTest - received msg -> 3:1:-1:0
13:36:25.197 [main:org.apache.pulsar.client.api.TopicReaderTest@653] INFO  org.apache.pulsar.client.api.TopicReaderTest - received msg -> 3:2:-1:0
13:36:25.198 [main:org.apache.pulsar.client.api.TopicReaderTest@653] INFO  org.apache.pulsar.client.api.TopicReaderTest - received msg -> 3:3:-1:0
13:36:25.199 [main:org.apache.pulsar.client.api.TopicReaderTest@653] INFO  org.apache.pulsar.client.api.TopicReaderTest - received msg -> 3:4:-1:0
13:36:25.199 [main:org.apache.pulsar.client.api.TopicReaderTest@653] INFO  org.apache.pulsar.client.api.TopicReaderTest - received msg -> 3:5:-1:0
13:36:25.200 [main:org.apache.pulsar.client.api.TopicReaderTest@653] INFO  org.apache.pulsar.client.api.TopicReaderTest - received msg -> 3:6:-1:0
13:36:25.201 [main:org.apache.pulsar.client.api.TopicReaderTest@653] INFO  org.apache.pulsar.client.api.TopicReaderTest - received msg -> 3:7:-1:0
13:36:25.201 [main:org.apache.pulsar.client.api.TopicReaderTest@653] INFO  org.apache.pulsar.client.api.TopicReaderTest - received msg -> 3:8:-1:0
13:36:25.201 [main:org.apache.pulsar.client.api.TopicReaderTest@653] INFO  org.apache.pulsar.client.api.TopicReaderTest - received msg -> 3:9:-1:0

I will keep debugging this to see if I find something. 👍

*Motivation* Trying to fix apache#3976 According to what was discussed in pull apache#3983 it would be an acceptable solution to add seek() command to Reader in order to reset a non durable cursor after Reader instance was build. *Modifications* - Bugfix reset() by timestamp on a non-durable consumer, previously the cached cursor was not present, therefore the state set by reset() was missed resulting in a reset() at the beginning of the cursor instead of a reset() at the expected position. - Copy seek() commands to Reader interface from Consumer interface. - Fix inconsistency with lastDequeuedMessage field after seek() command was performed successfully. - Fix consumer discarding messages on receive (after seek() command) due to messages being present on acknowledge grouping tacker.

- Add functional test to assert seek() with timestamp behaviour from Reader to reach the beginning of the topic. - Add functional test to assert seek() with timestamp behaviour from Reader to reach the middle of the topic. - Add functional test to assert seek() with MessageId behaviour from Reader to reach the middle of the topic.

lovelle · 2019-04-15T23:59:39Z

@ConcurrencyPractitioner the last mentioned missed test case is now solved and added, please take a look now 👍

@sijie @jai1 this pull is ready to be fully reviewed, thanks!.

ConcurrencyPractitioner · 2019-04-16T00:14:47Z

👍 I see now, you made some changes to ManagedLedgerImpl. I could see now why seek(long timestamp) was not working. Nice work!

sijie

LGTM

sijie · 2019-04-16T08:31:43Z

run java8 tests

sijie · 2019-04-18T05:31:20Z

@jai1 can you please take a look again?

sijie · 2019-04-22T01:29:47Z

ping @jai1

lovelle mentioned this pull request Apr 12, 2019

[Issue#3976] [component/client] Adding config for ReaderImpl #3983

Closed

lovelle force-pushed the feature/support_seek_on_reader_impl branch from f333d36 to 80db9aa Compare April 12, 2019 18:19

lovelle force-pushed the feature/support_seek_on_reader_impl branch 3 times, most recently from cb61dbe to 5f157a2 Compare April 12, 2019 19:51

jai1 reviewed Apr 12, 2019

View reviewed changes

lovelle force-pushed the feature/support_seek_on_reader_impl branch from 5f157a2 to 0a5ea4c Compare April 13, 2019 01:38

lovelle force-pushed the feature/support_seek_on_reader_impl branch from 0a5ea4c to 71f10d9 Compare April 15, 2019 23:36

lovelle added 2 commits April 15, 2019 20:42

lovelle force-pushed the feature/support_seek_on_reader_impl branch from 71f10d9 to 12c8ab9 Compare April 15, 2019 23:52

sijie approved these changes Apr 16, 2019

View reviewed changes

jiazhai assigned lovelle Apr 21, 2019

jiazhai approved these changes Apr 21, 2019

View reviewed changes

sijie added area/client type/feature The PR added a new feature or issue requested a new feature labels Apr 23, 2019

sijie added this to the 2.4.0 milestone Apr 23, 2019

sijie merged commit c70438c into apache:master Apr 23, 2019

lovelle deleted the feature/support_seek_on_reader_impl branch April 23, 2019 20:33

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Feature - support seek() on Reader #4031

Feature - support seek() on Reader #4031

lovelle commented Apr 12, 2019 •

edited

Loading

ConcurrencyPractitioner commented Apr 12, 2019 •

edited

Loading

ConcurrencyPractitioner commented Apr 12, 2019

lovelle commented Apr 12, 2019 •

edited

Loading

jai1 Apr 12, 2019

lovelle Apr 13, 2019

jai1 Apr 12, 2019

lovelle Apr 13, 2019 •

edited

Loading

jai1 Apr 12, 2019

lovelle Apr 16, 2019 •

edited

Loading

ConcurrencyPractitioner commented Apr 13, 2019 •

edited

Loading

ConcurrencyPractitioner commented Apr 13, 2019

lovelle commented Apr 13, 2019 •

edited

Loading

lovelle commented Apr 15, 2019

ConcurrencyPractitioner commented Apr 16, 2019

sijie left a comment

sijie commented Apr 16, 2019

sijie commented Apr 18, 2019

sijie commented Apr 22, 2019

Feature - support seek() on Reader #4031

Feature - support seek() on Reader #4031

Conversation

lovelle commented Apr 12, 2019 • edited Loading

Verifying this change

ConcurrencyPractitioner commented Apr 12, 2019 • edited Loading

ConcurrencyPractitioner commented Apr 12, 2019

lovelle commented Apr 12, 2019 • edited Loading

jai1 Apr 12, 2019

Choose a reason for hiding this comment

lovelle Apr 13, 2019

Choose a reason for hiding this comment

jai1 Apr 12, 2019

Choose a reason for hiding this comment

lovelle Apr 13, 2019 • edited Loading

Choose a reason for hiding this comment

jai1 Apr 12, 2019

Choose a reason for hiding this comment

lovelle Apr 16, 2019 • edited Loading

Choose a reason for hiding this comment

ConcurrencyPractitioner commented Apr 13, 2019 • edited Loading

ConcurrencyPractitioner commented Apr 13, 2019

lovelle commented Apr 13, 2019 • edited Loading

lovelle commented Apr 15, 2019

ConcurrencyPractitioner commented Apr 16, 2019

sijie left a comment

Choose a reason for hiding this comment

sijie commented Apr 16, 2019

sijie commented Apr 18, 2019

sijie commented Apr 22, 2019

lovelle commented Apr 12, 2019 •

edited

Loading

ConcurrencyPractitioner commented Apr 12, 2019 •

edited

Loading

lovelle commented Apr 12, 2019 •

edited

Loading

lovelle Apr 13, 2019 •

edited

Loading

lovelle Apr 16, 2019 •

edited

Loading

ConcurrencyPractitioner commented Apr 13, 2019 •

edited

Loading

lovelle commented Apr 13, 2019 •

edited

Loading