BewareMyPower commented on code in PR #16943:
URL: https://github.com/apache/pulsar/pull/16943#discussion_r950028292
##########
pulsar-client-cpp/tests/ConsumerTest.cc:
##########
@@ -763,4 +763,36 @@ TEST(ConsumerTest, testPartitionsWithCloseUnblock) {
thread.join();
}
+TEST(ConsumerTest, testGetLastMessageIdBlockWhenConnectionDisconnected) {
+ int operationTimeout = 10;
Review Comment:
10 seconds might be too long for a single test, maybe you can reduce it to 5
seconds (just a suggestion).
##########
pulsar-client-cpp/lib/ConsumerImpl.cc:
##########
@@ -1326,8 +1336,26 @@ void
ConsumerImpl::getLastMessageIdAsync(BrokerGetLastMessageIdCallback callback
callback(ResultUnsupportedVersionError, MessageId());
}
} else {
- LOG_ERROR(getName() << " Client Connection not ready for Consumer");
- callback(ResultNotConnected, MessageId());
+ TimeDuration next = std::min(remainTime, backoff->next());
+ if (next.total_milliseconds() <= 0) {
+ LOG_ERROR(getName() << " Client Connection not ready for
Consumer");
+ callback(ResultNotConnected, MessageId());
+ return;
+ }
+ remainTime -= next;
+
+ timer->expires_from_now(next);
+
+ timer->async_wait([this, backoff, remainTime, timer, next,
+ callback](const boost::system::error_code& ec) ->
void {
+ if (ec) {
+ LOG_DEBUG(getName() << " Get last message id operation was
cancelled, code[" << ec << "].");
+ return;
+ }
Review Comment:
`if (ec)` (in short for `if (ec != 0)`) doesn't always mean the operation is
cancelled. See how we process the error code in
https://github.com/apache/pulsar/blob/3ecf34abf99afd7c53a681ab82d5bcec2bc499a1/pulsar-client-cpp/lib/ClientConnection.cc#L594-L598
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscr...@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
