fix

Author: zhanglistar

lib/ConsumerImpl.cc

@@ -334,11 +334,11 @@ Result ConsumerImpl::handleCreateConsumer(const ClientConnectionPtr& cnx, Result
                 return ResultAlreadyClosed;
             }
 
+            mutexLock.unlock();
+            LOG_INFO(getName() << "Created consumer on broker " << cnx->cnxString());
             incomingMessages_.clear();
             possibleSendToDeadLetterTopicMessages_.clear();
             backoff_.reset();
-            mutexLock.unlock();
-            LOG_INFO(getName() << "Created consumer on broker " << cnx->cnxString());
             if (!messageListener_ && config_.getReceiverQueueSize() == 0) {
                 // Complicated logic since we don't have a isLocked() function for mutex
                 if (waitingForZeroQueueSizeMessage) {
@@ -1846,9 +1846,10 @@ 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);
+                    });
             }
         });
 }

tests/ConsumerSeekTest.cc

@@ -258,6 +258,31 @@ TEST_F(ConsumerSeekTest, testReconnectionSlow) {
     client.close();
 }
 
+TEST_F(ConsumerSeekTest, testSeekFailureIsPropagated) {
+    using namespace std::chrono_literals;
+
+    Client client(lookupUrl, ClientConfiguration().setOperationTimeoutSeconds(1));
+    Consumer consumer;
+    ASSERT_EQ(ResultOk, client.subscribe("testSeekFailureIsPropagated", "sub", consumer));
+
+    auto connection = *PulsarFriend::getConnections(client).begin();
+    auto mockServer = std::make_shared<MockServer>(connection);
+    connection->attachMockServer(mockServer);
+    mockServer->setRequestDelay({{"SEEK", 5000}});
+
+    std::promise<Result> promise;
+    auto future = promise.get_future();
+    consumer.seekAsync(MessageId::earliest(), [&promise](Result result) { promise.set_value(result); });
+
+    // Cancel the mocked SEEK success so request completes with timeout.
+    ASSERT_GE(mockServer->close(), 1);
+
+    ASSERT_EQ(future.wait_for(5s), std::future_status::ready);
+    ASSERT_EQ(future.get(), ResultTimeout);
+
+    client.close();
+}
+
 INSTANTIATE_TEST_SUITE_P(Pulsar, ConsumerSeekTest, ::testing::Values(true, false));
 
 }  // namespace pulsar