Merge remote-tracking branch 'apache/main' into scalable-topics-cpp-api

Author: merlimat

README.md

@@ -27,6 +27,19 @@ For the supported Pulsar features, see [Client Feature Matrix](https://pulsar.ap
 
 For how to use APIs to publish and consume messages, see [examples](https://github.com/apache/pulsar-client-cpp/tree/main/examples).
 
+## Custom logger lifetime
+
+The C++ client supports one custom logger factory per process. A logger configured
+through `ClientConfiguration::setLogger` or the C API logger functions is shared
+by all clients in the same process and is not scoped to an individual client
+instance. Set the custom logger before creating clients, and do not expect
+different clients to have independent log handlers.
+
+If an application or language binding exposes logger callbacks, the callback and
+its context must remain valid until all Pulsar clients are closed and no
+background thread can emit client logs. Avoid tying the callback lifetime to a
+single client when multiple clients can exist in the process.
+
 ## Import the library into your project
 
 ### CMake with vcpkg integration

include/pulsar/ClientConfiguration.h

@@ -187,15 +187,49 @@ class PULSAR_PUBLIC ClientConfiguration {
     int getMaxBackoffIntervalMs() const;
 
     /**
-     * Configure a custom logger backend to route of Pulsar client library
+     * Configure whether to send authentication credentials when following HTTP redirects
+     * to a different host during lookup requests.
+     *
+     * When enabled, the Authorization header will be forwarded on cross-origin redirects.
+     *
+     * HTTP lookup redirects typically occur when the broker receiving the lookup request
+     * is not the owner of the requested topic. In this case, the broker responds with
+     * an HTTP redirect (3xx) pointing to the correct owner broker. If authentication is
+     * enabled, the redirected request needs to carry the auth credentials to be accepted
+     * by the target broker.
+     *
+     * If this option is not enabled and the cluster has authentication enabled, the
+     * redirected request will not carry credentials, which may result in a 401
+     * Unauthorized error from the target broker.
+     *
+     * The default value is false.
+     *
+     * @param allow whether to allow sending auth credentials on redirects
+     */
+    ClientConfiguration& setHttpLookupAuthAllowRedirect(bool allow);
+
+    /**
+     * @return whether auth credentials are sent on HTTP redirects
+     */
+    bool isHttpLookupAuthAllowRedirect() const;
+
+    /**
+     * Configure a custom logger backend to route Pulsar client library logs
      * to a different logger implementation.
      *
      * By default, log messages are printed on standard output.
      *
      * When passed in, the configuration takes ownership of the loggerFactory object.
-     * The logger factory can only be set once per process. Any subsequent calls to
-     * set the logger factory will have no effect, though the logger factory object
-     * will be cleaned up.
+     * The logger factory is process-wide and is not scoped to a Client instance.
+     * It can only be set once per process. Any subsequent calls to set the logger
+     * factory will have no effect, though the logger factory object will be
+     * cleaned up.
+     *
+     * Applications and language bindings that use callback-based logger factories
+     * should set the logger before creating clients and ensure callback state
+     * outlives all Pulsar clients and background threads that can emit logs.
+     * Avoid using per-client callback objects that can be destroyed while another
+     * client in the same process is still running.
      */
     ClientConfiguration& setLogger(LoggerFactory* loggerFactory);
 

include/pulsar/c/client_configuration.h

@@ -141,9 +141,25 @@ PULSAR_PUBLIC void pulsar_client_configuration_set_concurrent_lookup_request(
 PULSAR_PUBLIC int pulsar_client_configuration_get_concurrent_lookup_request(
     pulsar_client_configuration_t *conf);
 
+/**
+ * Configure a custom logger for Pulsar client library logs.
+ *
+ * The logger is process-wide and is not scoped to a pulsar_client_t instance.
+ * It can only be set once per process. Applications and language bindings
+ * should set the logger before creating clients and ensure the logger callback
+ * context outlives all Pulsar clients and background threads that can emit logs.
+ */
 PULSAR_PUBLIC void pulsar_client_configuration_set_logger(pulsar_client_configuration_t *conf,
                                                           pulsar_logger logger, void *ctx);
 
+/**
+ * Configure a custom logger for Pulsar client library logs.
+ *
+ * The logger is process-wide and is not scoped to a pulsar_client_t instance.
+ * It can only be set once per process. Applications and language bindings
+ * should set the logger before creating clients and ensure the logger callback
+ * context outlives all Pulsar clients and background threads that can emit logs.
+ */
 PULSAR_PUBLIC void pulsar_client_configuration_set_logger_t(pulsar_client_configuration_t *conf,
                                                             pulsar_logger_t logger);
 
@@ -203,6 +219,12 @@ PULSAR_PUBLIC void pulsar_client_configuration_set_keep_alive_interval_in_second
 PULSAR_PUBLIC unsigned int pulsar_client_configuration_get_keep_alive_interval_in_seconds(
     pulsar_client_configuration_t *conf);
 
+PULSAR_PUBLIC void pulsar_client_configuration_set_http_lookup_auth_allow_redirect(
+    pulsar_client_configuration_t *conf, int httpLookupAuthAllowRedirect);
+
+PULSAR_PUBLIC int pulsar_client_configuration_is_http_lookup_auth_allow_redirect(
+    pulsar_client_configuration_t *conf);
+
 #ifdef __cplusplus
 }
 #endif

include/pulsar/c/result.h

@@ -92,6 +92,7 @@ typedef enum
 
     pulsar_result_MemoryBufferIsFull,  /// Client-wide memory limit has been reached
     pulsar_result_Interrupted,         /// Interrupted while waiting to dequeue
+    pulsar_result_Disconnected,        /// Client connection has been disconnected
 } pulsar_result;
 
 // Return string representation of result code

lib/ClientConfiguration.cc

@@ -231,4 +231,11 @@ ClientConfiguration& ClientConfiguration::setDescription(const std::string& desc
 
 const std::string& ClientConfiguration::getDescription() const noexcept { return impl_->description; }
 
+ClientConfiguration& ClientConfiguration::setHttpLookupAuthAllowRedirect(bool allow) {
+    impl_->httpLookupAuthAllowRedirect = allow;
+    return *this;
+}
+
+bool ClientConfiguration::isHttpLookupAuthAllowRedirect() const { return impl_->httpLookupAuthAllowRedirect; }
+
 }  // namespace pulsar

lib/ClientConfigurationImpl.h

@@ -53,6 +53,7 @@ struct ClientConfigurationImpl {
     std::string description;
     std::string proxyServiceUrl;
     ClientConfiguration::ProxyProtocol proxyProtocol;
+    bool httpLookupAuthAllowRedirect{false};
 
     std::unique_ptr<LoggerFactory> takeLogger() { return std::move(loggerFactory); }
 

lib/CurlWrapper.h

@@ -54,6 +54,7 @@ class CurlWrapper {
         std::string userAgent;
         int timeoutInSeconds{0};
         int maxLookupRedirects{-1};
+        bool authAllowRedirect{false};
     };
 
     struct TlsContext {
@@ -128,6 +129,9 @@ inline CurlWrapper::Result CurlWrapper::get(const std::string& url, const std::s
     // Redirects
     curl_easy_setopt(handle_, CURLOPT_FOLLOWLOCATION, 1L);
     curl_easy_setopt(handle_, CURLOPT_MAXREDIRS, options.maxLookupRedirects);
+    if (options.authAllowRedirect) {
+        curl_easy_setopt(handle_, CURLOPT_UNRESTRICTED_AUTH, 1L);
+    }
 
     char errorBuffer[CURL_ERROR_SIZE] = "";
     curl_easy_setopt(handle_, CURLOPT_ERRORBUFFER, errorBuffer);

lib/HTTPLookupService.cc

@@ -61,7 +61,8 @@ HTTPLookupService::HTTPLookupService(const ServiceInfo &serviceInfo,
       tlsTrustCertsFilePath_(serviceInfo.tlsTrustCertsFilePath().value_or("")),
       isUseTls_(serviceInfo.useTls()),
       tlsAllowInsecure_(clientConfiguration.isTlsAllowInsecureConnection()),
-      tlsValidateHostname_(clientConfiguration.isValidateHostName()) {}
+      tlsValidateHostname_(clientConfiguration.isValidateHostName()),
+      httpLookupAuthAllowRedirect_(clientConfiguration.isHttpLookupAuthAllowRedirect()) {}
 
 auto HTTPLookupService::getBroker(const TopicName &topicName) -> LookupResultFuture {
     LookupResultPromise promise;
@@ -228,6 +229,7 @@ Error HTTPLookupService::sendHTTPRequest(const std::string &completeUrl, std::st
     options.timeoutInSeconds = lookupTimeoutInSeconds_;
     options.userAgent = std::string("Pulsar-CPP-v") + PULSAR_VERSION_STR;
     options.maxLookupRedirects = maxLookupRedirects_;
+    options.authAllowRedirect = httpLookupAuthAllowRedirect_;
     auto result = curl.get(completeUrl, authDataContent->getHttpHeaders(), options, tlsContext.get());
 
     responseData = result.responseData;

lib/HTTPLookupService.h

@@ -54,6 +54,7 @@ class HTTPLookupService : public LookupService, public std::enable_shared_from_t
     bool isUseTls_;
     bool tlsAllowInsecure_;
     bool tlsValidateHostname_;
+    bool httpLookupAuthAllowRedirect_;
 
     static LookupDataResultPtr parsePartitionData(const std::string&);
     static LookupDataResultPtr parseLookupData(const std::string&);

lib/c/c_ClientConfiguration.cc

@@ -204,3 +204,12 @@ unsigned int pulsar_client_configuration_get_keep_alive_interval_in_seconds(
     pulsar_client_configuration_t *conf) {
     return conf->conf.getKeepAliveIntervalInSeconds();
 }
+
+void pulsar_client_configuration_set_http_lookup_auth_allow_redirect(pulsar_client_configuration_t *conf,
+                                                                     int httpLookupAuthAllowRedirect) {
+    conf->conf.setHttpLookupAuthAllowRedirect(httpLookupAuthAllowRedirect);
+}
+
+int pulsar_client_configuration_is_http_lookup_auth_allow_redirect(pulsar_client_configuration_t *conf) {
+    return conf->conf.isHttpLookupAuthAllowRedirect();
+}