[C++] fix cpp client do AcknowledgeCumulative not clean up previous message by saosir · Pull Request #8606 · apache/pulsar

saosir · 2020-11-18T07:53:53Z

pulsar-client-cpp Consumer UnAckedMessageTrackerEnabled::removeMessagesTill should erase message whose id <= msgId in messageIdPartitionMap

Motivation

pulsar-client-cpp Consumer do AcknowledgeCumulative just clean up msgId, not <= msgId in UnAckedMessageTrackerEnabled::removeMessagesTill

Modifications

When do AcknowledgeCumulative from application, earse <= msgId in UnAckedMessageTrackerEnabled, avoid redeliver unnecessary unacknowledged messages to Broker
add unit test for UnAckedMessageTrackerEnabled

Does this pull request potentially affect one of the following parts:

If yes was chosen, please highlight the changes

Dependencies (does it add or upgrade a dependency): (no)
The public API: (no)
The schema: (no)
The default values of configurations: (no)
The wire protocol: (no)
The rest endpoints: (no)
The admin cli options: (no)
Anything that affects deployment: (no)

Documentation

Does this pull request introduce a new feature? (no)

BewareMyPower · 2020-11-21T15:38:42Z

Could you explain why the last message id should not be removed? See

pulsar/pulsar-client-cpp/lib/ConsumerImpl.cc

