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 cd738faf fix(nodejs): transport layer self-healing and logger close
deferral (#1375)
cd738faf is described below
commit cd738faf51f9396af169229a68507bf076cc2eb9
Author: zhaohai <[email protected]>
AuthorDate: Wed Sep 23 11:19:00 2026 +0800
fix(nodejs): transport layer self-healing and logger close deferral (#1375)
- Endpoints: prefix gRPC target with resolver scheme (ipv4:/ipv6:/dns:),
bracket IPv6 hosts for grpc-js
- RpcClientManager: key rpc clients by endpoints facade string instead of
object identity (endpoints are recreated on every route fetch, leaking
duplicate channels); add evict()
- RpcClient: add keepalive (5min/30s + permit_without_calls) and max
message size channel options
- BaseClient: rejoin isolated endpoints after successful heartbeat;
rebuild transport (evict rpc client + refresh telemetry session) after
consecutive heartbeat failures with a 30s per-endpoints cooldown;
respond VerifyMessageResult (not echo command) to verify-message
command; return real process diagnostics for print-thread-stack-trace
- BaseClient: defer logger.close() to the next macrotask so subclass
shutdown logs do not throw 'log stream had been close'
---
nodejs/src/client/BaseClient.ts | 104 ++++++++++++++++++++++++++++++++--
nodejs/src/client/RpcClient.ts | 19 +++++--
nodejs/src/client/RpcClientManager.ts | 35 +++++++++---
nodejs/src/route/Endpoints.ts | 21 ++++++-
nodejs/test/route/Endpoints.test.ts | 56 ++++++++++++++++++
5 files changed, 216 insertions(+), 19 deletions(-)
diff --git a/nodejs/src/client/BaseClient.ts b/nodejs/src/client/BaseClient.ts
index 6b6dfaa3..d2b0e852 100644
--- a/nodejs/src/client/BaseClient.ts
+++ b/nodejs/src/client/BaseClient.ts
@@ -28,6 +28,7 @@ import {
QueryRouteRequest,
RecoverOrphanedTransactionCommand,
VerifyMessageCommand,
+ VerifyMessageResult,
PrintThreadStackTraceCommand,
ReconnectEndpointsCommand,
NotifyUnsubscribeLiteCommand,
@@ -49,6 +50,11 @@ import { ClientId } from './ClientId';
const debug = debuglog('rocketmq-client-nodejs:client:BaseClient');
+// Trigger transport self-healing (D-2) after this many consecutive heartbeat
failures
+const HEARTBEAT_TRANSPORT_RECOVERY_THRESHOLD = 2;
+// Minimum interval between transport-recovery attempts per endpoints
+const TRANSPORT_RECOVERY_COOLDOWN = 30 * 1000;
+
export interface BaseClientOptions {
sslEnabled?: boolean;
/**
@@ -93,6 +99,10 @@ export abstract class BaseClient {
#startupReject?: (err: Error) => void;
#timers: NodeJS.Timeout[] = [];
#running = false;
+ // Consecutive heartbeat failure counts keyed by endpoints facade (D-2
transport self-healing)
+ readonly #heartbeatFailureCounts = new Map<string, number>();
+ // Last transport-recovery timestamps per endpoints facade, to throttle
recovery attempts
+ readonly #lastTransportRecoveryTimes = new Map<string, number>();
/**
* Get the client type.
@@ -272,7 +282,11 @@ export abstract class BaseClient {
this.isolated.clear();
this.logger.info('Shutdown the rocketmq client successfully, clientId=%s',
this.clientId);
- this.logger.close && this.logger.close();
+ // Defer closing the logger to the next macrotask: subclass shutdown()
+ // implementations (Producer / PushConsumer) emit their final log line
right
+ // after super.shutdown(), and egg-logger throws "log stream had been
closed"
+ // if it is used again after close().
+ setImmediate(() => this.logger.close && this.logger.close());
}
async #doHeartbeat() {
@@ -286,10 +300,18 @@ export abstract class BaseClient {
for (const endpoints of endpointsList) {
try {
await this.rpcClientManager.heartbeat(endpoints, request,
this.requestTimeout);
+ // Heartbeat succeeded: the remote is reachable again, so rejoin any
+ // isolated endpoints (mirrors Java ClientImpl#doHeartbeat).
+ if (this.isolated.delete(endpoints.facade)) {
+ this.logger.info('Isolated endpoints rejoined after successful
heartbeat, endpoints=%s, clientId=%s',
+ endpoints.facade, this.clientId);
+ }
+ this.#heartbeatFailureCounts.delete(endpoints.facade);
} catch (e) {
// Log but don't throw - heartbeat is best-effort
this.logger.warn('Heartbeat failed for endpoints=%s, clientId=%s,
error=%s',
endpoints.facade, this.clientId, e instanceof Error ? e.message :
String(e));
+ this.#onHeartbeatFailure(endpoints);
}
}
} catch (e) {
@@ -298,6 +320,47 @@ export abstract class BaseClient {
}
}
+ /**
+ * Handle a heartbeat failure: after consecutive failures, rebuild the
transport
+ * layer for the affected endpoints (evict the stale RpcClient channel and
refresh
+ * the telemetry session), throttled by a per-endpoints cooldown.
+ */
+ #onHeartbeatFailure(endpoints: Endpoints) {
+ const failureCount = (this.#heartbeatFailureCounts.get(endpoints.facade)
?? 0) + 1;
+ this.#heartbeatFailureCounts.set(endpoints.facade, failureCount);
+ if (failureCount < HEARTBEAT_TRANSPORT_RECOVERY_THRESHOLD) {
+ return;
+ }
+ const now = Date.now();
+ if (now - (this.#lastTransportRecoveryTimes.get(endpoints.facade) ?? 0) <
TRANSPORT_RECOVERY_COOLDOWN) {
+ debug('Transport recovery throttled by cooldown, endpoints=%s,
clientId=%s',
+ endpoints.facade, this.clientId);
+ return;
+ }
+ this.#lastTransportRecoveryTimes.set(endpoints.facade, now);
+ this.logger.warn('Consecutive heartbeat failures detected, rebuilding
transport layer, endpoints=%s, failureCount=%d, clientId=%s',
+ endpoints.facade, failureCount, this.clientId);
+ // 1. Evict the possibly stale RpcClient so a fresh channel is established
on next use.
+ try {
+ this.rpcClientManager.evict(endpoints);
+ } catch (e) {
+ this.logger.warn('Failed to evict rpc client, endpoints=%s, clientId=%s,
error=%s',
+ endpoints.facade, this.clientId, e instanceof Error ? e.message :
String(e));
+ }
+ // 2. Release and drop the telemetry session, then eagerly rebuild it to
re-sync settings.
+ const session = this.#telemetrySessions.get(endpoints.facade);
+ if (session) {
+ session.release();
+ this.#telemetrySessions.delete(endpoints.facade);
+ }
+ try {
+ this.getTelemetrySession(endpoints).syncSettings();
+ } catch (e) {
+ this.logger.warn('Failed to rebuild telemetry session, endpoints=%s,
clientId=%s, error=%s',
+ endpoints.facade, this.clientId, e instanceof Error ? e.message :
String(e));
+ }
+ }
+
#getTotalRouteEndpointsMap() {
const endpointsMap = new Map<string, Endpoints>();
for (const topicRoute of this.topicRouteCache.values()) {
@@ -518,9 +581,11 @@ export abstract class BaseClient {
const obj = command.toObject();
this.logger.warn('Ignore verify message command from remote, which is not
expected, clientId=%s, command=%j',
this.clientId, obj);
+ // Respond with VerifyMessageResult carrying the same nonce, mirroring the
Java
+ // client (BaseClient#onVerifyMessageCommand), instead of echoing the
command.
const telemetryCommand = new TelemetryCommand();
telemetryCommand.setStatus(new Status().setCode(Code.NOT_IMPLEMENTED));
- telemetryCommand.setVerifyMessageCommand(new
VerifyMessageCommand().setNonce(obj.nonce));
+ telemetryCommand.setVerifyMessageResult(new
VerifyMessageResult().setNonce(obj.nonce));
this.telemetry(endpoints, telemetryCommand);
}
@@ -531,15 +596,44 @@ export abstract class BaseClient {
onPrintThreadStackTraceCommand(endpoints: Endpoints, command:
PrintThreadStackTraceCommand) {
const obj = command.toObject();
- this.logger.warn('Ignore orphaned transaction recovery command from
remote, which is not expected, clientId=%s, command=%j',
+ this.logger.info('Received print thread stack trace command from remote,
clientId=%s, command=%j',
this.clientId, obj);
- const nonce = obj.nonce;
const telemetryCommand = new TelemetryCommand();
- telemetryCommand.setThreadStackTrace(new
ThreadStackTrace().setThreadStackTrace('mock stack').setNonce(nonce));
+ telemetryCommand.setThreadStackTrace(new ThreadStackTrace()
+ .setThreadStackTrace(this.#buildProcessDiagnostics())
+ .setNonce(obj.nonce));
telemetryCommand.setStatus(new Status().setCode(Code.OK));
this.telemetry(endpoints, telemetryCommand);
}
+ /**
+ * Build Node.js process diagnostics in place of Java-style thread stack
traces.
+ * Java sends per-thread stacks via ThreadMXBean; Node.js is single-threaded
per
+ * process, so we expose the closest equivalent runtime snapshot.
+ */
+ #buildProcessDiagnostics(): string {
+ const mem = process.memoryUsage();
+ const formatBytes = (bytes: number) => `${(bytes / 1024 /
1024).toFixed(2)}MB`;
+ const lines = [
+ `Process: pid=${process.pid}, node=${process.version},
platform=${process.platform}, arch=${process.arch}`,
+ `Uptime: ${process.uptime().toFixed(3)}s`,
+ `Memory: rss=${formatBytes(mem.rss)},
heapUsed=${formatBytes(mem.heapUsed)}, ` +
+ `heapTotal=${formatBytes(mem.heapTotal)},
external=${formatBytes(mem.external)}, ` +
+ `arrayBuffers=${formatBytes(mem.arrayBuffers)}`,
+ ];
+ try {
+ const activeHandles = (process as unknown as { _getActiveHandles?: () =>
object[] })._getActiveHandles?.() ?? [];
+ lines.push(`Active handles: ${activeHandles.length}`);
+ for (const handle of activeHandles.slice(0, 20)) {
+ const name = handle?.constructor?.name ?? typeof handle;
+ lines.push(` - ${name}`);
+ }
+ } catch {
+ // active handles are best-effort
+ }
+ return lines.join('\n');
+ }
+
// eslint-disable-next-line @typescript-eslint/no-unused-vars
onReconnectEndpointsCommand(endpoints: Endpoints, _command:
ReconnectEndpointsCommand) {
this.logger.info('Received reconnect endpoints command from remote, will
refresh telemetry session, endpoints=%s, clientId=%s',
diff --git a/nodejs/src/client/RpcClient.ts b/nodejs/src/client/RpcClient.ts
index 78b15327..2fa35d73 100644
--- a/nodejs/src/client/RpcClient.ts
+++ b/nodejs/src/client/RpcClient.ts
@@ -53,6 +53,19 @@ import {
} from '../../proto/apache/rocketmq/v2/service_pb';
import { Endpoints } from '../route';
+// Channel options mirroring the Java client's Netty channel configuration:
+// keepalive probing plus unbounded message sizes (Java defaults to
Integer.MAX_VALUE).
+const GRPC_CHANNEL_OPTIONS = {
+ 'grpc.keepalive_time_ms': 300000, // 5 minutes, same as Java
DEFAULT_KEEP_ALIVE_TIME_MILLIS
+ 'grpc.keepalive_timeout_ms': 30000, // 30 seconds, same as Java
DEFAULT_KEEP_ALIVE_TIMEOUT_MILLIS
+ 'grpc.keepalive_permit_without_calls': 1,
+ 'grpc.max_send_message_length': 2 ** 31 - 1,
+ 'grpc.max_receive_message_length': 2 ** 31 - 1,
+ // Use a local subchannel pool so each RpcClient owns its own connection
+ // instead of sharing one via grpc-js's global pool, aligning with the Java
client.
+ 'grpc.use_local_subchannel_pool': 1,
+};
+
export class RpcClient {
#client: MessagingServiceClient;
#activityTime = Date.now();
@@ -60,11 +73,7 @@ export class RpcClient {
constructor(endpoints: Endpoints, sslEnabled: boolean) {
const address = endpoints.getGrpcTarget();
const grpcCredentials = sslEnabled ? ChannelCredentials.createSsl() :
ChannelCredentials.createInsecure();
- // Use a local subchannel pool so each RpcClient owns its own connection
- // instead of sharing one via grpc-js's global pool, aligning with the
Java client.
- this.#client = new MessagingServiceClient(address, grpcCredentials, {
- 'grpc.use_local_subchannel_pool': 1,
- });
+ this.#client = new MessagingServiceClient(address, grpcCredentials,
GRPC_CHANNEL_OPTIONS);
}
#getAndActivityRpcClient() {
diff --git a/nodejs/src/client/RpcClientManager.ts
b/nodejs/src/client/RpcClientManager.ts
index c363eb71..5740cc46 100644
--- a/nodejs/src/client/RpcClientManager.ts
+++ b/nodejs/src/client/RpcClientManager.ts
@@ -37,7 +37,10 @@ const RPC_CLIENT_MAX_IDLE_DURATION = 30 * 60000; // 30
minutes
const RPC_CLIENT_IDLE_CHECK_PERIOD = 60000;
export class RpcClientManager {
- #rpcClients = new Map<Endpoints, RpcClient>();
+ // Keyed by endpoints.facade (string) instead of the Endpoints object itself:
+ // Endpoints instances are recreated on every route fetch, so object keys
would
+ // leak duplicate RpcClients for logically identical endpoints.
+ #rpcClients = new Map<string, RpcClient>();
#baseClient: BaseClient;
#logger: ILogger;
#clearIdleRpcClientsTimer: NodeJS.Timeout;
@@ -55,26 +58,42 @@ export class RpcClientManager {
}
#clearIdleRpcClients() {
- for (const [ endpoints, rpcClient ] of this.#rpcClients.entries()) {
+ for (const [ facade, rpcClient ] of this.#rpcClients.entries()) {
const idleDuration = rpcClient.idleDuration();
if (idleDuration > RPC_CLIENT_MAX_IDLE_DURATION) {
rpcClient.close();
- this.#rpcClients.delete(endpoints);
+ this.#rpcClients.delete(facade);
this.#logger.info('[RpcClientManager] Rpc client has been idle for a
long time, endpoints=%s, idleDuration=%s, clientId=%s',
- endpoints, idleDuration, RPC_CLIENT_MAX_IDLE_DURATION,
this.#baseClient.clientId);
+ facade, idleDuration, RPC_CLIENT_MAX_IDLE_DURATION,
this.#baseClient.clientId);
}
}
}
#getRpcClient(endpoints: Endpoints) {
- let rpcClient = this.#rpcClients.get(endpoints);
+ const facade = endpoints.facade;
+ let rpcClient = this.#rpcClients.get(facade);
if (!rpcClient) {
rpcClient = new RpcClient(endpoints, this.#baseClient.sslEnabled);
- this.#rpcClients.set(endpoints, rpcClient);
+ this.#rpcClients.set(facade, rpcClient);
}
return rpcClient;
}
+ /**
+ * Close and remove the RPC client bound to the given endpoints (e.g. after
+ * consecutive heartbeat failures), so a fresh channel is established on
next use.
+ */
+ evict(endpoints: Endpoints) {
+ const facade = endpoints.facade;
+ const rpcClient = this.#rpcClients.get(facade);
+ if (rpcClient) {
+ rpcClient.close();
+ this.#rpcClients.delete(facade);
+ this.#logger.info('[RpcClientManager] Rpc client evicted, endpoints=%s,
clientId=%s',
+ facade, this.#baseClient.clientId);
+ }
+ }
+
close() {
// Clear idle check timer first
if (this.#clearIdleRpcClientsTimer) {
@@ -82,12 +101,12 @@ export class RpcClientManager {
}
// Close all RPC clients and clear the map
- for (const [ endpoints, rpcClient ] of this.#rpcClients.entries()) {
+ for (const [ facade, rpcClient ] of this.#rpcClients.entries()) {
try {
rpcClient.close();
} catch (e) {
this.#logger.warn('Failed to close RPC client for endpoints=%s,
clientId=%s, error=%s',
- endpoints.facade, this.#baseClient.clientId, e instanceof Error ?
e.message : String(e));
+ facade, this.#baseClient.clientId, e instanceof Error ? e.message :
String(e));
}
}
this.#rpcClients.clear();
diff --git a/nodejs/src/route/Endpoints.ts b/nodejs/src/route/Endpoints.ts
index 840faf5e..210a98e4 100644
--- a/nodejs/src/route/Endpoints.ts
+++ b/nodejs/src/route/Endpoints.ts
@@ -78,8 +78,27 @@ export class Endpoints {
this.facade = this.addressesList.map(addr =>
`${addr.host}:${addr.port}`).join(',');
}
+ /**
+ * gRPC target with resolver scheme prefix, mirroring the Java client:
+ * - IPv4 addresses: ipv4:127.0.0.1:10911,127.0.0.2:10912
+ * - IPv6 addresses: ipv6:[::1]:10911,[fe80::1]:10912 (brackets required by
grpc-js)
+ * - Domain names: dns:example.com:8080,example.org:8081
+ */
getGrpcTarget() {
- return this.facade;
+ const targets = this.addressesList.map(addr => {
+ const host = this.scheme === AddressScheme.IPV6 ? `[${addr.host}]` :
addr.host;
+ return `${host}:${addr.port}`;
+ }).join(',');
+ switch (this.scheme) {
+ case AddressScheme.IPV4:
+ return `ipv4:${targets}`;
+ case AddressScheme.IPV6:
+ return `ipv6:${targets}`;
+ case AddressScheme.DOMAIN_NAME:
+ return `dns:${targets}`;
+ default:
+ return targets;
+ }
}
toString() {
diff --git a/nodejs/test/route/Endpoints.test.ts
b/nodejs/test/route/Endpoints.test.ts
new file mode 100644
index 00000000..bbf74fde
--- /dev/null
+++ b/nodejs/test/route/Endpoints.test.ts
@@ -0,0 +1,56 @@
+/**
+ * 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 gRPC target resolver scheme fix (B-1):
+ * getGrpcTarget() must prefix the scheme (ipv4:/ipv6:/dns:) so grpc-js
+ * resolves multi-address and bare-IPv6 targets correctly.
+ */
+
+import { describe, it } from 'node:test';
+import * as assert from 'node:assert';
+import { Endpoints } from '../../src/route';
+import {
+ AddressScheme,
+ Endpoints as EndpointsPB,
+} from '../../proto/apache/rocketmq/v2/definition_pb';
+
+function endpointsFromPb(scheme: AddressScheme, host: string, port: number):
Endpoints {
+ const pb = new EndpointsPB();
+ pb.setScheme(scheme);
+ pb.addAddresses().setHost(host).setPort(port);
+ return new Endpoints(pb.toObject());
+}
+
+describe('Endpoints.getGrpcTarget with resolver scheme (B-1)', () => {
+ it('should prefix ipv4: scheme for IPv4 addresses', () => {
+ assert.strictEqual(new Endpoints('127.0.0.1:10911').getGrpcTarget(),
'ipv4:127.0.0.1:10911');
+ });
+
+ it('should prefix ipv4: scheme for multiple IPv4 addresses', () => {
+ const target = new
Endpoints('127.0.0.1:8081;127.0.0.2:8082').getGrpcTarget();
+ assert.strictEqual(target, 'ipv4:127.0.0.1:8081,127.0.0.2:8082');
+ });
+
+ it('should prefix ipv6: scheme with brackets for IPv6 addresses', () => {
+ assert.strictEqual(endpointsFromPb(AddressScheme.IPV6, '::1',
10911).getGrpcTarget(), 'ipv6:[::1]:10911');
+ });
+
+ it('should prefix dns: scheme for domain names', () => {
+ assert.strictEqual(new Endpoints('example.com:8080').getGrpcTarget(),
'dns:example.com:8080');
+ });
+});
