This is an automated email from the ASF dual-hosted git repository.
RongtongJin pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/rocketmq-clients.git
The following commit(s) were added to refs/heads/master by this push:
new f9829e51 fix(nodejs): create the first-settings promise before any
await in startup (#1405)
f9829e51 is described below
commit f9829e51ad5018fb4526bad084bb3d6ddd45a53e
Author: zhaohai <[email protected]>
AuthorDate: Thu Oct 22 17:47:18 2026 +0800
fix(nodejs): create the first-settings promise before any await in startup
(#1405)
* fix(nodejs): create the first-settings promise before any await in startup
startup() waits for the first onSettingsCommand before it completes, but it
registered the promise resolvers only after updateRoutes() had finished. The
route fetch itself opens telemetry sessions and syncs settings while it is
still iterating the topics: after the first topic, the loop awaits the next
queryRoute and yields the event loop, so a server-pushed Settings callback
can fire while the resolvers are still undefined. The first callback is then
silently dropped and startup hangs on the await until the periodic settings
sync fires five minutes later (or forever, if that sync fails).
Create the promise (and its resolvers) before any await in #startup, so the
first Settings callback can resolve it no matter when it arrives. The
onUnknownCommand reject path is covered by the same change.
* fix(nodejs): attach a rejection handler to the first-settings promise at
creation
onUnknownCommand() rejects the first-settings promise as soon as a
status-only
telemetry command arrives. The promise is created before updateRoutes() but
only
awaited after it finishes, so a rejection arriving during another topic's
route
query or the retry backoff window had no rejection handler attached; under
Node.js's default unhandled-rejection behavior that terminates the process
even
though the startup() caller would have handled the rejection.
Attach a no-op catch handler immediately after creating the promise; the
await
below still observes the error and propagates it to the caller. Also clear
the
startup resolvers in a finally block so they never dangle across the
early-throw
(route fetch failure) and first-settings rejection paths.
Reproduced with a minimal script on Node v22.22.2 (exit code 1 pre-fix); a
regression test covering a telemetry error before route updates complete is
included.
---
nodejs/src/client/BaseClient.ts | 150 +++++++++++++++++++--------------
nodejs/test/client/BaseClient.test.ts | 152 ++++++++++++++++++++++++++++++++++
2 files changed, 238 insertions(+), 64 deletions(-)
diff --git a/nodejs/src/client/BaseClient.ts b/nodejs/src/client/BaseClient.ts
index d8cdce85..14929159 100644
--- a/nodejs/src/client/BaseClient.ts
+++ b/nodejs/src/client/BaseClient.ts
@@ -199,80 +199,102 @@ export abstract class BaseClient {
this.logger.info('Begin to execute startup flow, clientId=%s, topics=%d',
this.clientId, this.topics.size);
- // Fetch topic route with retry mechanism (align with Java client)
- const maxAttempts = 3;
- let lastError: Error | undefined;
- for (let attempt = 1; attempt <= maxAttempts; attempt++) {
- try {
- await this.updateRoutes();
- this.logger.info('Fetch topic route successfully during startup,
clientId=%s, attempt=%d',
- this.clientId, attempt);
- break;
- } catch (e) {
- lastError = e as Error;
- // Not-found errors will never succeed on retry — fail fast, aligned
- // with the Java client which surfaces NotFoundException immediately.
- if (e instanceof NotFoundException) {
- throw e;
- }
- if (attempt < maxAttempts) {
- const backoffMs = 1000 * attempt; // Simple linear backoff: 1s, 2s,
3s
- this.logger.warn('Fetch topic route failed during startup, will
retry, clientId=%s, attempt=%d/%d, error=%s, backoff=%dms',
- this.clientId, attempt, maxAttempts, e instanceof Error ?
e.message : String(e), backoffMs);
- await this.sleep(backoffMs);
- } else {
- this.logger.error('Fetch topic route failed after %d attempts,
clientId=%s, error=%s',
- maxAttempts, this.clientId, e instanceof Error ? e.message :
String(e));
+ // Create the first-settings promise before any await: updateRoutes()
below opens
+ // telemetry sessions and syncs settings while it is still iterating the
topics, so a
+ // server-pushed Settings callback can fire long before the code below
reaches the
+ // await. Registering the resolvers only at the await point would silently
drop that
+ // first callback and leave startup hanging until the periodic settings
sync fires
+ // (5 minutes later).
+ const firstSettingsPromise = this.topics.size > 0
+ ? new Promise<void>((resolve, reject) => {
+ this.#startupReject = reject;
+ this.#startupResolve = resolve;
+ })
+ : undefined;
+ // A status-only telemetry command can reach onUnknownCommand() while
+ // updateRoutes() (or its retry backoff) is still pending, rejecting this
+ // promise before the await below attaches a rejection handler. Under
+ // Node.js's default unhandled-rejection behavior that terminates the
+ // process even though startup() would surface the error to its caller,
+ // so attach a no-op handler immediately; the await below still observes
+ // the error and propagates it to the caller.
+ void firstSettingsPromise?.catch(() => undefined);
+
+ try {
+ // Fetch topic route with retry mechanism (align with Java client)
+ const maxAttempts = 3;
+ let lastError: Error | undefined;
+ for (let attempt = 1; attempt <= maxAttempts; attempt++) {
+ try {
+ await this.updateRoutes();
+ this.logger.info('Fetch topic route successfully during startup,
clientId=%s, attempt=%d',
+ this.clientId, attempt);
+ break;
+ } catch (e) {
+ lastError = e as Error;
+ // Not-found errors will never succeed on retry — fail fast, aligned
+ // with the Java client which surfaces NotFoundException immediately.
+ if (e instanceof NotFoundException) {
+ throw e;
+ }
+ if (attempt < maxAttempts) {
+ const backoffMs = 1000 * attempt; // Simple linear backoff: 1s,
2s, 3s
+ this.logger.warn('Fetch topic route failed during startup, will
retry, clientId=%s, attempt=%d/%d, error=%s, backoff=%dms',
+ this.clientId, attempt, maxAttempts, e instanceof Error ?
e.message : String(e), backoffMs);
+ await this.sleep(backoffMs);
+ } else {
+ this.logger.error('Fetch topic route failed after %d attempts,
clientId=%s, error=%s',
+ maxAttempts, this.clientId, e instanceof Error ? e.message :
String(e));
+ }
}
}
- }
- if (lastError && !this.topicRouteCache.size) {
- throw new Error(`Failed to fetch topic routes after ${maxAttempts}
attempts`, { cause: lastError });
- }
-
- this.logger.info('Topic routes updated, clientId=%s', this.clientId);
- // update topic route every 30s
- this.#timers.push(setInterval(async () => {
- try {
- await this.updateRoutes();
- } catch (e) {
- this.logger.error('Failed to update routes periodically, clientId=%s,
error=%s',
- this.clientId, e instanceof Error ? e.message : String(e));
- }
- }, 30000));
-
- // sync settings every 5m
- this.#timers.push(setInterval(async () => {
- try {
- this.#syncSettings();
- } catch (e) {
- this.logger.error('Failed to sync settings, clientId=%s, error=%s',
- this.clientId, e instanceof Error ? e.message : String(e));
+ if (lastError && !this.topicRouteCache.size) {
+ throw new Error(`Failed to fetch topic routes after ${maxAttempts}
attempts`, { cause: lastError });
}
- }, 5 * 60000));
- // heartbeat every 10s
- this.#timers.push(setInterval(async () => {
- this.#doHeartbeat();
- }, 10000));
+ this.logger.info('Topic routes updated, clientId=%s', this.clientId);
+ // update topic route every 30s
+ this.#timers.push(setInterval(async () => {
+ try {
+ await this.updateRoutes();
+ } catch (e) {
+ this.logger.error('Failed to update routes periodically,
clientId=%s, error=%s',
+ this.clientId, e instanceof Error ? e.message : String(e));
+ }
+ }, 30000));
+
+ // sync settings every 5m
+ this.#timers.push(setInterval(async () => {
+ try {
+ this.#syncSettings();
+ } catch (e) {
+ this.logger.error('Failed to sync settings, clientId=%s, error=%s',
+ this.clientId, e instanceof Error ? e.message : String(e));
+ }
+ }, 5 * 60000));
- // doStats every 60s
- // doStats()
+ // heartbeat every 10s
+ this.#timers.push(setInterval(async () => {
+ this.#doHeartbeat();
+ }, 10000));
- if (this.topics.size > 0) {
- this.logger.info('Waiting for first onSettingsCommand, clientId=%s',
this.clientId);
- // wait for this first onSettingsCommand call
- // eslint-disable-next-line @typescript-eslint/no-unused-vars
- await new Promise<void>((resolve, reject) => {
- this.#startupReject = reject;
- this.#startupResolve = resolve;
- });
- this.logger.info('Received first onSettingsCommand, clientId=%s',
this.clientId);
+ // doStats every 60s
+ // doStats()
+
+ if (firstSettingsPromise) {
+ this.logger.info('Waiting for first onSettingsCommand, clientId=%s',
this.clientId);
+ await firstSettingsPromise;
+ this.logger.info('Received first onSettingsCommand, clientId=%s',
this.clientId);
+ }
+ this.#running = true;
+ } finally {
+ // Never leave the resolvers dangling across exits: on the early-throw
+ // (route fetch failure) and first-settings rejection paths they would
+ // otherwise settle an abandoned promise long after startup is over.
this.#startupReject = undefined;
this.#startupResolve = undefined;
}
- this.#running = true;
}
isRunning(): boolean {
diff --git a/nodejs/test/client/BaseClient.test.ts
b/nodejs/test/client/BaseClient.test.ts
new file mode 100644
index 00000000..960d25c8
--- /dev/null
+++ b/nodejs/test/client/BaseClient.test.ts
@@ -0,0 +1,152 @@
+/**
+ * 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.
+ */
+
+/**
+ * Regression tests for the startup first-settings promise:
+ * - a server-pushed Settings arriving while route updates are still pending
+ * must complete startup (no 5-minute hang), and
+ * - a status-only telemetry error arriving in the same window must reject
+ * startup() without being flagged as an unhandled rejection (which would
+ * terminate the process under Node.js's default behavior).
+ */
+
+import { describe, it } from 'node:test';
+import * as assert from 'node:assert';
+import {
+ ClientType,
+ Code,
+ Settings as SettingsPB,
+ Status,
+} from '../../proto/apache/rocketmq/v2/definition_pb';
+import {
+ HeartbeatRequest,
+ NotifyClientTerminationRequest,
+ QueryRouteResponse,
+} from '../../proto/apache/rocketmq/v2/service_pb';
+import { BaseClient } from '../../src/client/BaseClient';
+import { Settings } from '../../src/client/Settings';
+import { Endpoints } from '../../src/route';
+
+const ACCESS_POINT = new Endpoints('127.0.0.1:8081');
+
+class StubSettings extends Settings {
+ constructor() {
+ super('test-namespace', 'test-client-id', ClientType.PRODUCER,
ACCESS_POINT, 3000);
+ }
+
+ toProtobuf(): SettingsPB {
+ return new SettingsPB();
+ }
+
+ sync(settings: SettingsPB): void {
+ void settings; // no-op: the startup race only depends on the callback
firing
+ }
+}
+
+class TestClient extends BaseClient {
+ protected getClientType(): ClientType {
+ return ClientType.PRODUCER;
+ }
+
+ protected getSettings(): Settings {
+ return new StubSettings();
+ }
+
+ protected wrapHeartbeatRequest(): HeartbeatRequest {
+ return new HeartbeatRequest();
+ }
+
+ protected wrapNotifyClientTerminationRequest():
NotifyClientTerminationRequest {
+ return new NotifyClientTerminationRequest();
+ }
+
+ /** Test hook: swap the route query for a delayed OK response. */
+ mockQueryRoute(delayMs: number) {
+ const self = this as unknown as { rpcClientManager: Record<string,
unknown> };
+ self.rpcClientManager.queryRoute = async () => {
+ await sleep(delayMs);
+ const response = new QueryRouteResponse();
+ response.setStatus(new Status().setCode(Code.OK));
+ return response;
+ };
+ }
+}
+
+function sleep(ms: number): Promise<void> {
+ return new Promise(resolve => setTimeout(resolve, ms));
+}
+
+describe('BaseClient startup first-settings promise', () => {
+ it('completes startup when the first settings command arrives before route
updates finish', async () => {
+ const client = new TestClient({
+ endpoints: '127.0.0.1:8081',
+ namespace: 'test-namespace',
+ topics: [ 'topic-A', 'topic-B' ],
+ });
+ client.mockQueryRoute(80);
+
+ const startupPromise = client.startup();
+ // Route updates for the two topics are pending now; the server-pushed
+ // settings callback fires inside this window. Pre-fix, the resolvers were
+ // registered only after updateRoutes() finished, so this callback was
+ // silently dropped and startup hung until the 5-minute periodic sync.
+ await sleep(20);
+ client.onSettingsCommand(client.getEndpoints(), new SettingsPB());
+ await startupPromise;
+ assert.strictEqual(client.isRunning(), true);
+ });
+
+ it('propagates a telemetry error received before route updates complete
without an unhandled rejection', async () => {
+ const client = new TestClient({
+ endpoints: '127.0.0.1:8081',
+ namespace: 'test-namespace',
+ topics: [ 'topic-A', 'topic-B' ],
+ });
+ client.mockQueryRoute(80);
+
+ // Record unhandled rejections instead of letting them kill the test
+ // process. On stock Node.js (no fix) this scenario rejects the
+ // first-settings promise with no handler attached, which terminates the
+ // process by default; the listener both keeps the test alive and captures
+ // the evidence for the assertion below.
+ const unhandled: unknown[] = [];
+ const onUnhandled = (reason: unknown) => {
+ unhandled.push(reason);
+ };
+ process.on('unhandledRejection', onUnhandled);
+ try {
+ const startupPromise = client.startup();
+ // A status-only telemetry command arrives while the first route query
+ // is still pending: onUnknownCommand rejects the first-settings promise
+ // before the code in #startup() attaches a rejection handler to it.
+ await sleep(20);
+ client.onUnknownCommand(client.getEndpoints(), {
+ code: Code.BAD_REQUEST,
+ message: 'mock telemetry failure',
+ requestId: '',
+ } as unknown as Status.AsObject);
+
+ // startup() must surface the telemetry error to the caller...
+ await assert.rejects(startupPromise, /mock telemetry failure/);
+ // ...and the rejection must never have been flagged as unhandled.
+ await sleep(20);
+ assert.deepStrictEqual(unhandled, []);
+ } finally {
+ process.off('unhandledRejection', onUnhandled);
+ }
+ });
+});
