You are viewing a plain text version of this content. The canonical link for it is here.

Posted to commits@pulsar.apache.org by "zhanglistar (via GitHub)" <gi...@apache.org> on 2026/03/08 10:47:35 UTC

[PR] [Test] Fix flaky test: ReaderSeekTest.testHasMessageAvailableAfterSeekToEnd [pulsar-client-cpp]

zhanglistar opened a new pull request, #549:
URL: https://github.com/apache/pulsar-client-cpp/pull/549

   <!--
   ### Contribution Checklist
     
     - PR title format should be *[type][component] summary*. For details, see *[Guideline - Pulsar PR Naming Convention](https://docs.google.com/document/d/1d8Pw6ZbWk-_pCKdOmdvx9rnhPiyuxwq60_TrD68d7BA/edit#heading=h.trs9rsex3xom)*. 
   
     - Fill out the template below to describe the changes contributed by the pull request. That will give reviewers the context they need to do the review.
     
     - Each pull request should address only one issue, not mix up code from multiple issues.
     
     - Each commit in the pull request has a meaningful commit message
   
     - Once all items of the checklist are addressed, remove the above text and this checklist, leaving only the filled out template below.
   -->
   
   Fixes #<xyz>
   
   ### Motivation
   
   When a reader seeks to `MessageId::latest()`, the broker may close the consumer and trigger a reconnect. In `seekAsyncInternal`, when the seek succeeds but the connection is already expired or `reconnectionPending_` is true, we only set `seekStatus_ = COMPLETED` and did not clear the local receive queue or update `startMessageId_`. As a result, `incomingMessages_` still held prefetched messages from before the seek until `connectionOpened()` later called `clearReceiveQueue()`. During that window, `hasMessageAvailable()` saw `!incomingMessages_.empty()` and returned true, causing the flaky test `ReaderSeekTest.testHasMessageAvailableAfterSeekToEnd`.
   
   ### Modifications
   
   - In `ConsumerImpl::seekAsyncInternal`, when seek result is OK and `getCnx().expired() || reconnectionPending_` (reconnection path), we now perform the same local state updates as the non-reconnection path:
     - `ackGroupingTrackerPtr_->flushAndClean()`
     - `incomingMessages_.clear()`
     - If `lastSeekArg_` holds a `MessageId`, set `startMessageId_` from it
   
   So after the seek is acknowledged, stale prefetched messages are cleared immediately and `hasMessageAvailable()` no longer returns true from old queue content before reconnect completes.
   
   ### Verifying this change
   
   - [ ] Make sure that the change passes the CI checks.
   
   This change is already covered by existing tests:
   
   - `ReaderSeekTest.testHasMessageAvailableAfterSeekToEnd` (parameterized) – asserts that after seek to latest, `hasMessageAvailable()` becomes false, then after producing a new message it becomes true and the message can be read; the fix removes the flakiness when the broker triggers reconnect on seek-to-end.
   
   ### Documentation
   
   - [x] `doc-not-needed`
   
   Bug fix in C++ client implementation only; no API or behavior contract change and no user-facing docs update required.


-- 
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-unsubscribe@pulsar.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

Re: [PR] [fix][cpp-client] Pass seek error to callback when SEEK command fails [pulsar-client-cpp]

Posted by "zhanglistar (via GitHub)" <gi...@apache.org>.

zhanglistar commented on code in PR #549:
URL: https://github.com/apache/pulsar-client-cpp/pull/549#discussion_r2999360526


##########
lib/ConsumerImpl.cc:
##########
@@ -1844,9 +1846,8 @@ void ConsumerImpl::seekAsyncInternal(long requestId, const SharedBuffer& seek, c
                 LockGuard lock{mutex_};
                 seekStatus_ = SeekStatus::NOT_STARTED;
                 lastSeekArg_ = previousLastSeekArg;
-                executor_->postWork([self, callback{std::exchange(seekCallback_, std::nullopt).value()}]() {
-                    callback(ResultOk);
-                });
+                executor_->postWork([self, callback{std::exchange(seekCallback_, std::nullopt).value()},
+                                     result]() { callback(result); });

Review Comment:
   Yes, done.



-- 
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-unsubscribe@pulsar.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org