LookupServiceTest.cc

/**
 * Licensed to the Apache Software Foundation (ASF) under one
 * or more contributor license agreements.  See the NOTICE file
 * distributed with this work for additional information
 * regarding copyright ownership.  The ASF licenses this file
 * to you under the Apache License, Version 2.0 (the
 * "License"); you may not use this file except in compliance
 * with the License.  You may obtain a copy of the License at
 *
 *   http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing,
 * software distributed under the License is distributed on an
 * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
 * KIND, either express or implied.  See the License for the
 * specific language governing permissions and limitations
 * under the License.
 */
#include <gtest/gtest.h>
#include <pulsar/Authentication.h>
#include <pulsar/Client.h>

#include <algorithm>
#include <boost/exception/all.hpp>
#include <chrono>
#include <future>
#include <memory>
#include <stdexcept>

#include "HttpHelper.h"
#include "PulsarFriend.h"
#include "PulsarWrapper.h"
#include "lib/BinaryProtoLookupService.h"
#include "lib/ClientConnection.h"
#include "lib/ConnectionPool.h"
#include "lib/Future.h"
#include "lib/HTTPLookupService.h"
#include "lib/LogUtils.h"
#include "lib/RetryableLookupService.h"
#include "lib/TimeUtils.h"
#include "lib/Utils.h"

DECLARE_LOG_OBJECT()

static std::string binaryLookupUrl = "pulsar://localhost:6650";
static std::string httpLookupUrl = "http://localhost:8080";

extern std::string unique_str();

namespace pulsar {

class LookupServiceTest : public ::testing::TestWithParam<std::string> {
   public:
    void SetUp() override {
        serviceUrl_ = GetParam();
        client_ = Client{serviceUrl_};
    }
    void TearDown() override { client_.close(); }

    template <typename T>
    static bool isEmpty(const RetryableOperationCache<T>& cache) {
        std::lock_guard<std::mutex> lock{cache.mutex_};
        return cache.operations_.empty();
    }

    static size_t isEmpty(const RetryableLookupService& service) {
        return isEmpty(*service.lookupCache_) && isEmpty(*service.partitionLookupCache_) &&
               isEmpty(*service.namespaceLookupCache_) && isEmpty(*service.getSchemaCache_);
    }

   protected:
    std::string serviceUrl_;
    Client client_{httpLookupUrl};
};

}  // namespace pulsar

using namespace pulsar;

TEST(LookupServiceTest, basicLookup) {
    ExecutorServiceProviderPtr service = std::make_shared<ExecutorServiceProvider>(1);
    AuthenticationPtr authData = AuthFactory::Disabled();
    std::string url = "pulsar://localhost:6650";
    ClientConfiguration conf;
    ExecutorServiceProviderPtr ioExecutorProvider_(std::make_shared<ExecutorServiceProvider>(1));
    ConnectionPool pool_(conf, ioExecutorProvider_, authData, "");
    BinaryProtoLookupService lookupService(url, pool_, conf);

    TopicNamePtr topicName = TopicName::get("topic");

    Future<Result, LookupDataResultPtr> partitionFuture = lookupService.getPartitionMetadataAsync(topicName);
    LookupDataResultPtr lookupData;
    partitionFuture.get(lookupData);
    ASSERT_TRUE(lookupData != NULL);
    ASSERT_EQ(0, lookupData->getPartitions());

    const auto topicNamePtr = TopicName::get("topic");
    auto future = lookupService.getBroker(*topicNamePtr);
    LookupService::LookupResult lookupResult;
    auto result = future.get(lookupResult);

    ASSERT_EQ(ResultOk, result);
    ASSERT_EQ(url, lookupResult.logicalAddress);
    ASSERT_EQ(url, lookupResult.physicalAddress);
}

