[pulsar-client-cpp] branch main updated: [fix] Fix consumer doesn't clear incomingMessageQueue during seek (#101)

[kw...@apache.org]

This is an automated email from the ASF dual-hosted git repository.

kwang pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/pulsar-client-cpp.git


The following commit(s) were added to refs/heads/main by this push:
     new 94b909e  [fix] Fix consumer doesn't clear incomingMessageQueue during seek (#101)
94b909e is described below

commit 94b909e5684e72370279d69205d5f333dd4039d7
Author: Zike Yang <zi...@apache.org>
AuthorDate: Wed Nov 2 18:26:40 2022 +0800

    [fix] Fix consumer doesn't clear incomingMessageQueue during seek (#101)
    
    ### Motivation
    
    Currently, the consumer will receive old messages after the seek operation. Because the consumer doesn't clear the incomingMessageQueue before reconnecting. Therefore the consumer will dispatch some old messages to the user after the seek operation.
    
    This issue can be easily reproduced by advancing the process of creating consumers to before producing message in `testSeekForMessageId`.
    
    ### Modifications
    
    * Clear the incomingMessageQueue after seeking complete.
---
 lib/ConsumerImpl.cc   |  1 +
 tests/ConsumerTest.cc | 14 ++++++++------
 2 files changed, 9 insertions(+), 6 deletions(-)

diff --git a/lib/ConsumerImpl.cc b/lib/ConsumerImpl.cc
index bff5258..5698b46 100644
--- a/lib/ConsumerImpl.cc
+++ b/lib/ConsumerImpl.cc
@@ -1408,6 +1408,7 @@ void ConsumerImpl::seekAsyncInternal(long requestId, SharedBuffer seek, const Me
             if (result == ResultOk) {
                 LOG_INFO(getName() << "Seek successfully");
                 ackGroupingTrackerPtr_->flushAndClean();
+                incomingMessages_.clear();
                 Lock lock(mutexForMessageId_);
                 lastDequedMessageId_ = MessageId::earliest();
                 lock.unlock();
diff --git a/tests/ConsumerTest.cc b/tests/ConsumerTest.cc
index e35a1f0..a77636e 100644
--- a/tests/ConsumerTest.cc
+++ b/tests/ConsumerTest.cc
@@ -893,6 +893,14 @@ TEST_P(ConsumerSeekTest, testSeekForMessageId) {
     Producer producer;
     ASSERT_EQ(ResultOk, client.createProducer(topic, producerConf_, producer));
 
+    Consumer consumerExclusive;
+    ASSERT_EQ(ResultOk, client.subscribe(topic, "sub-0", consumerExclusive));
+
+    Consumer consumerInclusive;
+    ASSERT_EQ(ResultOk,
+              client.subscribe(topic, "sub-1", ConsumerConfiguration().setStartMessageIdInclusive(true),
+                               consumerInclusive));
+
     const auto numMessages = 100;
     MessageId seekMessageId;
 
@@ -909,16 +917,10 @@ TEST_P(ConsumerSeekTest, testSeekForMessageId) {
 
     LOG_INFO("The seekMessageId is: " << seekMessageId << ", r : " << r);
 
-    Consumer consumerExclusive;
-    ASSERT_EQ(ResultOk, client.subscribe(topic, "sub-0", consumerExclusive));
     consumerExclusive.seek(seekMessageId);
     Message msg0;
     ASSERT_EQ(ResultOk, consumerExclusive.receive(msg0, 3000));
 
-    Consumer consumerInclusive;
-    ASSERT_EQ(ResultOk,
-              client.subscribe(topic, "sub-1", ConsumerConfiguration().setStartMessageIdInclusive(true),
-                               consumerInclusive));
     consumerInclusive.seek(seekMessageId);
     Message msg1;
     ASSERT_EQ(ResultOk, consumerInclusive.receive(msg1, 3000));