address comments

Author: BewareMyPower

include/pulsar/Client.h

@@ -78,8 +78,9 @@ class PULSAR_PUBLIC Client {
      * dynamically. For example, if it detects a primary Pulsar service is down, it can switch to a secondary
      * service and update the client with the new service information.
      *
-     * When `close` is called, the client will call `ServiceInfoProvider::close` to guarantee the lifetime of
-     * the provider is properly managed.
+     * The Client instance takes ownership of the given ServiceInfoProvider. The provider will be destroyed
+     * as part of the client's shutdown lifecycle, for example when `Client::close()` or
+     * `Client::closeAsync()` is called, ensuring that its lifetime is properly managed.
      */
     static Client create(std::unique_ptr<ServiceInfoProvider> serviceInfoProvider,
                          const ClientConfiguration& clientConfiguration);

include/pulsar/ServiceInfoProvider.h

@@ -19,9 +19,10 @@
 #ifndef PULSAR_SERVICE_INFO_PROVIDER_H_
 #define PULSAR_SERVICE_INFO_PROVIDER_H_
 
-#include <pulsar/ClientConfiguration.h>
 #include <pulsar/ServiceInfo.h>
 
+#include <functional>
+
 namespace pulsar {
 
 class PULSAR_PUBLIC ServiceInfoProvider {
@@ -35,18 +36,23 @@ class PULSAR_PUBLIC ServiceInfoProvider {
     /**
      * Get the initial `ServiceInfo` connection for the client.
      * This method is called **only once** internally in `Client::create()` to get the initial `ServiceInfo`
-     * for the client to connect to the Pulsar service. Since it's only called once, it's legal to return a
-     * moved `ServiceInfo` object to avoid unnecessary copying.
+     * for the client to connect to the Pulsar service, typically before {@link initialize} is invoked.
+     * Since it's only called once, it's legal to return a moved `ServiceInfo` object to avoid unnecessary
+     * copying.
      */
     virtual ServiceInfo initialServiceInfo() = 0;
 
     /**
      * Initialize the ServiceInfoProvider.
      *
-     * @param onServiceInfoUpdate the callback to update `client` with the new `ServiceInfo`
+     * After the client has obtained the initial `ServiceInfo` via {@link initialServiceInfo}, this method is
+     * called to allow the provider to start any background work (for example, service discovery or watching
+     * configuration changes) and to report subsequent updates to the service information.
+     *
+     * @param onServiceInfoUpdate the callback to deliver updated `ServiceInfo` values to the client after
+     *                            the initial connection has been established
      *
-     * Note: the implementation is responsible to invoke `onServiceInfoUpdate` at least once to provide the
-     * initial `ServiceInfo` for the client.
+     * Implementations may choose not to invoke `onServiceInfoUpdate` if the `ServiceInfo` never changes.
      */
     virtual void initialize(std::function<void(ServiceInfo)> onServiceInfoUpdate) = 0;
 };

lib/ClientConnection.h

@@ -157,7 +157,9 @@ class PULSAR_PUBLIC ClientConnection : public std::enable_shared_from_this<Clien
      * Close the connection.
      *
      * @param result all pending futures will complete with this result
-     * @param detach remove it from the pool if it's true
+     * @param detach remove it from the pool if it's true. When false, the connection remains
+     *        associated with the pool but is logically closed; this is currently used when the
+     *        pool itself is being closed or when switching clusters.
      * @param switchCluster whether the close is triggered by cluster switching
      *
      * `detach` should only be false when the connection pool is closed.

lib/ClientImpl.cc

@@ -34,7 +34,7 @@
 #include "Commands.h"
 #include "ConsumerImpl.h"
 #include "ConsumerInterceptors.h"
-#include "DefaultServiceUrlProvider.h"
+#include "DefaultServiceInfoProvider.h"
 #include "ExecutorService.h"
 #include "HTTPLookupService.h"
 #include "LogUtils.h"
@@ -97,8 +97,8 @@ ClientImpl::ClientImpl(const std::string& serviceUrl, const ClientConfiguration&
 
 ClientImpl::ClientImpl(const std::string& serviceUrl, const ClientConfiguration& clientConfiguration,
                        LookupServiceFactory&& lookupServiceFactory)
-    : ClientImpl(std::make_unique<DefaultServiceUrlProvider>(std::cref(serviceUrl),
-                                                             std::cref(*clientConfiguration.impl_)),
+    : ClientImpl(std::make_unique<DefaultServiceInfoProvider>(std::cref(serviceUrl),
+                                                              std::cref(*clientConfiguration.impl_)),
                  clientConfiguration, std::move(lookupServiceFactory)) {}
 
 ClientImpl::ClientImpl(std::unique_ptr<ServiceInfoProvider> serviceInfoProvider,

lib/ClientImpl.h

@@ -206,7 +206,7 @@ class ClientImpl : public std::enable_shared_from_this<ClientImpl> {
                                           const std::string& logicalAddress);
 
     // This overload is only used for blue-green migration, where only the service URL is modified, the other
-    // paramters remain the same
+    // parameters remain the same
     LookupServicePtr createRedirectedLookup(const std::string& redirectedUrl) {
         auto serviceInfo = serviceInfo_.load();
         return createLookup(

lib/DefaultServiceUrlProvider.h lib/DefaultServiceInfoProvider.hlib/DefaultServiceUrlProvider.h renamed to lib/DefaultServiceInfoProvider.h

@@ -26,9 +26,9 @@
 
 namespace pulsar {
 
-class DefaultServiceUrlProvider : public ServiceInfoProvider {
+class DefaultServiceInfoProvider : public ServiceInfoProvider {
    public:
-    DefaultServiceUrlProvider(const std::string& serviceUrl, const ClientConfigurationImpl& config)
+    DefaultServiceInfoProvider(const std::string& serviceUrl, const ClientConfigurationImpl& config)
         : serviceInfo_(config.toServiceInfo(serviceUrl)) {}
 
     ServiceInfo initialServiceInfo() override { return std::move(serviceInfo_); }

perf/PerfConsumer.cc

@@ -57,7 +57,6 @@ static int64_t currentTimeMillis() {
 struct Arguments {
     std::string authParams;
     std::string authPlugin;
-    bool isUseTls;
     bool isTlsAllowInsecureConnection;
     std::string tlsTrustCertsFilePath;
     std::string topic;
@@ -261,9 +260,6 @@ int main(int argc, char** argv) {
         ("auth-plugin,a", po::value<std::string>(&args.authPlugin)->default_value(""),
          "Authentication plugin class library path")  //
 
-        ("use-tls,b", po::value<bool>(&args.isUseTls)->default_value(false),
-         "Whether tls connection is used")  //
-
         ("allow-insecure,d", po::value<bool>(&args.isTlsAllowInsecureConnection)->default_value(true),
          "Whether insecure tls connection is allowed")  //
 

perf/PerfProducer.cc

@@ -47,7 +47,6 @@ typedef std::shared_ptr<pulsar::RateLimiter> RateLimiterPtr;
 struct Arguments {
     std::string authParams;
     std::string authPlugin;
-    bool isUseTls;
     bool isTlsAllowInsecureConnection;
     std::string tlsTrustCertsFilePath;
     std::string topic;
@@ -223,9 +222,6 @@ int main(int argc, char** argv) {
         ("auth-plugin,a", po::value<std::string>(&args.authPlugin)->default_value(""),
          "Authentication plugin class library path")  //
 
-        ("use-tls,b", po::value<bool>(&args.isUseTls)->default_value(false),
-         "Whether tls connection is used")  //
-
         ("allow-insecure,d", po::value<bool>(&args.isTlsAllowInsecureConnection)->default_value(true),
          "Whether insecure tls connection is allowed")  //
 

tests/ServiceInfoProviderTest.cc

@@ -94,7 +94,7 @@ class TestServiceInfoProvider : public ServiceInfoProvider {
 };
 
 TEST(ServiceInfoProviderTest, testSwitchCluster) {
-    extern std::string getToken();  // from tests/AuthToken.cc
+    extern std::string getToken();  // from tests/AuthTokenTest.cc
     // Access "private/auth" namespace in cluster 1
     ServiceInfo info1{"pulsar://localhost:6650", AuthToken::createWithToken(getToken())};
     // Access "private/auth" namespace in cluster 2