[Bug] Immediate seek after a seek will fail with NotConnected

[BewareMyPower]

Search before asking

• I searched in the issues and found nothing similar.

Version

master

Minimal reproduce step

TEST(ConsumerTest, testSeekAfterSeek) {
    const auto topic = "test-seek-after-seek-" + std::to_string(time(nullptr));
    const auto subscription = "sub";
    Client client(lookupUrl);
    Consumer consumer;
    ASSERT_EQ(ResultOk, client.subscribe(topic, subscription, consumer));
    ASSERT_EQ(ResultOk, consumer.seek(TimeUtils::currentTimeMillis()));
    ASSERT_EQ(ResultOk, consumer.seek(MessageId::earliest()));
    ASSERT_EQ(ResultOk, consumer.seek(TimeUtils::currentTimeMillis()));
    client.close();
}

Run the test above

What did you expect to see?

All seek calls should succeed

What did you see instead?

Expected equality of these values:
  ResultOk
    Which is: Ok
  consumer.seek(MessageId::earliest())
    Which is: NotConnected

Anything else?

No response

Are you willing to submit a PR?

• I'm willing to submit a PR!

[zhanglistar]

I got this, not same as yours.

Running main() from /work/pulsar-client-cpp/vcpkg/buildtrees/gtest/src/v1.16.0-a2e0b328c1.clean/googletest/src/gtest_main.cc
[==========] Running 1 test from 1 test suite.
[----------] Global test environment set-up.
[----------] 1 test from ConsumerTest
[ RUN      ] ConsumerTest.testSeekAfterSeek
2026-01-16 17:02:08.911 INFO  [139817330392064] Client:86 | Subscribing on Topic :test-seek-after-seek-1768554128
2026-01-16 17:02:08.911 INFO  [139817330392064] ClientConnection:196 | [<none> -> pulsar://localhost:6650] Create ClientConnection, timeout=10000
2026-01-16 17:02:08.911 INFO  [139817330392064] ConnectionPool:125 | Created connection for pulsar://localhost:6650-pulsar://localhost:6650-0
2026-01-16 17:02:08.912 INFO  [139817330243328] ClientConnection:416 | [127.0.0.1:53344 -> 127.0.0.1:6650] Connected to broker
2026-01-16 17:02:08.913 INFO  [139817330243328] HandlerBase:112 | [persistent://public/default/test-seek-after-seek-1768554128, sub, 0] Getting connection from pool
2026-01-16 17:02:08.913 INFO  [139817330243328] BinaryProtoLookupService:85 | Lookup response for persistent://public/default/test-seek-after-seek-1768554128, lookup-broker-url pulsar://localhost:6650, from [127.0.0.1:53344 -> 127.0.0.1:6650] 
2026-01-16 17:02:08.923 INFO  [139817330243328] ConsumerImpl:337 | [persistent://public/default/test-seek-after-seek-1768554128, sub, 0] Created consumer on broker [127.0.0.1:53344 -> 127.0.0.1:6650] 
2026-01-16 17:02:08.923 INFO  [139817330243328] HandlerBase:134 | Finished connecting to broker after 9 ms
2026-01-16 17:02:08.923 INFO  [139817330392064] ConsumerImpl:1786 | [persistent://public/default/test-seek-after-seek-1768554128, sub, 0] Seeking subscription to 1768554128923
2026-01-16 17:02:08.923 INFO  [139817330243328] ConsumerImpl:1334 | Broker notification of Closed consumer: 0
2026-01-16 17:02:08.923 INFO  [139817330243328] HandlerBase:186 | [persistent://public/default/test-seek-after-seek-1768554128, sub, 0] Schedule reconnection in 0.1 s
2026-01-16 17:02:08.934 INFO  [139817330243328] ConsumerImpl:1801 | [persistent://public/default/test-seek-after-seek-1768554128, sub, 0] Delay the seek future until the reconnection is done
2026-01-16 17:02:09.023 INFO  [139817330243328] HandlerBase:112 | [persistent://public/default/test-seek-after-seek-1768554128, sub, 0] Getting connection from pool
2026-01-16 17:02:09.023 INFO  [139817330243328] BinaryProtoLookupService:85 | Lookup response for persistent://public/default/test-seek-after-seek-1768554128, lookup-broker-url pulsar://localhost:6650, from [127.0.0.1:53344 -> 127.0.0.1:6650] 
2026-01-16 17:02:09.024 INFO  [139817330243328] ConsumerImpl:337 | [persistent://public/default/test-seek-after-seek-1768554128, sub, 0] Created consumer on broker [127.0.0.1:53344 -> 127.0.0.1:6650] 
2026-01-16 17:02:09.024 INFO  [139817330243328] HandlerBase:134 | Finished connecting to broker after 0 ms
2026-01-16 17:02:09.024 INFO  [139817330392064] ConsumerImpl:1786 | [persistent://public/default/test-seek-after-seek-1768554128, sub, 0] Seeking subscription to (-1,-1,-1,-1)
2026-01-16 17:02:09.024 INFO  [139817330243328] ConsumerImpl:1334 | Broker notification of Closed consumer: 0
2026-01-16 17:02:09.024 INFO  [139817330243328] HandlerBase:186 | [persistent://public/default/test-seek-after-seek-1768554128, sub, 0] Schedule reconnection in 0.1 s
2026-01-16 17:02:09.026 INFO  [139817330243328] ConsumerImpl:1801 | [persistent://public/default/test-seek-after-seek-1768554128, sub, 0] Delay the seek future until the reconnection is done
2026-01-16 17:02:09.124 INFO  [139817330243328] HandlerBase:112 | [persistent://public/default/test-seek-after-seek-1768554128, sub, 0] Getting connection from pool
2026-01-16 17:02:09.125 INFO  [139817330243328] BinaryProtoLookupService:85 | Lookup response for persistent://public/default/test-seek-after-seek-1768554128, lookup-broker-url pulsar://localhost:6650, from [127.0.0.1:53344 -> 127.0.0.1:6650] 
2026-01-16 17:02:09.125 INFO  [139817330243328] ConsumerImpl:337 | [persistent://public/default/test-seek-after-seek-1768554128, sub, 0] Created consumer on broker [127.0.0.1:53344 -> 127.0.0.1:6650] 
2026-01-16 17:02:09.125 INFO  [139817330243328] HandlerBase:134 | Finished connecting to broker after 0 ms
2026-01-16 17:02:09.125 INFO  [139817330392064] ConsumerImpl:1786 | [persistent://public/default/test-seek-after-seek-1768554128, sub, 0] Seeking subscription to 1768554129125
2026-01-16 17:02:09.126 INFO  [139817330243328] ConsumerImpl:1334 | Broker notification of Closed consumer: 0
2026-01-16 17:02:09.126 INFO  [139817330243328] HandlerBase:186 | [persistent://public/default/test-seek-after-seek-1768554128, sub, 0] Schedule reconnection in 0.1 s
2026-01-16 17:02:09.127 INFO  [139817330243328] ConsumerImpl:1801 | [persistent://public/default/test-seek-after-seek-1768554128, sub, 0] Delay the seek future until the reconnection is done
2026-01-16 17:02:09.226 INFO  [139817330243328] HandlerBase:112 | [persistent://public/default/test-seek-after-seek-1768554128, sub, 0] Getting connection from pool
2026-01-16 17:02:09.226 INFO  [139817330243328] BinaryProtoLookupService:85 | Lookup response for persistent://public/default/test-seek-after-seek-1768554128, lookup-broker-url pulsar://localhost:6650, from [127.0.0.1:53344 -> 127.0.0.1:6650] 
2026-01-16 17:02:09.226 INFO  [139817330243328] ConsumerImpl:337 | [persistent://public/default/test-seek-after-seek-1768554128, sub, 0] Created consumer on broker [127.0.0.1:53344 -> 127.0.0.1:6650] 
2026-01-16 17:02:09.226 INFO  [139817330243328] HandlerBase:134 | Finished connecting to broker after 0 ms
2026-01-16 17:02:09.226 INFO  [139817330392064] ClientImpl:686 | Closing Pulsar client with 0 producers and 1 consumers
2026-01-16 17:02:09.226 INFO  [139817330392064] ConsumerImpl:1361 | [persistent://public/default/test-seek-after-seek-1768554128, sub, 0] Closing consumer for topic persistent://public/default/test-seek-after-seek-1768554128
2026-01-16 17:02:09.226 INFO  [139817330243328] ConsumerImpl:1345 | [persistent://public/default/test-seek-after-seek-1768554128, sub, 0] Closed consumer 0
2026-01-16 17:02:09.227 INFO  [139817304983296] ClientConnection:1326 | [127.0.0.1:53344 -> 127.0.0.1:6650] Connection disconnected (refCnt: 1)
2026-01-16 17:02:09.227 INFO  [139817304983296] ClientConnection:288 | [127.0.0.1:53344 -> 127.0.0.1:6650] Destroyed connection to pulsar://localhost:6650-0
[       OK ] ConsumerTest.testSeekAfterSeek (315 ms)
[----------] 1 test from ConsumerTest (315 ms total)

[----------] Global test environment tear-down
[==========] 1 test from 1 test suite ran. (316 ms total)
[  PASSED  ] 1 test.

[BewareMyPower]

It's a race condition because before #533, the seek callback can be triggered before setCnx(cnx) is called (when receiving the subscribe response).