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 13b9299e Feat/nodejs ip name resolver (#1385)
13b9299e is described below
commit 13b9299e5a80d53cf5b448e5a0590ff6ebecba1a
Author: zhaohai <[email protected]>
AuthorDate: Fri Oct 9 10:14:53 2026 +0800
Feat/nodejs ip name resolver (#1385)
* fix(nodejs): transport layer self-healing and logger close deferral
- 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'
* feat(nodejs): add custom ip: gRPC name resolver for endpoints targets
Register an `ip` scheme resolver (src/route/IpNameResolver.ts) so that
multi-address proxy endpoints are resolved as a static address group with
shuffle-based load balancing across proxies, instead of relying on the
default DNS resolution.
Endpoints.getGrpcTarget() now emits `ip:` for both IPv4 and IPv6
addresses; the resolver brackets bare IPv6 hosts for grpc-js.
---
nodejs/src/route/Endpoints.ts | 14 ++--
nodejs/src/route/IpNameResolver.ts | 118 +++++++++++++++++++++++++++++
nodejs/test/route/Endpoints.test.ts | 21 +++---
nodejs/test/route/IpNameResolver.test.ts | 124 +++++++++++++++++++++++++++++++
4 files changed, 262 insertions(+), 15 deletions(-)
diff --git a/nodejs/src/route/Endpoints.ts b/nodejs/src/route/Endpoints.ts
index 210a98e4..dff4fe03 100644
--- a/nodejs/src/route/Endpoints.ts
+++ b/nodejs/src/route/Endpoints.ts
@@ -18,6 +18,9 @@
import { isIPv4, isIPv6 } from 'node:net';
import { hashCodeOfString } from '../util';
import { Address, AddressScheme, Endpoints as EndpointsPB } from
'../../proto/apache/rocketmq/v2/definition_pb';
+// Side-effect import: registers the custom `ip` name resolver (shuffle +
multi-address
+// load balancing), mirroring the Java client's IpNameResolverFactory.
+import './IpNameResolver';
const DEFAULT_PORT = 80;
@@ -79,9 +82,8 @@ export class Endpoints {
}
/**
- * 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)
+ * gRPC target with resolver scheme prefix:
+ * - IP addresses: ip:127.0.0.1:10911,127.0.0.2:10912 (custom resolver,
IPv6 hosts bracketed)
* - Domain names: dns:example.com:8080,example.org:8081
*/
getGrpcTarget() {
@@ -91,9 +93,11 @@ export class Endpoints {
}).join(',');
switch (this.scheme) {
case AddressScheme.IPV4:
- return `ipv4:${targets}`;
case AddressScheme.IPV6:
- return `ipv6:${targets}`;
+ // Route IP addresses through the custom `ip` resolver registered in
+ // IpNameResolver, which shuffles the address list for multi-proxy load
+ // balancing (equivalent to Java's IpNameResolverFactory).
+ return `ip:${targets}`;
case AddressScheme.DOMAIN_NAME:
return `dns:${targets}`;
default:
diff --git a/nodejs/src/route/IpNameResolver.ts
b/nodejs/src/route/IpNameResolver.ts
new file mode 100644
index 00000000..d2bb2c60
--- /dev/null
+++ b/nodejs/src/route/IpNameResolver.ts
@@ -0,0 +1,118 @@
+/**
+ * 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.
+ */
+
+import { experimental, Metadata, status as Status, ChannelOptions } from
'@grpc/grpc-js';
+
+/**
+ * Custom gRPC name resolver registered under the `ip` scheme, mirroring the
Java
+ * client's `IpNameResolverFactory`.
+ *
+ * The Java resolver converts each socket address into its own
+ * `EquivalentAddressGroup` and then {@code Collections.shuffle}s the list, so
the
+ * gRPC load balancer starts from a random address instead of always the first
+ * one. RocketMQ proxies are typically deployed as multiple endpoints behind
one
+ * `Endpoints`, and shuffling spreads clients across them.
+ *
+ * grpc-js models an `EquivalentAddressGroup` as a single `Endpoint` carrying
one
+ * or more equivalent `SubchannelAddress`es, so we emit one `Endpoint` per
address
+ * and shuffle the array — exactly the Java semantics.
+ */
+const DEFAULT_PORT = 80;
+
+/**
+ * Fisher–Yates shuffle (pure, does not mutate the input array).
+ */
+function shuffle<T>(input: readonly T[]): T[] {
+ const arr = input.slice();
+ for (let i = arr.length - 1; i > 0; i--) {
+ const j = Math.floor(Math.random() * (i + 1));
+ const tmp = arr[i];
+ arr[i] = arr[j];
+ arr[j] = tmp;
+ }
+ return arr;
+}
+
+class IpResolver {
+ private readonly listener: experimental.ResolverListener;
+ private endpoints: experimental.Endpoint[] = [];
+ private error: { code: Status; details: string; metadata: Metadata } | null
= null;
+ private hasReturnedResult = false;
+
+ constructor(target: experimental.GrpcUri, listener:
experimental.ResolverListener, _channelOptions: ChannelOptions) {
+ this.listener = listener;
+ const pathList = (target.path || '').split(',').map(s =>
s.trim()).filter(Boolean);
+ const addresses: experimental.SubchannelAddress[] = [];
+ for (const path of pathList) {
+ const hostPort = experimental.splitHostPort(path);
+ if (!hostPort) {
+ this.error = {
+ code: Status.UNAVAILABLE,
+ details: `Failed to parse ip address: ${path}`,
+ metadata: new Metadata(),
+ };
+ return;
+ }
+ addresses.push({
+ host: hostPort.host,
+ port: hostPort.port ?? DEFAULT_PORT,
+ });
+ }
+ this.endpoints = shuffle(addresses).map(address => ({ addresses: [address]
}));
+ }
+
+ updateResolution(): void {
+ if (this.hasReturnedResult) {
+ return;
+ }
+ this.hasReturnedResult = true;
+ // Defer the callback to the next tick, mirroring the built-in ip resolver
so
+ // listeners are never invoked synchronously from the constructor path.
+ process.nextTick(() => {
+ if (this.error) {
+ this.listener(
+ experimental.statusOrFromError(this.error),
+ {},
+ null,
+ '',
+ );
+ } else {
+ this.listener(
+ experimental.statusOrFromValue(this.endpoints),
+ {},
+ null,
+ '',
+ );
+ }
+ });
+ }
+
+ destroy(): void {
+ this.hasReturnedResult = false;
+ }
+
+ // eslint-disable-next-line @typescript-eslint/naming-convention
+ static getDefaultAuthority(target: experimental.GrpcUri): string {
+ return (target.path || '').split(',')[0] || 'localhost';
+ }
+}
+
+// Register the resolver so that gRPC targets prefixed with `ip:` are handled
by
+// this class. Registration is idempotent at module-load time.
+experimental.registerResolver('ip', IpResolver);
+
+export { IpResolver };
diff --git a/nodejs/test/route/Endpoints.test.ts
b/nodejs/test/route/Endpoints.test.ts
index bbf74fde..d8d692ec 100644
--- a/nodejs/test/route/Endpoints.test.ts
+++ b/nodejs/test/route/Endpoints.test.ts
@@ -16,9 +16,10 @@
*/
/**
- * 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.
+ * Tests for the gRPC target resolver scheme: getGrpcTarget() routes IP
+ * addresses through the custom `ip` name resolver (registered by
+ * IpNameResolver) and prefixes domain names with dns:, so grpc-js resolves
+ * multi-address and bare-IPv6 targets correctly.
*/
import { describe, it } from 'node:test';
@@ -36,18 +37,18 @@ function endpointsFromPb(scheme: AddressScheme, host:
string, port: number): End
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');
+describe('Endpoints.getGrpcTarget with resolver scheme (custom ip resolver)',
() => {
+ it('should prefix ip: scheme for IPv4 addresses', () => {
+ assert.strictEqual(new Endpoints('127.0.0.1:10911').getGrpcTarget(),
'ip:127.0.0.1:10911');
});
- it('should prefix ipv4: scheme for multiple IPv4 addresses', () => {
+ it('should prefix ip: 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');
+ assert.strictEqual(target, 'ip: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 ip: scheme with brackets for IPv6 addresses', () => {
+ assert.strictEqual(endpointsFromPb(AddressScheme.IPV6, '::1',
10911).getGrpcTarget(), 'ip:[::1]:10911');
});
it('should prefix dns: scheme for domain names', () => {
diff --git a/nodejs/test/route/IpNameResolver.test.ts
b/nodejs/test/route/IpNameResolver.test.ts
new file mode 100644
index 00000000..a76d8f0d
--- /dev/null
+++ b/nodejs/test/route/IpNameResolver.test.ts
@@ -0,0 +1,124 @@
+/**
+ * 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.
+ */
+
+import { describe, it } from 'node:test';
+import * as assert from 'node:assert';
+import { Endpoints } from '../../src/route';
+import { IpResolver } from '../../src/route/IpNameResolver';
+import {
+ AddressScheme,
+ Endpoints as EndpointsPB,
+} from '../../proto/apache/rocketmq/v2/definition_pb';
+
+type Endpoint = { addresses: { host: string; port: number }[] };
+
+function makeTarget(path: string) {
+ return { scheme: 'ip', authority: '', path } as any;
+}
+
+/**
+ * Drive the resolver with a capturing listener and return the resolved
endpoints.
+ * The listener is invoked on nextTick, so we await it.
+ */
+function resolve(path: string): Promise<Endpoint[]> {
+ return new Promise((resolvePromise, rejectPromise) => {
+ const listener = (endpointList: any, _attrs: any, _cfg: any, _note:
string) => {
+ if (endpointList.ok) {
+ resolvePromise(endpointList.value as Endpoint[]);
+ } else {
+ rejectPromise(new Error(endpointList.error.details));
+ }
+ };
+ const resolver = new IpResolver(makeTarget(path), listener as any, {} as
any);
+ resolver.updateResolution();
+ });
+}
+
+function toKeySet(eps: Endpoint[]): Set<string> {
+ return new Set(eps.map(e =>
`${e.addresses[0].host}:${e.addresses[0].port}`));
+}
+
+describe('IpNameResolver (custom ip scheme, mirrors Java
IpNameResolverFactory)', () => {
+ it('parses comma-separated host:port and resolves every address', async ()
=> {
+ const eps = await resolve('127.0.0.1:10911,127.0.0.2:10912');
+ assert.strictEqual(eps.length, 2);
+ const keys = toKeySet(eps);
+ assert.ok(keys.has('127.0.0.1:10911'));
+ assert.ok(keys.has('127.0.0.2:10912'));
+ for (const ep of eps) {
+ assert.strictEqual(ep.addresses.length, 1, 'each address becomes its own
EquivalentAddressGroup');
+ }
+ });
+
+ it('handles bracketed IPv6 addresses', async () => {
+ const eps = await resolve('[::1]:10911,[fe80::1]:10912');
+ const keys = toKeySet(eps);
+ assert.ok(keys.has('::1:10911'));
+ assert.ok(keys.has('fe80::1:10912'));
+ });
+
+ it('applies default port 80 when omitted', async () => {
+ const eps = await resolve('127.0.0.1');
+ assert.strictEqual(eps.length, 1);
+ assert.strictEqual(eps[0].addresses[0].port, 80);
+ });
+
+ it('reports an error for an unparsable address', async () => {
+ await assert.rejects(() => resolve('localhost:notaport'));
+ });
+
+ it('shuffles the address order (Fisher-Yates)', async () => {
+ const original = Math.random;
+ // Force a deterministic Fisher-Yates permutation. With random=0:
+ // input [a,b,c] -> i=2: swap(2,0)=[c,b,a] -> i=1: swap(1,0)=[b,c,a]
+ Math.random = () => 0;
+ try {
+ const eps = await resolve('10.0.0.1:80,10.0.0.2:80,10.0.0.3:80');
+ const order = eps.map(e => e.addresses[0].host);
+ assert.deepStrictEqual(order, ['10.0.0.2', '10.0.0.3', '10.0.0.1']);
+ } finally {
+ Math.random = original;
+ }
+ });
+
+ it('returns a permutation (all original addresses, no duplicates) across
runs', async () => {
+ const input = '10.0.0.1:80,10.0.0.2:80,10.0.0.3:80,10.0.0.4:80';
+ const expected = new Set(['10.0.0.1:80', '10.0.0.2:80', '10.0.0.3:80',
'10.0.0.4:80']);
+ for (let i = 0; i < 20; i++) {
+ const eps = await resolve(input);
+ assert.strictEqual(eps.length, 4);
+ const keys = toKeySet(eps);
+ assert.deepStrictEqual(keys, expected);
+ }
+ });
+});
+
+describe('Endpoints.getGrpcTarget uses the custom ip scheme', () => {
+ it('emits ip: scheme for IPv4/IPv6 endpoints', () => {
+ assert.ok(new
Endpoints('127.0.0.1:10911;127.0.0.2:10912').getGrpcTarget().startsWith('ip:'));
+ // Bracketed IPv6 string parsing lands with the endpoints-parsing fix;
build
+ // the IPv6 endpoints from protobuf so this test only covers the scheme.
+ const pb = new EndpointsPB();
+ pb.setScheme(AddressScheme.IPV6);
+ pb.addAddresses().setHost('::1').setPort(10911);
+ assert.ok(new Endpoints(pb.toObject()).getGrpcTarget().startsWith('ip:'));
+ });
+
+ it('keeps dns: scheme for domain names', () => {
+ assert.ok(new
Endpoints('example.com:8080;example.org:8081').getGrpcTarget().startsWith('dns:'));
+ });
+});