static void testMultiAddresses(LookupService& lookupService) {
    std::vector<Result> results;
    constexpr int numRequests = 6;

    auto verifySuccessCount = [&results] {
        // Only half of them succeeded
        ASSERT_EQ(std::count(results.cbegin(), results.cend(), ResultOk), numRequests / 2);
        ASSERT_EQ(std::count(results.cbegin(), results.cend(), ResultRetryable), numRequests / 2);
    };

    for (int i = 0; i < numRequests; i++) {
        const auto topicNamePtr = TopicName::get("topic");
        LookupService::LookupResult lookupResult;
        const auto result = lookupService.getBroker(*topicNamePtr).get(lookupResult);
        LOG_INFO("getBroker [" << i << "] " << result << ", " << lookupResult);
        results.emplace_back(result);
    }
    verifySuccessCount();

    results.clear();
    for (int i = 0; i < numRequests; i++) {
        LookupDataResultPtr data;
        const auto result = lookupService.getPartitionMetadataAsync(TopicName::get("topic")).get(data);
        LOG_INFO("getPartitionMetadataAsync [" << i << "] " << result);
        results.emplace_back(result);
    }
    verifySuccessCount();

    results.clear();
    for (int i = 0; i < numRequests; i++) {
        NamespaceTopicsPtr data;
        const auto result = lookupService
                                .getTopicsOfNamespaceAsync(TopicName::get("topic")->getNamespaceName(),
                                                           CommandGetTopicsOfNamespace_Mode_PERSISTENT)
                                .get(data);
        LOG_INFO("getTopicsOfNamespaceAsync [" << i << "] " << result);
        results.emplace_back(result);
    }
    verifySuccessCount();
}

TEST(LookupServiceTest, testMultiAddresses) {
    ConnectionPool pool({}, std::make_shared<ExecutorServiceProvider>(1), AuthFactory::Disabled(), "");
    ClientConfiguration conf;
    BinaryProtoLookupService binaryLookupService("pulsar://localhost,localhost:9999", pool, conf);
    testMultiAddresses(binaryLookupService);

    // HTTPLookupService calls shared_from_this() internally, we must create a shared pointer to test
    auto httpLookupServicePtr = std::make_shared<HTTPLookupService>(
        "http://localhost,localhost:9999", ClientConfiguration{}, AuthFactory::Disabled());
    testMultiAddresses(*httpLookupServicePtr);
}
TEST(LookupServiceTest, testRetry) {
    auto executorProvider = std::make_shared<ExecutorServiceProvider>(1);
    ConnectionPool pool({}, executorProvider, AuthFactory::Disabled(), "");
    ClientConfiguration conf;

    auto lookupService = RetryableLookupService::create(
        std::make_shared<BinaryProtoLookupService>("pulsar://localhost:9999,localhost", pool, conf),
        std::chrono::seconds(30), executorProvider);
    ServiceNameResolver& serviceNameResolver = lookupService->getServiceNameResolver();

    PulsarFriend::setServiceUrlIndex(serviceNameResolver, 0);
    auto topicNamePtr = TopicName::get("lookup-service-test-retry");
    auto future1 = lookupService->getBroker(*topicNamePtr);
    LookupService::LookupResult lookupResult;
    ASSERT_EQ(ResultOk, future1.get(lookupResult));
    LOG_INFO("getBroker returns logicalAddress: " << lookupResult.logicalAddress
                                                  << ", physicalAddress: " << lookupResult.physicalAddress);

    PulsarFriend::setServiceUrlIndex(serviceNameResolver, 0);
    auto future2 = lookupService->getPartitionMetadataAsync(topicNamePtr);
    LookupDataResultPtr lookupDataResultPtr;
    ASSERT_EQ(ResultOk, future2.get(lookupDataResultPtr));
    LOG_INFO("getPartitionMetadataAsync returns " << lookupDataResultPtr->getPartitions() << " partitions");

    PulsarFriend::setServiceUrlIndex(serviceNameResolver, 0);
    auto future3 = lookupService->getTopicsOfNamespaceAsync(topicNamePtr->getNamespaceName(),
                                                            CommandGetTopicsOfNamespace_Mode_PERSISTENT);
    NamespaceTopicsPtr namespaceTopicsPtr;
    ASSERT_EQ(ResultOk, future3.get(namespaceTopicsPtr));
    LOG_INFO("getTopicPartitionName Async returns " << namespaceTopicsPtr->size() << " topics");

    ASSERT_TRUE(LookupServiceTest::isEmpty(*lookupService));
}

