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 4123af49 fix(nodejs): correct message-group routing and consumption
lifecycle guards (#1388)
4123af49 is described below
commit 4123af49bcef731746fe3fdfc1823ba01f10f80e
Author: zhaohai <[email protected]>
AuthorDate: Fri Oct 9 10:15:52 2026 +0800
fix(nodejs): correct message-group routing and consumption lifecycle guards
(#1388)
* fix(nodejs): correct message-group routing and consumption lifecycle
guards
- PublishingLoadBalancer: interpret SipHash-2-4 as signed int64 and apply
floorMod for message-group routing (the previous unsigned modulo routed
'negative' hashes to different queues)
- PushConsumer: bound the cached-message drain wait by
requestTimeout + longPollingTimeout so shutdown cannot hang forever on
a stuck consumption chain
- SimpleConsumer: guard receive/ack/changeInvisibleDuration with
isRunning() checks
- LiteSubscriptionManager: sync lite subscriptions to all route endpoints
(not only the client's configured endpoints); expose
LitePushConsumerImpl#getTotalRouteEndpoints()
* chore(nodejs): untrack the local scratch telemetry test
It was picked up accidentally by a broad `git add` during the master merge
and has no place in the repository; license-eye rejects it for the missing
ASF header.
---
nodejs/src/consumer/LitePushConsumerImpl.ts | 7 +-
nodejs/src/consumer/LiteSubscriptionManager.ts | 2 +-
nodejs/src/consumer/PushConsumer.ts | 13 +++
nodejs/src/consumer/SimpleConsumer.ts | 9 +++
nodejs/src/producer/PublishingLoadBalancer.ts | 10 ++-
.../test/producer/PublishingLoadBalancer.test.ts | 93 ++++++++++++++++++++++
6 files changed, 130 insertions(+), 4 deletions(-)
diff --git a/nodejs/src/consumer/LitePushConsumerImpl.ts
b/nodejs/src/consumer/LitePushConsumerImpl.ts
index 158a883d..7693cb89 100644
--- a/nodejs/src/consumer/LitePushConsumerImpl.ts
+++ b/nodejs/src/consumer/LitePushConsumerImpl.ts
@@ -283,9 +283,14 @@ export class LitePushConsumerImpl extends PushConsumer
implements LitePushConsum
/**
* Endpoints the lite subscription manager should sync to.
*
+ * Lite subscriptions must reach every proxy serving the topic route, not
only
+ * the endpoint the client was configured with, so the route cache is used
and
+ * the configured endpoints act as the fallback before any route is known.
+ *
* @internal
*/
getSyncEndpoints(): Endpoints[] {
- return [ this.endpoints ];
+ const endpointsList = this.getTotalRouteEndpoints();
+ return endpointsList.length > 0 ? endpointsList : [ this.endpoints ];
}
}
diff --git a/nodejs/src/consumer/LiteSubscriptionManager.ts
b/nodejs/src/consumer/LiteSubscriptionManager.ts
index 179bad71..00084a6a 100644
--- a/nodejs/src/consumer/LiteSubscriptionManager.ts
+++ b/nodejs/src/consumer/LiteSubscriptionManager.ts
@@ -297,7 +297,7 @@ export class LiteSubscriptionManager {
if (status && status.getCode() !== Code.OK) {
throw new ClientException(
status.getCode(),
- `Failed to sync lite subscription: ${status.getMessage()}`,
+ `Failed to sync lite subscription to
endpoints=${endpoints.facade}: ${status.getMessage()}`,
);
}
}));
diff --git a/nodejs/src/consumer/PushConsumer.ts
b/nodejs/src/consumer/PushConsumer.ts
index 4cf5f8e4..f680425c 100644
--- a/nodejs/src/consumer/PushConsumer.ts
+++ b/nodejs/src/consumer/PushConsumer.ts
@@ -168,7 +168,20 @@ export class PushConsumer extends Consumer {
// Every received message is cached before consumption and evicted from
// the cache once its consumption chain settles, so empty caches on all
// process queues mean consumption has fully quiesced.
+ // Bound the wait with the same timeout as the inflight-receive drain
+ // (request timeout + long-polling timeout) so shutdown cannot hang forever
+ // on a stuck consumption chain.
+ const maxWaitingTime = this.requestTimeoutValue
+ + this.getPushConsumerSettings().getLongPollingTimeout();
+ const endTime = Date.now() + maxWaitingTime;
while ([ ...this.#processQueueTable.values() ].some(({ pq }) =>
pq.cachedMessagesCount() > 0)) {
+ if (Date.now() > endTime) {
+ const remaining = [ ...this.#processQueueTable.values() ]
+ .reduce((count, { pq }) => count + pq.cachedMessagesCount(), 0);
+ this.logger.warn('Timeout waiting for the cached messages to be
consumed, clientId=%s, '
+ + 'remainingCachedMessages=%d', this.clientId, remaining);
+ return;
+ }
await this.sleep(100);
}
}
diff --git a/nodejs/src/consumer/SimpleConsumer.ts
b/nodejs/src/consumer/SimpleConsumer.ts
index 1ba6654d..788ed63f 100644
--- a/nodejs/src/consumer/SimpleConsumer.ts
+++ b/nodejs/src/consumer/SimpleConsumer.ts
@@ -134,6 +134,9 @@ export class SimpleConsumer extends Consumer {
}
async receive(maxMessageNum = 10, invisibleDuration = 15000) {
+ if (!this.isRunning()) {
+ throw new Error('Simple consumer is not running, please call startup()
first or check shutdown state');
+ }
if (maxMessageNum <= 0) {
throw new Error(`maxMessageNum must be greater than 0, but got
${maxMessageNum}`);
}
@@ -177,10 +180,16 @@ export class SimpleConsumer extends Consumer {
}
async ack(message: MessageView) {
+ if (!this.isRunning()) {
+ throw new Error('Simple consumer is not running, cannot ack message');
+ }
await this.ackMessage(message);
}
async changeInvisibleDuration(message: MessageView, invisibleDuration:
number) {
+ if (!this.isRunning()) {
+ throw new Error('Simple consumer is not running, cannot change invisible
duration');
+ }
const response = await this.invisibleDuration(message, invisibleDuration);
// Refresh receipt handle manually
(message as any).receiptHandle = response;
diff --git a/nodejs/src/producer/PublishingLoadBalancer.ts
b/nodejs/src/producer/PublishingLoadBalancer.ts
index c4061256..9039573f 100644
--- a/nodejs/src/producer/PublishingLoadBalancer.ts
+++ b/nodejs/src/producer/PublishingLoadBalancer.ts
@@ -78,8 +78,14 @@ export class PublishingLoadBalancer {
}
takeMessageQueueByMessageGroup(messageGroup: string) {
- const hashCode = calculateStringSipHash24(messageGroup);
- const index = parseInt(`${hashCode % BigInt(this.#messageQueues.length)}`);
+ // SipHash-2-4 returns an unsigned 64-bit value in JS (readBigUInt64BE),
while
+ // Java's String.hashCode-style paths use signed semantics. Interpret the
hash as
+ // a signed 64-bit integer and apply floorMod so the result is always
non-negative,
+ // exactly matching Java LongMath.mod(hashCode, size) — otherwise FIFO
messages
+ // with "negative" hashes route to different queues than the Java client.
+ const hashCode = BigInt.asIntN(64, calculateStringSipHash24(messageGroup));
+ const size = this.#messageQueues.length;
+ const index = Number(((hashCode % BigInt(size)) + BigInt(size)) %
BigInt(size));
return this.#messageQueues[index];
}
diff --git a/nodejs/test/producer/PublishingLoadBalancer.test.ts
b/nodejs/test/producer/PublishingLoadBalancer.test.ts
new file mode 100644
index 00000000..aaf6de41
--- /dev/null
+++ b/nodejs/test/producer/PublishingLoadBalancer.test.ts
@@ -0,0 +1,93 @@
+/**
+ * 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 FIFO message-group routing fix (C-1):
+ * SipHash-2-4 must be interpreted as a signed int64 and combined with
+ * floorMod, matching Java LongMath.mod, otherwise "negative" hashes are
+ * routed to different queues than the Java client.
+ */
+
+import { describe, it } from 'node:test';
+import * as assert from 'node:assert';
+import { PublishingLoadBalancer } from '../../src/producer';
+import { TopicRouteData } from '../../src/route';
+import { calculateStringSipHash24 } from '../../src/util';
+import {
+ MessageQueue as MessageQueuePB,
+ Broker as BrokerPB,
+ Resource as ResourcePB,
+ Endpoints as EndpointsPB,
+ Permission,
+ AddressScheme,
+} from '../../proto/apache/rocketmq/v2/definition_pb';
+
+describe('takeMessageQueueByMessageGroup floorMod semantics (C-1)', () => {
+ // One queue (queueId=0, the writable master queue) per broker across
queueCount brokers,
+ // since PublishingLoadBalancer only keeps queues with queueId ===
MASTER_BROKER_ID (0).
+ // The returned queue is identified by its broker endpoints port (10911 +
expected index).
+ function buildLoadBalancer(queueCount: number): PublishingLoadBalancer {
+ const pbs: MessageQueuePB[] = [];
+ for (let i = 0; i < queueCount; i++) {
+ const endpointsPb = new EndpointsPB();
+ endpointsPb.setScheme(AddressScheme.IPV4);
+ endpointsPb.addAddresses().setHost('127.0.0.1').setPort(10911 + i);
+ const brokerPb = new BrokerPB();
+ brokerPb.setName('broker-' + i);
+ brokerPb.setId(0);
+ brokerPb.setEndpoints(endpointsPb);
+ const mqPb = new MessageQueuePB();
+ mqPb.setId(0);
+ mqPb.setTopic(new ResourcePB().setName('topic'));
+ mqPb.setBroker(brokerPb);
+ mqPb.setPermission(Permission.READ_WRITE);
+ pbs.push(mqPb);
+ }
+ return new PublishingLoadBalancer(new TopicRouteData(pbs));
+ }
+
+ it('should always yield a non-negative index matching Java LongMath.mod', ()
=> {
+ const loadBalancer = buildLoadBalancer(8);
+ const groups = ['group-a', 'fifo-group', 'group-中文', 'x', 'order-12345',
''];
+ for (const group of groups) {
+ const mq = loadBalancer.takeMessageQueueByMessageGroup(group);
+ assert.ok(mq, 'should resolve a message queue for group=' + group);
+ // Expected index under floorMod semantics: ((signed hash % size) +
size) % size
+ const signedHash = BigInt.asIntN(64, calculateStringSipHash24(group));
+ const expectedIndex = Number(((signedHash % 8n) + 8n) % 8n);
+ assert.strictEqual(mq.broker.endpoints.facade, '127.0.0.1:' + (10911 +
expectedIndex));
+ }
+ });
+
+ it('should match Java floorMod for hashes that are negative when interpreted
signed', () => {
+ // Find a message group whose SipHash-2-4 has its high bit set (negative
as int64)
+ let negativeGroup: string | undefined;
+ for (let i = 0; i < 10000; i++) {
+ const candidate = 'probe-' + i;
+ if (BigInt.asIntN(64, calculateStringSipHash24(candidate)) < 0n) {
+ negativeGroup = candidate;
+ break;
+ }
+ }
+ assert.ok(negativeGroup, 'expected to find a negative hash among probes');
+ const loadBalancer = buildLoadBalancer(4);
+ const mq = loadBalancer.takeMessageQueueByMessageGroup(negativeGroup!);
+ const signedHash = BigInt.asIntN(64,
calculateStringSipHash24(negativeGroup!));
+ const expectedIndex = Number(((signedHash % 4n) + 4n) % 4n);
+ assert.strictEqual(mq.broker.endpoints.facade, '127.0.0.1:' + (10911 +
expectedIndex));
+ });
+});
