[C++] Fix Consumer send redeliverMessages repeatedly #9072

saosir · 2020-12-28T12:24:28Z

Motivation

Both PartitionedConsumerImpl and ConsumerImpl have member variable unAckedMessageTrackerPtr_ (class UnAckedMessageTrackerEnabled), and PartitionedConsumerImpl is composed of ConsumerImpl. If the acknowledgement times out, they will send redeliverMessages repeatedly, MultiTopicsConsumerImpl has same problem.

Modifications

add hasParent_ field to whether there is a parent in ConsumerImpl
add unit test verify the redelivery request won't be sent repeatedly
add GTest header gtest/gtest_prod.h to access private members in unit tests
fix typo

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)

codelipenghui · 2020-12-28T12:49:16Z

@BewareMyPower Please help review this PR.

BewareMyPower

Could you add a test like #4595 did?

saosir · 2020-12-29T02:24:44Z

@BewareMyPower Is it necessary to reset the batchIndex of MessageId when tracking a message . Because only the ack of the last message in batchMessage will be removed from UnAckedMessageTrackerEnabled

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

Lines 808 to 817 in b09c0d7

    
           void ConsumerImpl::acknowledgeAsync(const MessageId& msgId, ResultCallback callback) { 
        
               ResultCallback cb = std::bind(&ConsumerImpl::statsCallback, shared_from_this(), std::placeholders::_1, 
        
                                             callback, proto::CommandAck_AckType_Individual); 
        
               if (msgId.batchIndex() != -1 && 
        
                   !batchAcknowledgementTracker_.isBatchReady(msgId, proto::CommandAck_AckType_Individual)) { 
        
                   cb(ResultOk); 
        
                   return; 
        
               } 
        
               doAcknowledgeIndividual(msgId, cb); 
        
           }

But the param msgId of add UnAckedMessageTrackerEnabled#add may contain batchIndex

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

Lines 93 to 102 in b09c0d7

    
           bool UnAckedMessageTrackerEnabled::add(const MessageId& m) { 
        
               std::lock_guard<std::mutex> acquire(lock_); 
        
               if (messageIdPartitionMap.count(m) == 0) { 
        
                   std::set<MessageId>& partition = timePartitions.back(); 
        
                   bool emplace = messageIdPartitionMap.emplace(m, partition).second; 
        
                   bool insert = partition.insert(m).second; 
        
                   return emplace && insert; 
        
               } 
        
               return false; 
        
           }

BewareMyPower · 2020-12-29T10:35:49Z

@saosir Yeah, your concern is right.

sijie · 2021-01-04T17:41:20Z

@BewareMyPower Can you review this PR again?

BewareMyPower · 2021-01-05T07:21:49Z

@sijie I think a unit tests is required for this change, but the recent changes don't contain a test.

@saosir could you add a test to verify the redelivery request won't be sent repeatedly?

saosir · 2021-01-05T07:42:43Z

@BewareMyPower Test cases have been added, and the test cases use gtest to access private member variables. I don’t know if this is appropriate.

Help review it, thanks!

BewareMyPower · 2021-01-05T07:46:24Z

@saosir It's not a problem to access private members in tests. I'll take a look.

saosir · 2021-01-05T07:57:08Z

#9072 (comment)

I will submit another pr to deal with this problem later.

BewareMyPower

The code itself looks good to me. However, introducing FRIEND_TEST to code under lib/ is not a good way.

Currently, users could define -DBUILD_TESTS=OFF in the CMake build phase to skip tests and build libraries only. However, after changes of this PR, the build will fail because it added an extra dependency (GTest) to lib/ subdirectory.

I think you can find an alternative way of FRIEND_TEST or use a conditional macro to hide the gtest codes under lib/. Though you can also just add the GTest dependency in CMakeLists.txt, it's not recommended because the source code without tests should not depend on GTest.

saosir · 2021-01-05T08:28:06Z

The code itself looks good to me. However, introducing FRIEND_TEST to code under lib/ is not a good way.

Currently, users could define -DBUILD_TESTS=OFF in the CMake build phase to skip tests and build libraries only. However, after changes of this PR, the build will fail because it added an extra dependency (GTest) to lib/ subdirectory.

I think you can find an alternative way of FRIEND_TEST or use a conditional macro to hide the gtest codes under lib/. Though you can also just add the GTest dependency in CMakeLists.txt, it's not recommended because the source code without tests should not depend on GTest.

How about add Gtest header gtest/gtest_prod.h to dir pulsar-client-cpp/include

BewareMyPower · 2021-01-05T09:01:18Z

How about add Gtest header gtest/gtest_prod.h to dir pulsar-client-cpp/include

Yeah, it's a simple and good solution.

saosir · 2021-01-05T09:40:37Z

@BewareMyPower I have a question.

Why default router policy of PartitionedProducer is SinglePartition in pulsar-client-cpp

pulsar/pulsar-client-cpp/lib/ProducerConfigurationImpl.h

Line 56 in afe6af2

routingMode(ProducerConfiguration::UseSinglePartition),

but in pulsar-client-java is RoundRobinPartition

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