TEST(LookupServiceTest, testTimeout) {
    auto executorProvider = std::make_shared<ExecutorServiceProvider>(1);
    ConnectionPool pool({}, executorProvider, AuthFactory::Disabled(), "");
    ClientConfiguration conf;

    constexpr int timeoutInSeconds = 2;
    auto lookupService = RetryableLookupService::create(
        std::make_shared<BinaryProtoLookupService>("pulsar://localhost:9990,localhost:9902,localhost:9904",
                                                   pool, conf),
        std::chrono::seconds(timeoutInSeconds), executorProvider);
    auto topicNamePtr = TopicName::get("lookup-service-test-retry");

    decltype(std::chrono::high_resolution_clock::now()) startTime;
    auto beforeMethod = [&startTime] { startTime = std::chrono::high_resolution_clock::now(); };
    auto afterMethod = [&startTime](const std::string& name) {
        auto timeInterval = std::chrono::duration_cast<std::chrono::milliseconds>(
                                std::chrono::high_resolution_clock::now() - startTime)
                                .count();
        LOG_INFO(name << " took " << timeInterval << " seconds");
        ASSERT_TRUE(timeInterval >= timeoutInSeconds * 1000L);
    };

    beforeMethod();
    auto future1 = lookupService->getBroker(*topicNamePtr);
    LookupService::LookupResult lookupResult;
    ASSERT_EQ(ResultTimeout, future1.get(lookupResult));
    afterMethod("getBroker");

    beforeMethod();
    auto future2 = lookupService->getPartitionMetadataAsync(topicNamePtr);
    LookupDataResultPtr lookupDataResultPtr;
    ASSERT_EQ(ResultTimeout, future2.get(lookupDataResultPtr));
    afterMethod("getPartitionMetadataAsync");

    beforeMethod();
    auto future3 = lookupService->getTopicsOfNamespaceAsync(topicNamePtr->getNamespaceName(),
                                                            CommandGetTopicsOfNamespace_Mode_PERSISTENT);
    NamespaceTopicsPtr namespaceTopicsPtr;
    ASSERT_EQ(ResultTimeout, future3.get(namespaceTopicsPtr));
    afterMethod("getTopicsOfNamespaceAsync");

    ASSERT_TRUE(LookupServiceTest::isEmpty(*lookupService));
}

TEST_P(LookupServiceTest, basicGetNamespaceTopics) {
    Result result;

    auto nsName = NamespaceName::get("public", GetParam().substr(0, 4) + std::to_string(time(nullptr)));
    std::string topicName1 = "persistent://" + nsName->toString() + "/basicGetNamespaceTopics1";
    std::string topicName2 = "persistent://" + nsName->toString() + "/basicGetNamespaceTopics2";
    std::string topicName3 = "non-persistent://" + nsName->toString() + "/basicGetNamespaceTopics3";

    // 0. create a namespace
    auto createNsUrl = httpLookupUrl + "/admin/v2/namespaces/" + nsName->toString();
    auto res = makePutRequest(createNsUrl, "");
    ASSERT_FALSE(res != 204 && res != 409);

    // 1. trigger auto create topic
    Producer producer1;
    result = client_.createProducer(topicName1, producer1);
    ASSERT_EQ(ResultOk, result);
    Producer producer2;
    result = client_.createProducer(topicName2, producer2);
    ASSERT_EQ(ResultOk, result);
    Producer producer3;
    result = client_.createProducer(topicName3, producer3);
    ASSERT_EQ(ResultOk, result);

    // 2. verify getTopicsOfNamespace by regex mode.
    auto lookupServicePtr = PulsarFriend::getClientImplPtr(client_)->getLookup();
    auto verifyGetTopics = [&](CommandGetTopicsOfNamespace_Mode mode,
                               const std::set<std::string>& expectedTopics) {
        Future<Result, NamespaceTopicsPtr> getTopicsFuture =
            lookupServicePtr->getTopicsOfNamespaceAsync(nsName, mode);
        NamespaceTopicsPtr topicsData;
        result = getTopicsFuture.get(topicsData);
        ASSERT_EQ(ResultOk, result);
        ASSERT_TRUE(topicsData != NULL);
        std::set<std::string> actualTopics(topicsData->begin(), topicsData->end());
        ASSERT_EQ(expectedTopics, actualTopics);
    };
    verifyGetTopics(CommandGetTopicsOfNamespace_Mode_PERSISTENT, {topicName1, topicName2});
    verifyGetTopics(CommandGetTopicsOfNamespace_Mode_NON_PERSISTENT, {topicName3});
    verifyGetTopics(CommandGetTopicsOfNamespace_Mode_ALL, {topicName1, topicName2, topicName3});

    client_.close();
}

