You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pulsar.apache.org by "shibd (via GitHub)" <gi...@apache.org> on 2026/06/02 11:02:12 UTC
[PR] Add auto cluster failover support [pulsar-client-node]
shibd opened a new pull request, #487:
URL: https://github.com/apache/pulsar-client-node/pull/487
## Motivation
The Node.js client did not expose the cluster-level failover support added in the C++ client, so applications could not automatically switch from a primary Pulsar cluster to a secondary cluster when the primary became unavailable.
## Changes
- Add `serviceUrlProvider` support to `ClientConfig` and wire it to the C++ `pulsar::AutoClusterFailover` implementation.
- Support primary and secondary service info entries with optional per-cluster authentication and TLS trust cert configuration.
- Preserve the existing `serviceUrl` client path and validate that `serviceUrl` and `serviceUrlProvider` are not configured together.
- Keep authentication bindings alive for top-level and per-cluster authentication objects.
- Add a dedicated `failoverClientTest` integration test that starts two standalone Pulsar clusters, stops the primary, and verifies the same client can continue producing and consuming through the secondary.
## Validation
- `npm run type-check`
- `./node_modules/.bin/node-pre-gyp build`
- `npx jest tests/failoverClientTest.test.js --runInBand`
Related to #479.
--
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: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add auto cluster failover support [pulsar-client-node]
Posted by "BewareMyPower (via GitHub)" <gi...@apache.org>.
BewareMyPower merged PR #487:
URL: https://github.com/apache/pulsar-client-node/pull/487
--
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: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add auto cluster failover support [pulsar-client-node]
Posted by "Copilot (via GitHub)" <gi...@apache.org>.
Copilot commented on code in PR #487:
URL: https://github.com/apache/pulsar-client-node/pull/487#discussion_r3340881561
##########
src/Client.cc:
##########
@@ -103,16 +307,28 @@ Client::Client(const Napi::CallbackInfo &info) : Napi::ObjectWrap<Client>(info)
Napi::HandleScope scope(env);
Napi::Object clientConfig = info[0].As<Napi::Object>();
- if (!clientConfig.Has(CFG_SERVICE_URL) || !clientConfig.Get(CFG_SERVICE_URL).IsString() ||
- clientConfig.Get(CFG_SERVICE_URL).ToString().Utf8Value().empty()) {
- Napi::Error::New(env, "Service URL is required and must be specified as a string")
+ bool hasServiceUrlProvider =
+ clientConfig.Has(CFG_SERVICE_URL_PROVIDER) && IsPresent(clientConfig.Get(CFG_SERVICE_URL_PROVIDER));
+ bool hasServiceUrl = clientConfig.Has(CFG_SERVICE_URL) && IsPresent(clientConfig.Get(CFG_SERVICE_URL));
+ if (hasServiceUrlProvider && hasServiceUrl) {
+ Napi::Error::New(env, "Only one of serviceUrl or serviceUrlProvider can be configured")
+ .ThrowAsJavaScriptException();
+ return;
+ }
+
+ if (!hasServiceUrlProvider && (!hasServiceUrl || !clientConfig.Get(CFG_SERVICE_URL).IsString() ||
+ clientConfig.Get(CFG_SERVICE_URL).ToString().Utf8Value().empty())) {
+ Napi::Error::New(env,
+ "Service URL is required and must be specified as a string unless serviceUrlProvider "
+ "is configured")
.ThrowAsJavaScriptException();
return;
}
Review Comment:
The updated validation throws the extended message ("... unless serviceUrlProvider is configured") not only when serviceUrl is missing, but also when serviceUrl is present-but-invalid (empty / non-string). That makes the error less precise and also breaks the existing tests that still expect the original message for those cases. Consider splitting the checks so only the missing-serviceUrl case mentions serviceUrlProvider, while invalid serviceUrl keeps the original message.
##########
src/Client.cc:
##########
@@ -103,16 +307,28 @@ Client::Client(const Napi::CallbackInfo &info) : Napi::ObjectWrap<Client>(info)
Napi::HandleScope scope(env);
Napi::Object clientConfig = info[0].As<Napi::Object>();
- if (!clientConfig.Has(CFG_SERVICE_URL) || !clientConfig.Get(CFG_SERVICE_URL).IsString() ||
- clientConfig.Get(CFG_SERVICE_URL).ToString().Utf8Value().empty()) {
- Napi::Error::New(env, "Service URL is required and must be specified as a string")
+ bool hasServiceUrlProvider =
+ clientConfig.Has(CFG_SERVICE_URL_PROVIDER) && IsPresent(clientConfig.Get(CFG_SERVICE_URL_PROVIDER));
+ bool hasServiceUrl = clientConfig.Has(CFG_SERVICE_URL) && IsPresent(clientConfig.Get(CFG_SERVICE_URL));
+ if (hasServiceUrlProvider && hasServiceUrl) {
+ Napi::Error::New(env, "Only one of serviceUrl or serviceUrlProvider can be configured")
+ .ThrowAsJavaScriptException();
+ return;
Review Comment:
There is new behavior rejecting configs that set both `serviceUrl` and `serviceUrlProvider`, but the unit tests only cover the missing-serviceUrl case. Adding a test that asserts the new "Only one of serviceUrl or serviceUrlProvider can be configured" error would help prevent regressions.
##########
src/Client.cc:
##########
@@ -227,8 +445,23 @@ Client::Client(const Napi::CallbackInfo &info) : Napi::ObjectWrap<Client>(info)
}
try {
- this->cClient = std::shared_ptr<pulsar_client_t>(
- pulsar_client_create(serviceUrl.Utf8Value().c_str(), cClientConfig.get()), pulsar_client_free);
+ if (hasServiceUrlProvider) {
+ std::unique_ptr<pulsar::ServiceInfoProvider> serviceInfoProvider = BuildServiceInfoProvider(
+ clientConfig, this->authRefs_, defaultAuthentication, defaultTlsTrustCertsFilePath);
+ if (serviceInfoProvider == nullptr) {
+ return;
+ }
+
+ pulsar_client_t *rawClient = new pulsar_client_t;
+ rawClient->client.reset(new pulsar::Client(
+ pulsar::Client::create(std::move(serviceInfoProvider), cClientConfig.get()->conf)));
+ this->cClient =
+ std::shared_ptr<pulsar_client_t>(rawClient, [](pulsar_client_t *client) { delete client; });
+ } else {
Review Comment:
If `pulsar::Client::create(...)` throws, `rawClient` is leaked because it is allocated with `new` before the call and only wrapped in a `shared_ptr` afterwards. Wrapping it in a `unique_ptr` until construction succeeds avoids the leak on exceptional paths.
##########
tests/failoverClientTest.test.js:
##########
@@ -0,0 +1,259 @@
+/**
+ * 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.
+ */
+
+const childProcess = require('child_process');
+const fs = require('fs');
+const http = require('http');
+const os = require('os');
+const path = require('path');
+const { promisify } = require('util');
+const Pulsar = require('../index');
+
+const execFile = promisify(childProcess.execFile);
+const dockerImage = process.env.PULSAR_TEST_IMAGE || 'apachepulsar/pulsar:latest';
+const testRunId = `${Date.now()}-${process.pid}`;
+const tempRoot = path.join(os.tmpdir(), `pulsar-node-failover-client-test-${testRunId}`);
+const startedContainers = [];
+
+jest.setTimeout(180000);
+
+const delay = (ms) => new Promise((resolve) => setTimeout(resolve, ms));
+
+const docker = async (args, options = {}) => {
+ const { stdout } = await execFile('docker', args, {
+ maxBuffer: 1024 * 1024,
+ ...options,
+ });
+ return stdout.trim();
+};
+
+const waitForHttpOk = async (url, timeoutMs = 60000) => {
+ const deadline = Date.now() + timeoutMs;
+
+ while (Date.now() < deadline) {
+ try {
+ await new Promise((resolve, reject) => {
+ const request = http.get(url, (response) => {
+ response.resume();
+ if (response.statusCode >= 200 && response.statusCode < 500) {
+ resolve();
+ } else {
+ reject(new Error(`Unexpected status code ${response.statusCode}`));
+ }
Review Comment:
`waitForHttpOk` currently treats any HTTP 2xx–4xx response as success. That means a 404/401 could be interpreted as "ready" and hide readiness issues, making the test flaky. If the intention is to wait for a healthy endpoint, require a 2xx response (or check a dedicated health endpoint).
##########
src/Client.cc:
##########
@@ -145,9 +361,10 @@ Client::Client(const Napi::CallbackInfo &info) : Napi::ObjectWrap<Client>(info)
if (clientConfig.Has(CFG_AUTH) && clientConfig.Get(CFG_AUTH).IsObject()) {
Napi::Object obj = clientConfig.Get(CFG_AUTH).ToObject();
if (obj.Has(CFG_AUTH_PROP) && obj.Get(CFG_AUTH_PROP).IsObject()) {
- this->authRef_ = Napi::Persistent(obj.Get(CFG_AUTH_PROP).As<Napi::Object>());
- Authentication *auth = Authentication::Unwrap(this->authRef_.Value());
+ this->authRefs_.emplace_back(Napi::Persistent(obj.Get(CFG_AUTH_PROP).As<Napi::Object>()));
+ Authentication *auth = Authentication::Unwrap(this->authRefs_.back().Value());
Review Comment:
If the provided authentication object's `binding` property is not an instance of the native `Authentication` wrapper, `Authentication::Unwrap(...)` can return nullptr. The current code dereferences `auth` unconditionally (and now also reads `auth->GetCAuthentication()->auth`), which can segfault the process instead of throwing a JS error.
--
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: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org