Lines 851 to 854 in 281163b

    
           void ConsumerImpl::doAcknowledgeCumulative(const MessageId& messageId, ResultCallback callback) { 
        
               this->unAckedMessageTrackerPtr_->removeMessagesTill(messageId); 
        
               this->batchAcknowledgementTracker_.deleteAckedMessage(messageId, proto::CommandAck::Cumulative); 
        
               this->ackGroupingTrackerPtr_->addAcknowledgeCumulative(messageId);

After messages whose id is <= messageId are removed, the messageId would be added to ackGroupingTrackerPtr_ to mark it as an acknowledged message though it's a pending state.

And see Java client's implementation:

pulsar/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ConsumerImpl.java

Lines 784 to 786 in 281163b

    
           } else if (ackType == AckType.Cumulative) { 
        
               onAcknowledgeCumulative(messageId, null); 
        
               stats.incrementNumAcksSent(unAckedMessageTracker.removeMessagesTill(msgId));

pulsar/pulsar-client/src/main/java/org/apache/pulsar/client/impl/UnAckedMessageTracker.java

Lines 225 to 239 in 281163b

    
           public int removeMessagesTill(MessageId msgId) { 
        
               writeLock.lock(); 
        
               try { 
        
                   int removed = 0; 
        
                   Iterator<MessageId> iterator = messageIdPartitionMap.keySet().iterator(); 
        
                   while (iterator.hasNext()) { 
        
                       MessageId messageId = iterator.next(); 
        
                       if (messageId.compareTo(msgId) <= 0) { 
        
                           ConcurrentOpenHashSet<MessageId> exist = messageIdPartitionMap.get(messageId); 
        
                           if (exist != null) { 
        
                               exist.remove(messageId); 
        
                           } 
        
                           iterator.remove(); 
        
                           removed ++; 
        
                       }

The behaviors of Java and C++ client are the same.

saosir · 2020-11-23T01:59:47Z

@BewareMyPower Sorry，I did not describe clearly. UnAckedMessageTrackerEnabled::removeMessagesTill should erase message whose id <= msgId in messageIdPartitionMap

see https://github.com/apache/pulsar/pull/8606/files, the origin implementation of cpp is different from that of Java client. cpp client just earse msgId (function param msgId), not <= msgId

merlimat · 2020-11-23T22:45:29Z

@saosir Can you add a unit test that reproduces the issue? That will also help people to understand the problem.

BewareMyPower · 2020-11-24T01:10:07Z

@saosir OK, I see.

sijie · 2020-11-30T03:31:50Z

@BewareMyPower Can you review this pull request again?

BewareMyPower · 2020-11-30T03:35:40Z

@BewareMyPower Can you review this pull request again?

OK

saosir · 2020-11-30T06:59:52Z

@BewareMyPower I have two questions here, can you help me answer them?

Both PartitionedConsumerImpl and ConsumerImpl have member variable unAckedMessageTrackerPtr_ (class UnAckedMessageTrackerEnabled), and PartitionedConsumerImpl is composed of ConsumerImpl. if the acknowledgement times out, will they send redeliverMessages repeatedly?
in UnAckedMessageTrackerEnabled, param msg of UnAckedMessageTrackerEnabled::add contain partition, ledgerId, entryId,batchIndex. When ack timeout and do redeliverUnacknowledgedMessages, does it do deduplicte based on (ledgerId, entryId)? If not it will send repeat ack(ledgerId, entryId)

BewareMyPower · 2020-11-30T07:33:35Z

@saosir

From my perspective, both you concerned are right. But the Java client has the same behavior, like MultiTopicsConsumerImpl and ConsumerImpl both hold an UnAckedMessageTracker instance. Except that Java client supports chucked messages but C++ client doesn't.

saosir · 2020-11-30T08:21:10Z

@BewareMyPower
Does UnAckedMessageTrackerEnabled have same problem like #8519 ?

BewareMyPower · 2020-11-30T08:47:27Z

Does UnAckedMessageTrackerEnabled have same problem like #8519 ?

Yeah, consumerReference_ may be invalid when the timer callback is called.

saosir · 2020-11-30T08:53:51Z

@BewareMyPower
I will submit a fix and test case for this issue later, and help review it

BewareMyPower · 2020-11-30T09:58:39Z

@saosir It's pleasure to have your contribution.

By the way, I just looked at the UnAckedMessageTrackerEnabled again. There's something to correct.

When you created a consumer with unacked messages tracker enabled like:

    Consumer consumer;
    ConsumerConfiguration consumerConf;
    consumerConf.setUnAckedMessagesTimeoutMs(11000);  // must >= 10000
    consumerConf.setTickDurationInMs(11000);
    Result result = client.subscribe("my-topic", "consumer-1", consumerConf, consumer);

If my-topic has N partitions, N+1 UnAckedMessageTrackerEnabled will be created. However, only the tracker of PartitionedConsumerImpl will add message id. Because when the internal consumers were created, the listener was set with the partitioned consumer's method, see

pulsar/pulsar-client-cpp/lib/PartitionedConsumerImpl.cc

Lines 231 to 232 in 102fa9d

    
           const auto shared_this = const_cast<PartitionedConsumerImpl*>(this)->shared_from_this(); 
        
           config.setMessageListener(std::bind(&PartitionedConsumerImpl::messageReceived, shared_this, _1, _2));

And ConsumerImpl::messageReceived wouldn't be called, which means ConsumerImpl#notifyPendingReceivedCallback and ConsumerImpl#internalListener wouldn't be called.

Therefore, the trackers of sub-consumers only do the redelivery but not add any message id, see

pulsar/pulsar-client-cpp/lib/PartitionedConsumerImpl.cc

Lines 497 to 499 in 102fa9d

    
           for (ConsumerList::const_iterator i = consumers_.begin(); i != consumers_.end(); i++) { 
        
               (*i)->redeliverUnacknowledgedMessages(messageIds); 
        
           }

redeliverUnacknowledgedMessages(messageIds) for shared and key shared subscriptions, and redeliverUnacknowledgedMessages() for other subscriptions.

saosir · 2020-11-30T12:34:07Z

@BewareMyPower PartitionedConsumerImpl internal consumer ConsumerImpl::messageReceived would be called first in ClientConnection, see

pulsar/pulsar-client-cpp/lib/ClientConnection.cc

Lines 666 to 669 in 102fa9d

    
           // Unlock the mutex before notifying the consumer of the 
        
           // new received message 
        
           lock.unlock(); 
        
           consumer->messageReceived(shared_from_this(), msg, isChecksumValid, msgMetadata, payload);

and ConsumerImpl#internalListener will be called and notify PartitionedConsumerImpl receive messge by listener of ConsumerConfiguration(here is PartitionedConsumerImpl::messageReceived)

saosir · 2020-12-01T01:32:57Z

I reset branch at first commit

BewareMyPower · 2020-12-01T01:55:36Z

Thanks. Please update the associated PR description too.

Also I think the tests could still be added without the refactor, like what AckGroupingTrackerEnabled tests did.

saosir · 2020-12-01T11:39:21Z

Thanks. Please update the associated PR description too.

Also I think the tests could still be added without the refactor, like what AckGroupingTrackerEnabled tests did.

All is done

BewareMyPower · 2020-12-01T14:20:55Z

/pulsarbot run-failure-checks

BewareMyPower · 2020-12-01T14:49:17Z

pulsar-client-cpp/tests/BasicEndToEndTest.cc

+    long size() { return UnAckedMessageTrackerEnabled::size(); }
+};  // class UnAckedMessageTrackerEnabledMock
+
+TEST(BasicEndToEndTest, testtUnAckedMessageTrackerDefaultBehavior) {


A typo: testtUnAcked -> testUnAcked

Do I need to resubmit the code to fix this problem?How about someone fix it when merge pull request?

I think a committer could fix it when merge PR. So you needn't any changes. @jiazhai could you help take a look?

@saosir Can you create a follow-up PR to fix the typo?

BewareMyPower

LGTM, just leave a typo comment.

BewareMyPower · 2020-12-02T02:29:32Z

/pulsarbot run-failure-checks

BewareMyPower · 2020-12-02T12:48:03Z

/pulsarbot run-failure-checks

BewareMyPower · 2020-12-02T15:40:09Z

/pulsarbot run-failure-checks

BewareMyPower · 2020-12-24T08:10:14Z

/pulsarbot run-failure-checks

saosir · 2020-12-25T07:15:40Z

@BewareMyPower CI - Integration - Cli / cli (pull_request) fail, see #8790. Do I need to rebase to master branch?

BewareMyPower · 2020-12-25T12:54:46Z

@saosir Yeah, your branch is far behind the master, it's better to rebase to latest master.

…essage

saosir · 2020-12-26T15:26:11Z

/pulsarbot run-failure-checks

BewareMyPower · 2020-12-27T12:31:25Z

LGTM. @jiazhai Could you help take a look?

sijie · 2020-12-30T10:35:00Z

pulsar-client-cpp/tests/BasicEndToEndTest.cc

+    long size() { return UnAckedMessageTrackerEnabled::size(); }
+};  // class UnAckedMessageTrackerEnabledMock
+
+TEST(BasicEndToEndTest, testtUnAckedMessageTrackerDefaultBehavior) {


@saosir Can you create a follow-up PR to fix the typo?

…essage (#8606) ### Motivation pulsar-client-cpp Consumer do AcknowledgeCumulative just clean up `msgId`, not <= `msgId` in `UnAckedMessageTrackerEnabled::removeMessagesTill` ### Modifications - When do AcknowledgeCumulative from application, earse <= `msgId` in UnAckedMessageTrackerEnabled, avoid redeliver unnecessary unacknowledged messages to Broker - add unit test for `UnAckedMessageTrackerEnabled` (cherry picked from commit e75de48)

aahmed-se requested a review from merlimat November 20, 2020 08:16

saosir changed the title ~~fix cpp client AcknowledgeCumulative~~ [C++] fix cpp client AcknowledgeCumulative Nov 23, 2020

saosir changed the title ~~[C++] fix cpp client AcknowledgeCumulative~~ [C++] fix cpp client do AcknowledgeCumulative not clean up previous message in UnAckedMessageTrackerEnabled Nov 26, 2020

saosir changed the title ~~[C++] fix cpp client do AcknowledgeCumulative not clean up previous message in UnAckedMessageTrackerEnabled~~ [C++] fix cpp client do AcknowledgeCumulative not clean up previous message Nov 26, 2020

saosir force-pushed the patch-2 branch 4 times, most recently from a8336f7 to e98394a Compare November 27, 2020 05:00

sijie assigned saosir Nov 30, 2020

sijie added component/c++ area/client release/2.7.1 labels Nov 30, 2020

sijie added this to the 2.8.0 milestone Nov 30, 2020

saosir force-pushed the patch-2 branch 4 times, most recently from f846daa to ed3704c Compare November 30, 2020 14:56

saosir force-pushed the patch-2 branch from 5a93aed to c3217ef Compare December 1, 2020 01:37

saosir force-pushed the patch-2 branch 4 times, most recently from 8ed9502 to d3f76e5 Compare December 1, 2020 11:10

BewareMyPower reviewed Dec 1, 2020

View reviewed changes

BewareMyPower approved these changes Dec 1, 2020

View reviewed changes

saosir removed their assignment Dec 4, 2020

saosir mentioned this pull request Dec 22, 2020

[C++] PartitionedConsumerImpl send redeliverMessages repeatedly #9028

Closed

sijie mentioned this pull request Dec 22, 2020

ISSUE-9028: [C++] PartitionedConsumerImpl send redeliverMessages repeatedly streamnative/pulsar-archived#1925

Closed

saosir force-pushed the patch-2 branch from d3f76e5 to 2b3d2a7 Compare December 26, 2020 06:38

[C++] Fix cpp client do AcknowledgeCumulative not clean up previous m…

19f7c6e

…essage

saosir force-pushed the patch-2 branch from 2b3d2a7 to 19f7c6e Compare December 26, 2020 14:08

sijie approved these changes Dec 30, 2020

View reviewed changes

sijie merged commit e75de48 into apache:master Dec 30, 2020

saosir deleted the patch-2 branch January 5, 2021 13:53

codelipenghui added the cherry-picked/branch-2.7 Archived: 2.7 is end of life label Jan 7, 2021

Conversation

saosir commented Nov 18, 2020 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Motivation

Modifications

Does this pull request potentially affect one of the following parts:

Documentation

Uh oh!

BewareMyPower commented Nov 21, 2020 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

saosir commented Nov 23, 2020 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

merlimat commented Nov 23, 2020

Uh oh!

BewareMyPower commented Nov 24, 2020

Uh oh!

sijie commented Nov 30, 2020

Uh oh!

BewareMyPower commented Nov 30, 2020

Uh oh!

saosir commented Nov 30, 2020 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

BewareMyPower commented Nov 30, 2020

Uh oh!

saosir commented Nov 30, 2020

Uh oh!

BewareMyPower commented Nov 30, 2020

Uh oh!

saosir commented Nov 30, 2020

Uh oh!

BewareMyPower commented Nov 30, 2020

Uh oh!

saosir commented Nov 30, 2020 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

saosir commented Dec 1, 2020

Uh oh!

BewareMyPower commented Dec 1, 2020

Uh oh!

saosir commented Dec 1, 2020 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

BewareMyPower commented Dec 1, 2020

Uh oh!

BewareMyPower Dec 1, 2020

Choose a reason for hiding this comment

Uh oh!

saosir Dec 2, 2020

Choose a reason for hiding this comment

Uh oh!

BewareMyPower Dec 2, 2020

Choose a reason for hiding this comment

Uh oh!

sijie Dec 30, 2020 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

BewareMyPower left a comment

Choose a reason for hiding this comment

Uh oh!

BewareMyPower commented Dec 2, 2020

Uh oh!

BewareMyPower commented Dec 2, 2020

Uh oh!

BewareMyPower commented Dec 2, 2020

Uh oh!

BewareMyPower commented Dec 24, 2020

Uh oh!

saosir commented Dec 25, 2020

Uh oh!

BewareMyPower commented Dec 25, 2020

Uh oh!

saosir commented Dec 26, 2020

Uh oh!

BewareMyPower commented Dec 27, 2020

Uh oh!

sijie Dec 30, 2020 • edited Loading Uh oh! There was an error while loading. Please reload this page.

saosir commented Nov 18, 2020 •

edited

Loading

BewareMyPower commented Nov 21, 2020 •

edited

Loading

saosir commented Nov 23, 2020 •

edited

Loading

saosir commented Nov 30, 2020 •

edited

Loading

saosir commented Nov 30, 2020 •

edited

Loading

saosir commented Dec 1, 2020 •

edited

Loading

sijie Dec 30, 2020 •

edited

Loading

sijie Dec 30, 2020 •

edited

Loading