TEST_P(LookupServiceTest, testGetSchema) {
    const std::string topic = "testGetSchema" + std::to_string(time(nullptr)) + GetParam().substr(0, 4);
    std::string jsonSchema =
        R"({"type":"record","name":"cpx","fields":[{"name":"re","type":"double"},{"name":"im","type":"double"}]})";

    StringMap properties;
    properties.emplace("key1", "value1");
    properties.emplace("key2", "value2");

    ProducerConfiguration producerConfiguration;
    producerConfiguration.setSchema(SchemaInfo(SchemaType::JSON, "json", jsonSchema, properties));
    Producer producer;
    ASSERT_EQ(ResultOk, client_.createProducer(topic, producerConfiguration, producer));

    auto clientImplPtr = PulsarFriend::getClientImplPtr(client_);
    auto lookup = clientImplPtr->getLookup();

    SchemaInfo schemaInfo;
    auto future = lookup->getSchema(TopicName::get(topic));
    ASSERT_EQ(ResultOk, future.get(schemaInfo));
    ASSERT_EQ(jsonSchema, schemaInfo.getSchema());
    ASSERT_EQ(SchemaType::JSON, schemaInfo.getSchemaType());
    ASSERT_EQ(properties, schemaInfo.getProperties());
}

TEST_P(LookupServiceTest, testGetSchemaNotFound) {
    const std::string topic =
        "testGetSchemaNotFund" + std::to_string(time(nullptr)) + GetParam().substr(0, 4);

    Producer producer;
    ASSERT_EQ(ResultOk, client_.createProducer(topic, producer));

    auto clientImplPtr = PulsarFriend::getClientImplPtr(client_);
    auto lookup = clientImplPtr->getLookup();

    SchemaInfo schemaInfo;
    auto future = lookup->getSchema(TopicName::get(topic));
    ASSERT_EQ(ResultTopicNotFound, future.get(schemaInfo));
}

TEST_P(LookupServiceTest, testGetKeyValueSchema) {
    const std::string topic =
        "testGetKeyValueSchema" + std::to_string(time(nullptr)) + GetParam().substr(0, 4);
    StringMap properties;
    properties.emplace("key1", "value1");
    properties.emplace("key2", "value2");
    std::string jsonSchema =
        R"({"type":"record","name":"cpx","fields":[{"name":"re","type":"double"},{"name":"im","type":"double"}]})";
    SchemaInfo keySchema(JSON, "key-json", jsonSchema, properties);
    SchemaInfo valueSchema(JSON, "value-json", jsonSchema, properties);
    SchemaInfo keyValueSchema(keySchema, valueSchema, KeyValueEncodingType::INLINE);

    ProducerConfiguration producerConfiguration;
    producerConfiguration.setSchema(keyValueSchema);
    Producer producer;
    ASSERT_EQ(ResultOk, client_.createProducer(topic, producerConfiguration, producer));

    auto clientImplPtr = PulsarFriend::getClientImplPtr(client_);
    auto lookup = clientImplPtr->getLookup();

    SchemaInfo schemaInfo;
    auto future = lookup->getSchema(TopicName::get(topic));
    ASSERT_EQ(ResultOk, future.get(schemaInfo));
    ASSERT_EQ(keyValueSchema.getSchema(), schemaInfo.getSchema());
    ASSERT_EQ(SchemaType::KEY_VALUE, schemaInfo.getSchemaType());
    ASSERT_FALSE(schemaInfo.getProperties().empty());
}