Lines 317 to 319 in afe6af2

    
           if(conf.getMessageRoutingMode() == null && conf.getCustomMessageRouter() == null) { 
        
               messageRoutingMode(MessageRoutingMode.RoundRobinPartition); 
        
           } else if(conf.getMessageRoutingMode() == null && conf.getCustomMessageRouter() != null) {

BewareMyPower · 2021-01-05T10:02:02Z

@saosir I'm also not sure why it's the default routing mode. It could make pulsar beginners confused like I've experienced before. But considering the change of default behavior may affect existed applications, I'd like to keep it as default.

saosir · 2021-01-05T11:45:34Z

How about add Gtest header gtest/gtest_prod.h to dir pulsar-client-cpp/include

Yeah, it's a simple and good solution.

CI will check license header, but gtest/gtest_prod.h contain google license cause it fail.

Failed to execute goal com.mycila:license-maven-plugin:4.0.rc2:check (default) on project pulsar: Some files do not have the expected license header

BewareMyPower · 2021-01-05T11:50:44Z

@saosir You can add this file to pom.xml's excludes:

      <plugin>
        <groupId>com.mycila</groupId>
        <artifactId>license-maven-plugin</artifactId>
        <version>${license-maven-plugin.version}</version>
        <configuration>
          <licenseSets>
            <licenseSet>
              <header>${pulsar.basedir}/src/license-header.txt</header>
              <excludes>
                <!-- TODO: add it here -->

saosir · 2021-01-05T14:19:04Z

why blankPartitions + 1 in UnAckedMessageTrackerEnabled constructor ? If timeoutMs_ == tickDurationInMs, it will wait 2*timeoutMs_ to send redelivery messages at first timeout.

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

Lines 84 to 88 in afe6af2

    
           int blankPartitions = (int)std::ceil((double)timeoutMs_ / tickDurationInMs_); 
        
           for (int i = 0; i < blankPartitions + 1; i++) { 
        
               std::set<MessageId> msgIds; 
        
               timePartitions.push_back(msgIds); 
        
           }

BewareMyPower · 2021-01-05T14:50:10Z

why blankPartitions + 1 in UnAckedMessageTrackerEnabled constructor ? If timeoutMs_ == tickDurationInMs, it will wait 2*timeoutMs_ to send redelivery messages at first timeout.

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

Lines 84 to 88 in afe6af2

int blankPartitions = (int)std::ceil((double)timeoutMs_ / tickDurationInMs_);

for (int i = 0; i < blankPartitions + 1; i++) {

std::set<MessageId> msgIds;

timePartitions.push_back(msgIds);

}

I think It's just a mistake. I guess the original author intended to do:

 int blankPartitions = (timeoutMs_ / tickDurationInMs_) + ((timeoutMs_ % tickDurationInMs_ == 0) ? 0 : 1);
 for (int i = 0; i < blankPartitions; i++) {

saosir · 2021-01-05T15:05:13Z

Agree with you, java-client has same problem:

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

Lines 116 to 119 in afe6af2

    
           int blankPartitions = (int)Math.ceil((double)this.ackTimeoutMillis / this.tickDurationInMs); 
        
           for (int i = 0; i < blankPartitions + 1; i++) { 
        
               timePartitions.add(new ConcurrentOpenHashSet<>(16, 1)); 
        
           }

saosir · 2021-01-06T01:40:49Z

/pulsarbot run-failure-checks

saosir · 2021-01-06T07:13:54Z

/pulsarbot run-failure-checks

Fixes #9028 Both PartitionedConsumerImpl and ConsumerImpl have member variable unAckedMessageTrackerPtr_ (class UnAckedMessageTrackerEnabled), and PartitionedConsumerImpl is composed of ConsumerImpl. If the acknowledgement times out, they will send redeliverMessages repeatedly, MultiTopicsConsumerImpl has same problem. - add `hasParent_` field to whether there is a parent in ConsumerImpl - add unit test verify the redelivery request won't be sent repeatedly - add GTest header `gtest/gtest_prod.h` to access private members in unit tests - fix typo (cherry picked from commit 9894b99)

--- **Motivation** PR apache#9072 introduces the gtest file but does not exclude for apache-rat check. So that causes the apache-rat check to fail.

* Fix the apache-rat check --- **Motivation** PR #9072 introduces the gtest file but does not exclude for apache-rat check. So that causes the apache-rat check to fail. * Move the test to the another file * Address comments

* Fix the apache-rat check --- **Motivation** PR #9072 introduces the gtest file but does not exclude for apache-rat check. So that causes the apache-rat check to fail. * Move the test to the another file * Address comments (cherry picked from commit 0d5f070)

* Fix the apache-rat check --- **Motivation** PR apache#9072 introduces the gtest file but does not exclude for apache-rat check. So that causes the apache-rat check to fail. * Move the test to the another file * Address comments

* Fix the apache-rat check --- **Motivation** PR apache#9072 introduces the gtest file but does not exclude for apache-rat check. So that causes the apache-rat check to fail. * Move the test to the another file * Address comments (cherry picked from commit 0d5f070) (cherry picked from commit c568834)

saosir force-pushed the patch-4 branch from 13c5a2b to 8dba9ae Compare December 28, 2020 12:38

codelipenghui requested review from merlimat, rdhabalia, jiazhai and sijie December 28, 2020 12:48

codelipenghui assigned saosir Dec 28, 2020

codelipenghui added the release/2.7.1 label Dec 28, 2020

codelipenghui added this to the 2.8.0 milestone Dec 28, 2020

BewareMyPower requested changes Dec 28, 2020

View reviewed changes

saosir force-pushed the patch-4 branch from 8dba9ae to 9dc151b Compare December 31, 2020 08:20

saosir force-pushed the patch-4 branch from 53b9da8 to 2e4a54e Compare January 5, 2021 07:39

BewareMyPower requested changes Jan 5, 2021

View reviewed changes

BewareMyPower approved these changes Jan 5, 2021

View reviewed changes

saosir added 9 commits January 6, 2021 11:18

[C++] Fix Consumer send redeliverMessages repeatedly

85abe1a

fix typo

508a119

add unit test for consumer redelivery message

b859386

fix test case testUnAckedMessageTrackerEnabledCumulativeAck

d733316

fix code style

e58ce1d

fix test case fail

3dfa9fd

add GTest header file gtest/gtest_prod.h

c48634c

ignore gtest_prod.h license header check

147bd90

adjust the timeout period of test cases

0eee0a6

saosir force-pushed the patch-4 branch from 0bfb7c8 to 0eee0a6 Compare January 6, 2021 03:18

saosir mentioned this pull request Jan 6, 2021

Fix UnAckedMessageTracker timeout send redelivery messages #9134

Closed

sijie approved these changes Jan 6, 2021

View reviewed changes

sijie merged commit 9894b99 into apache:master Jan 6, 2021

saosir deleted the patch-4 branch January 6, 2021 11:01

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

saosir mentioned this pull request Jan 11, 2021

[C++] UnAckedMessageTracker handles batch message incorrectly #9170

Merged

zymap added a commit to zymap/pulsar that referenced this pull request Mar 5, 2021

Fix the apache-rat check

b2dd30b

--- **Motivation** PR apache#9072 introduces the gtest file but does not exclude for apache-rat check. So that causes the apache-rat check to fail.

zymap mentioned this pull request Mar 5, 2021

Remove gtest file from cpp library #9816

Merged

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

[C++] Fix Consumer send redeliverMessages repeatedly #9072

[C++] Fix Consumer send redeliverMessages repeatedly #9072

saosir commented Dec 28, 2020 •

edited

Loading

codelipenghui commented Dec 28, 2020

BewareMyPower left a comment

saosir commented Dec 29, 2020 •

edited

Loading

BewareMyPower commented Dec 29, 2020

sijie commented Jan 4, 2021

BewareMyPower commented Jan 5, 2021

saosir commented Jan 5, 2021

BewareMyPower commented Jan 5, 2021

saosir commented Jan 5, 2021 •

edited

Loading

BewareMyPower left a comment

saosir commented Jan 5, 2021 •

edited

Loading

BewareMyPower commented Jan 5, 2021

saosir commented Jan 5, 2021 •

edited

Loading

BewareMyPower commented Jan 5, 2021

saosir commented Jan 5, 2021

BewareMyPower commented Jan 5, 2021

saosir commented Jan 5, 2021 •

edited

Loading

BewareMyPower commented Jan 5, 2021

saosir commented Jan 5, 2021

saosir commented Jan 6, 2021

saosir commented Jan 6, 2021

[C++] Fix Consumer send redeliverMessages repeatedly #9072

[C++] Fix Consumer send redeliverMessages repeatedly #9072

Conversation

saosir commented Dec 28, 2020 • edited Loading

Motivation

Modifications

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

Documentation

codelipenghui commented Dec 28, 2020

BewareMyPower left a comment

Choose a reason for hiding this comment

saosir commented Dec 29, 2020 • edited Loading

BewareMyPower commented Dec 29, 2020

sijie commented Jan 4, 2021

BewareMyPower commented Jan 5, 2021

saosir commented Jan 5, 2021

BewareMyPower commented Jan 5, 2021

saosir commented Jan 5, 2021 • edited Loading

BewareMyPower left a comment

Choose a reason for hiding this comment

saosir commented Jan 5, 2021 • edited Loading

BewareMyPower commented Jan 5, 2021

saosir commented Jan 5, 2021 • edited Loading

BewareMyPower commented Jan 5, 2021

saosir commented Jan 5, 2021

BewareMyPower commented Jan 5, 2021

saosir commented Jan 5, 2021 • edited Loading

BewareMyPower commented Jan 5, 2021

saosir commented Jan 5, 2021

saosir commented Jan 6, 2021

saosir commented Jan 6, 2021

saosir commented Dec 28, 2020 •

edited

Loading

saosir commented Dec 29, 2020 •

edited

Loading

saosir commented Jan 5, 2021 •

edited

Loading

saosir commented Jan 5, 2021 •

edited

Loading

saosir commented Jan 5, 2021 •

edited

Loading

saosir commented Jan 5, 2021 •

edited

Loading