From 3b4de8c7a8f2a95643ec04b94714708ab02f520d Mon Sep 17 00:00:00 2001 From: cbcbq <464015046@qq.com> Date: Fri, 30 Oct 2026 17:47:05 +0800 Subject: [PATCH] [fix][client] Fix HTTP lookup executor leak --- lib/HTTPLookupService.cc | 6 +++ lib/HTTPLookupService.h | 4 ++ tests/LookupServiceTest.cc | 96 ++++++++++++++++++++++++++++++++++++++ tests/PulsarFriend.h | 2 + 4 files changed, 108 insertions(+) diff --git a/lib/HTTPLookupService.cc b/lib/HTTPLookupService.cc index cdccd935..fbb46bfb 100644 --- a/lib/HTTPLookupService.cc +++ b/lib/HTTPLookupService.cc @@ -64,6 +64,12 @@ HTTPLookupService::HTTPLookupService(const ServiceInfo &serviceInfo, tlsValidateHostname_(clientConfiguration.isValidateHostName()), httpLookupAuthAllowRedirect_(clientConfiguration.isHttpLookupAuthAllowRedirect()) {} +HTTPLookupService::~HTTPLookupService() { + // The last reference can be released by an HTTP lookup task on this executor. + // Stop without waiting for the current thread to exit. + executorProvider_->close(0); +} + auto HTTPLookupService::getBroker(const TopicName &topicName) -> LookupResultFuture { LookupResultPromise promise; diff --git a/lib/HTTPLookupService.h b/lib/HTTPLookupService.h index 96c252ce..e301fd60 100644 --- a/lib/HTTPLookupService.h +++ b/lib/HTTPLookupService.h @@ -35,6 +35,8 @@ using NamespaceTopicsPromisePtr = std::shared_ptr; using GetSchemaPromise = Promise; class HTTPLookupService : public LookupService, public std::enable_shared_from_this { + friend class LookupServiceTest; + enum RequestType : uint8_t { Lookup, @@ -72,6 +74,8 @@ class HTTPLookupService : public LookupService, public std::enable_shared_from_t public: HTTPLookupService(const ServiceInfo& serviceInfo, const ClientConfiguration& config); + ~HTTPLookupService() override; + LookupResultFuture getBroker(const TopicName& topicName) override; Future getPartitionMetadataAsync(const TopicNamePtr& topicName) override; diff --git a/tests/LookupServiceTest.cc b/tests/LookupServiceTest.cc index e98d972e..0696328c 100644 --- a/tests/LookupServiceTest.cc +++ b/tests/LookupServiceTest.cc @@ -31,6 +31,7 @@ #include "HttpHelper.h" #include "PulsarFriend.h" #include "PulsarWrapper.h" +#include "WaitUtils.h" #include "lib/AtomicSharedPtr.h" #include "lib/BinaryProtoLookupService.h" #include "lib/ClientConnection.h" @@ -70,6 +71,16 @@ class LookupServiceTest : public ::testing::TestWithParam { isEmpty(*service.namespaceLookupCache_) && isEmpty(*service.getSchemaCache_); } + static ExecutorServicePtr getExecutor(HTTPLookupService& service) { + return service.executorProvider_->get(); + } + + static std::shared_ptr getHttpLookup(Client& client) { + auto retryable = std::static_pointer_cast( + PulsarFriend::getLookupService(client)); + return std::static_pointer_cast(retryable->lookupService_); + } + protected: std::string serviceUrl_; Client client_{httpLookupUrl}; @@ -79,6 +90,91 @@ class LookupServiceTest : public ::testing::TestWithParam { using namespace pulsar; +// These lifecycle tests need no broker. Observe executor destruction rather than process-wide thread counts. +TEST(LookupServiceTest, httpLookupDestructorClosesExecutor) { + const ServiceInfo serviceInfo{"http://localhost:8080"}; + for (int i = 0; i < 5; ++i) { + auto lookup = std::make_shared(serviceInfo, ClientConfiguration{}); + auto executor = LookupServiceTest::getExecutor(*lookup); + std::weak_ptr weakExecutor = executor; + auto started = std::make_shared>(); + auto future = started->get_future(); + executor->postWork([started] { started->set_value(); }); + ASSERT_EQ(std::future_status::ready, future.wait_for(std::chrono::seconds(1))); + + // No test-owned strong reference may keep the executor alive. + executor.reset(); + lookup.reset(); + waitUntil(std::chrono::seconds(1), [&] { return weakExecutor.expired(); }); + ASSERT_TRUE(weakExecutor.expired()); + } +} + +TEST(LookupServiceTest, httpLookupDestructorOnWorker) { + const ServiceInfo serviceInfo{"http://localhost:8080"}; + auto lookup = std::make_shared(serviceInfo, ClientConfiguration{}); + std::weak_ptr weakLookup = lookup; + auto executor = LookupServiceTest::getExecutor(*lookup); + std::weak_ptr weakExecutor = executor; + auto release = std::make_shared>(); + auto gate = release->get_future().share(); + auto started = std::make_shared>(); + auto startedFuture = started->get_future(); + executor->postWork([lookup, gate, started] { + (void)lookup; // Keep the service alive throughout this task. + started->set_value(); + // Bound the wait even if the test fails before releasing the worker. + gate.wait_for(std::chrono::seconds(5)); + }); + for (int i = 0; i < 8; ++i) { + // Match the strong service reference held by each real HTTP lookup task. + executor->postWork([lookup] { (void)lookup; }); + } + auto completed = std::make_shared>(); + auto completedFuture = completed->get_future(); + executor->postWork([lookup, completed] { + (void)lookup; + completed->set_value(); + }); + + ASSERT_EQ(std::future_status::ready, startedFuture.wait_for(std::chrono::seconds(1))); + executor.reset(); + lookup.reset(); + EXPECT_FALSE(weakLookup.expired()); + release->set_value(); + ASSERT_EQ(std::future_status::ready, completedFuture.wait_for(std::chrono::seconds(1))); + // The final queued task releases the service on its own worker. A default close() + // would wait for itself for 3 seconds; an infinite wait would deadlock. + waitUntil(std::chrono::seconds(1), [&] { return weakExecutor.expired(); }); + EXPECT_TRUE(weakLookup.expired()); + EXPECT_TRUE(weakExecutor.expired()); +} + +TEST(LookupServiceTest, httpLookupExecutorClosesWithClient) { + for (bool explicitlyClose : {false, true}) { + std::weak_ptr weakExecutor; + { + Client client("http://localhost:8080"); + auto lookup = LookupServiceTest::getHttpLookup(client); + auto executor = LookupServiceTest::getExecutor(*lookup); + weakExecutor = executor; + auto started = std::make_shared>(); + auto future = started->get_future(); + executor->postWork([started] { started->set_value(); }); + ASSERT_EQ(std::future_status::ready, future.wait_for(std::chrono::seconds(1))); + executor.reset(); + lookup.reset(); + if (explicitlyClose) { + ASSERT_EQ(ResultOk, client.close()); + } + } + // ClientImpl -> RetryableLookupService -> HTTPLookupService destruction must + // close the fourth provider, including when Client::close() was not called. + waitUntil(std::chrono::seconds(1), [&] { return weakExecutor.expired(); }); + ASSERT_TRUE(weakExecutor.expired()); + } +} + TEST(LookupServiceTest, basicLookup) { ExecutorServiceProviderPtr service = std::make_shared(1); std::string url = "pulsar://localhost:6650"; diff --git a/tests/PulsarFriend.h b/tests/PulsarFriend.h index 02447437..677ce7aa 100644 --- a/tests/PulsarFriend.h +++ b/tests/PulsarFriend.h @@ -129,6 +129,8 @@ class PulsarFriend { static std::shared_ptr getClientImplPtr(Client client) { return client.impl_; } + static LookupServicePtr getLookupService(const Client& client) { return client.impl_->getLookup(""); } + static auto getProducers(const Client& client) -> decltype(ClientImpl::producers_)& { return getClientImplPtr(client)->producers_; }