TEST_P(LookupServiceTest, testGetSchemaByVersion) {
    const auto topic = "testGetSchemaByVersion" + unique_str() + GetParam().substr(0, 4);
    const std::string schema1 = R"({
  "type": "record",
  "name": "User",
  "namespace": "test",
  "fields": [
    {"name": "name", "type": ["null", "string"]},
    {"name": "age", "type": "int"}
  ]
})";
    const std::string schema2 = R"({
  "type": "record",
  "name": "User",
  "namespace": "test",
  "fields": [
    {"name": "age", "type": "int"},
    {"name": "name", "type": ["null", "string"]}
  ]
})";
    ProducerConfiguration producerConf1;
    producerConf1.setSchema(SchemaInfo{AVRO, "Avro", schema1});
    Producer producer1;
    ASSERT_EQ(ResultOk, client_.createProducer(topic, producerConf1, producer1));
    ProducerConfiguration producerConf2;
    producerConf2.setSchema(SchemaInfo{AVRO, "Avro", schema2});
    Producer producer2;
    ASSERT_EQ(ResultOk, client_.createProducer(topic, producerConf2, producer2));

    // Though these messages are invalid, the C++ client can send them successfully
    producer1.send(MessageBuilder().setContent("msg0").build());
    producer2.send(MessageBuilder().setContent("msg1").build());

    ConsumerConfiguration consumerConf;
    consumerConf.setSubscriptionInitialPosition(InitialPositionEarliest);
    Consumer consumer;
    ASSERT_EQ(ResultOk, client_.subscribe(topic, "sub", consumerConf, consumer));

    Message msg1;
    ASSERT_EQ(ResultOk, consumer.receive(msg1, 3000));
    Message msg2;
    ASSERT_EQ(ResultOk, consumer.receive(msg2, 3000));

    auto getSchemaInfo = [this](const std::string& topic, int64_t version) {
        std::promise<SchemaInfo> p;
        client_.getSchemaInfoAsync(topic, version, [&p](Result result, const SchemaInfo& info) {
            if (result == ResultOk) {
                p.set_value(info);
            } else {
                p.set_exception(std::make_exception_ptr(std::runtime_error(strResult(result))));
            }
        });
        return p.get_future().get();
    };
    {
        ASSERT_EQ(msg1.getLongSchemaVersion(), 0);
        const auto info = getSchemaInfo(topic, 0);
        ASSERT_EQ(info.getSchemaType(), SchemaType::AVRO);
        ASSERT_EQ(info.getSchema(), schema1);
    }
    {
        ASSERT_EQ(msg2.getLongSchemaVersion(), 1);
        const auto info = getSchemaInfo(topic, 1);
        ASSERT_EQ(info.getSchemaType(), SchemaType::AVRO);
        ASSERT_EQ(info.getSchema(), schema2);
    }
    {
        const auto info = getSchemaInfo(topic, -1);
        ASSERT_EQ(info.getSchemaType(), SchemaType::AVRO);
        ASSERT_EQ(info.getSchema(), schema2);
    }
    try {
        getSchemaInfo(topic, 2);
        FAIL();
    } catch (const std::runtime_error& e) {
        ASSERT_EQ(std::string{e.what()}, strResult(ResultTopicNotFound));
    }
    try {
        getSchemaInfo(topic + "-not-exist", 0);
        FAIL();
    } catch (const std::runtime_error& e) {
        ASSERT_EQ(std::string{e.what()}, strResult(ResultTopicNotFound));
    }

    consumer.close();
    producer1.close();
    producer2.close();
}

TEST_P(LookupServiceTest, testGetSchemaTimeout) {
    if (serviceUrl_.find("pulsar://") == std::string::npos) {
        // HTTP request timeout can only be configured with seconds
        return;
    }
    const auto topic = "lookup-service-test-get-schema-timeout";
    Producer producer;
    ProducerConfiguration producerConf;
    producerConf.setSchema(SchemaInfo(STRING, "", ""));
    ASSERT_EQ(ResultOk, client_.createProducer(topic, producerConf, producer));
    ASSERT_EQ(ResultOk, producer.send(MessageBuilder().setContent("msg").build()));
    client_.close();

    ClientConfiguration conf;
    PulsarWrapper::deref(conf).operationTimeout = std::chrono::nanoseconds(1);
    client_ = Client{serviceUrl_, conf};
    auto promise = std::make_shared<std::promise<Result>>();
    client_.getSchemaInfoAsync(topic, 0L,
                               [promise](Result result, const SchemaInfo&) { promise->set_value(result); });
    auto future = promise->get_future();
    ASSERT_EQ(future.wait_for(std::chrono::milliseconds(100)), std::future_status::ready);
    ASSERT_EQ(future.get(), ResultTimeout);
    client_.close();
}

INSTANTIATE_TEST_SUITE_P(Pulsar, LookupServiceTest, ::testing::Values(binaryLookupUrl, httpLookupUrl));

class BinaryProtoLookupServiceRedirectTestHelper : public BinaryProtoLookupService {
   public:
    BinaryProtoLookupServiceRedirectTestHelper(const std::string& serviceUrl, ConnectionPool& pool,
                                               const ClientConfiguration& clientConfiguration)
        : BinaryProtoLookupService(serviceUrl, pool, clientConfiguration) {}

    LookupResultFuture findBroker(const std::string& address, bool authoritative, const std::string& topic,
                                  size_t redirectCount) {
        return BinaryProtoLookupService::findBroker(address, authoritative, topic, redirectCount);
    }
};  // class BinaryProtoLookupServiceRedirectTestHelper

