pulsar
pulsar copied to clipboard
[fix][client-cpp] Fix getLastMessageIdAsync returns ResultNotConnected after seek
Fixes #16761
Motivation
When C++ client calls ConsumerImpl::getLastMessageIdAsync, if the connection is not established, the callback will complete with ResultNotConnected.
Modifications
- Wait until the connection is established while getting the last message id.
Verifying this change
Here is the test for the backoff of ConsumerImpl::getLastMessageIdAsync:
Client client(invalid_url);
Reader reader;
client.createReader(topic, MessageId::latest(), {}, reader);
bool t;
reader.hasMessageAvailable(t);
Here is the output log:
2022-08-04 19:10:44.081 WARN [0x7000064fa000] ConsumerImpl:1391 | [persistent://public/default/reader-test-receive-after-seek-1659611443, reader-c868ade2fb, 0] Could not get connection while getLastMessageId -- Will try again in 100 ms
2022-08-04 19:10:44.084 INFO [0x7000064fa000] ConsumerImpl:224 | [persistent://public/default/reader-test-receive-after-seek-1659611443, reader-c868ade2fb, 0] Created consumer on broker [127.0.0.1:52918 -> 127.0.0.1:6650]
2022-08-04 19:10:44.268 WARN [0x7000064fa000] ConsumerImpl:1391 | [persistent://public/default/reader-test-receive-after-seek-1659611443, reader-c868ade2fb, 0] Could not get connection while getLastMessageId -- Will try again in 186 ms
2022-08-04 19:10:44.649 WARN [0x7000064fa000] ConsumerImpl:1391 | [persistent://public/default/reader-test-receive-after-seek-1659611443, reader-c868ade2fb, 0] Could not get connection while getLastMessageId -- Will try again in 376 ms
2022-08-04 19:10:45.380 WARN [0x7000064fa000] ConsumerImpl:1391 | [persistent://public/default/reader-test-receive-after-seek-1659611443, reader-c868ade2fb, 0] Could not get connection while getLastMessageId -- Will try again in 728 ms
2022-08-04 19:10:46.985 WARN [0x7000064fa000] ConsumerImpl:1391 | [persistent://public/default/reader-test-receive-after-seek-1659611443, reader-c868ade2fb, 0] Could not get connection while getLastMessageId -- Will try again in 1600 ms
2022-08-04 19:10:49.933 WARN [0x7000064fa000] ConsumerImpl:1391 | [persistent://public/default/reader-test-receive-after-seek-1659611443, reader-c868ade2fb, 0] Could not get connection while getLastMessageId -- Will try again in 2944 ms
2022-08-04 19:10:55.890 WARN [0x7000064fa000] ConsumerImpl:1391 | [persistent://public/default/reader-test-receive-after-seek-1659611443, reader-c868ade2fb, 0] Could not get connection while getLastMessageId -- Will try again in 5952 ms
2022-08-04 19:11:07.925 WARN [0x7000064fa000] ConsumerImpl:1391 | [persistent://public/default/reader-test-receive-after-seek-1659611443, reader-c868ade2fb, 0] Could not get connection while getLastMessageId -- Will try again in 12032 ms
2022-08-04 19:11:14.011 WARN [0x7000064fa000] ConsumerImpl:1391 | [persistent://public/default/reader-test-receive-after-seek-1659611443, reader-c868ade2fb, 0] Could not get connection while getLastMessageId -- Will try again in 6082 ms
2022-08-04 19:11:14.011 ERROR [0x7000064fa000] ConsumerImpl:1376 | [persistent://public/default/reader-test-receive-after-seek-1659611443, reader-c868ade2fb, 0] Client Connection not ready for Consumer
Documentation
Check the box below or label this PR directly.
Need to update docs?
-
[ ]
doc-required(Your PR needs to update docs and you will update later) -
[x]
doc-not-needed(Please explain why) -
[ ]
doc(Your PR contains doc changes) -
[ ]
doc-complete(Docs have been already added)