Copilot commented on code in PR #521:
URL: https://github.com/apache/pulsar-client-cpp/pull/521#discussion_r2496887944
##########
lib/ClientImpl.cc:
##########
@@ -767,6 +776,7 @@ void ClientImpl::shutdown() {
<< " consumers have been shutdown.");
}
+ lookupServicePtr_->close();
Review Comment:
The `lookupServicePtr_->close()` is called twice in the client lifecycle -
once in `closeAsync()` at line 677 and again in `shutdown()` at line 779. While
this might work if `close()` is idempotent, calling it twice could cause
issues. Consider adding a guard to ensure `close()` is only called once, or
document that `close()` must be idempotent.
##########
tests/LookupServiceTest.cc:
##########
@@ -500,3 +501,62 @@ TEST(LookupServiceTest, testRedirectionLimit) {
}
}
}
+
+static std::atomic_bool firstTime{true};
+
+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);
+ }
+};
+
+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); });
+ client->shutdown();
+ EXPECT_EQ(ResultDisconnected, promise.get_future().get());
+
+ firstTime = true;
+ std::promise<Result> promise2;
+ client->subscribeAsync("lookup-service-test-retry-after-destroyed", "sub",
ConsumerConfiguration{},
+ [&promise2](Result result, const Consumer&) {
promise2.set_value(result); });
+ EXPECT_EQ(ResultAlreadyClosed, promise2.get_future().get());
+}
+
+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.
+ lookupService->close();
+ std::atomic<Result> result{ResultUnknownError};
+
lookupService->getPartitionMetadataAsync(TopicName::get("lookup-service-test-retry-after-destroyed"))
+ .addListener([&result](Result innerResult, const LookupDataResultPtr&)
{ result = innerResult; });
+ EXPECT_EQ(ResultAlreadyClosed, result.load());
Review Comment:
[nitpick] Using `std::atomic<Result>` with `load()` is unnecessary here
since the test is single-threaded and the value is only read after the async
operation completes. A simple `Result result = ResultUnknownError;` followed by
capturing by reference in the lambda would be clearer and sufficient.
```suggestion
Result result = ResultUnknownError;
lookupService->getPartitionMetadataAsync(TopicName::get("lookup-service-test-retry-after-destroyed"))
.addListener([&result](Result innerResult, const
LookupDataResultPtr&) { result = innerResult; });
EXPECT_EQ(ResultAlreadyClosed, result);
```
##########
tests/LookupServiceTest.cc:
##########
@@ -500,3 +501,62 @@ TEST(LookupServiceTest, testRedirectionLimit) {
}
}
}
+
+static std::atomic_bool firstTime{true};
Review Comment:
Using a global mutable variable for test state can cause test
interdependencies and flakiness. The `firstTime` flag is reset at line 537 but
shared between tests. Consider making this variable local to each test or a
member of the test fixture to avoid potential race conditions when tests run in
parallel.
--
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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