TEST(LookupServiceTest, testRedirectionLimit) {
    const auto redirect_limit = 5;
    AuthenticationPtr authData = AuthFactory::Disabled();
    ClientConfiguration conf;
    conf.setMaxLookupRedirects(redirect_limit);
    ExecutorServiceProviderPtr ioExecutorProvider_(std::make_shared<ExecutorServiceProvider>(1));
    ConnectionPool pool_(conf, ioExecutorProvider_, authData, "");
    string url = "pulsar://localhost:6650";
    BinaryProtoLookupServiceRedirectTestHelper lookupService(url, pool_, conf);

    const auto topicNamePtr = TopicName::get("topic");
    for (auto idx = 0; idx < redirect_limit + 5; ++idx) {
        auto future = lookupService.findBroker(lookupService.getServiceNameResolver().resolveHost(), false,
                                               topicNamePtr->toString(), idx);
        LookupService::LookupResult lookupResult;
        auto result = future.get(lookupResult);

        if (idx <= redirect_limit) {
            ASSERT_EQ(ResultOk, result);
            ASSERT_EQ(url, lookupResult.logicalAddress);
            ASSERT_EQ(url, lookupResult.physicalAddress);
        } else {
            ASSERT_EQ(ResultTooManyLookupRequestException, result);
        }
    }
}

class MockLookupService : public BinaryProtoLookupService {
   public:
    using BinaryProtoLookupService::BinaryProtoLookupService;

    Future<Result, LookupDataResultPtr> getPartitionMetadataAsync(const TopicNamePtr& topicName) override {
        bool expected = true;
        if (firstTime_.compare_exchange_strong(expected, false)) {
            // Trigger the retry
            LOG_INFO("Fail the lookup for " << topicName->toString() << " intentionally");
            Promise<Result, LookupDataResultPtr> promise;
            promise.setFailed(ResultRetryable);
            return promise.getFuture();
        }
        return BinaryProtoLookupService::getPartitionMetadataAsync(topicName);
    }

   private:
    std::atomic_bool firstTime_{true};
};

TEST(LookupServiceTest, testAfterClientShutdown) {
    auto client = std::make_shared<ClientImpl>("pulsar://localhost:6650", ClientConfiguration{},
                                               [](const std::string& serviceUrl, const ClientConfiguration&,
                                                  ConnectionPool& pool, const AuthenticationPtr&) {
                                                   return std::make_shared<MockLookupService>(
                                                       serviceUrl, pool, ClientConfiguration{});
                                               });
    std::promise<Result> promise;
    client->subscribeAsync("lookup-service-test-after-client-shutdown", "sub", ConsumerConfiguration{},
                           [&promise](Result result, const Consumer&) { promise.set_value(result); });
    // When shutdown is called, there is a pending lookup request due to the 1st lookup is failed in
    // MockLookupService. Verify shutdown will cancel it and return ResultDisconnected.
    client->shutdown();
    EXPECT_EQ(ResultDisconnected, promise.get_future().get());

    // A new subscribeAsync call will fail immediately in the current thread
    Result result = ResultOk;
    client->subscribeAsync("lookup-service-test-retry-after-destroyed", "sub", ConsumerConfiguration{},
                           [&result](Result innerResult, const Consumer&) { result = innerResult; });
    EXPECT_EQ(ResultAlreadyClosed, result);
}

TEST(LookupServiceTest, testRetryAfterDestroyed) {
    auto executorProvider = std::make_shared<ExecutorServiceProvider>(1);
    ConnectionPool pool({}, executorProvider, AuthFactory::Disabled(), "");

    auto internalLookupService =
        std::make_shared<MockLookupService>("pulsar://localhost:6650", pool, ClientConfiguration{});
    auto lookupService =
        RetryableLookupService::create(internalLookupService, std::chrono::seconds(30), executorProvider);

    // Simulate the race condition that `getPartitionMetadataAsync` is called after `close` is called on the
    // lookup service. It's expected the request fails immediately with ResultAlreadyClosed.
    lookupService->close();
    Result result = ResultOk;
    lookupService->getPartitionMetadataAsync(TopicName::get("lookup-service-test-retry-after-destroyed"))
        .addListener([&result](Result innerResult, const LookupDataResultPtr&) { result = innerResult; });
    EXPECT_EQ(ResultAlreadyClosed, result);
    pool.close();
    executorProvider->close();
}