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 6c96a836 feat(nodejs): add consumer-side rate limiting via semaphore
backpressure (#1386)
6c96a836 is described below
commit 6c96a836affe75f1f2f1fe9fe6856aff161e6322
Author: zhaohai <[email protected]>
AuthorDate: Fri Oct 9 10:53:14 2026 +0800
feat(nodejs): add consumer-side rate limiting via semaphore backpressure
(#1386)
Add src/util/Semaphore.ts and apply it to ProcessQueue, bounding in-flight
consumption by consumeConcurrentlyMax (default 32). The receive loop
pauses while all semaphore permits are taken, providing per-queue
consumption backpressure; the setting is parsed from broker-side settings
when present.
---
nodejs/src/consumer/ProcessQueue.ts | 34 +++++++++-
nodejs/src/consumer/PushConsumer.ts | 7 ++
nodejs/src/consumer/PushSubscriptionSettings.ts | 21 ++++++
nodejs/src/util/Semaphore.ts | 82 +++++++++++++++++++++++
nodejs/test/consumer/ProcessQueue.test.ts | 88 +++++++++++++++++++++++++
nodejs/test/util/Semaphore.test.ts | 80 ++++++++++++++++++++++
6 files changed, 311 insertions(+), 1 deletion(-)
diff --git a/nodejs/src/consumer/ProcessQueue.ts
b/nodejs/src/consumer/ProcessQueue.ts
index 6a0fde13..8563b06b 100644
--- a/nodejs/src/consumer/ProcessQueue.ts
+++ b/nodejs/src/consumer/ProcessQueue.ts
@@ -23,6 +23,7 @@ import { TooManyRequestsException } from '../exception';
import { ConsumeResult, ConsumeResultSuspend } from './ConsumeResult';
import { FilterExpression } from './FilterExpression';
import type { PushConsumer } from './PushConsumer';
+import { Semaphore } from '../util/Semaphore';
const ACK_MESSAGE_FAILURE_BACKOFF_DELAY = 1000;
const CHANGE_INVISIBLE_DURATION_FAILURE_BACKOFF_DELAY = 1000;
@@ -31,6 +32,7 @@ const FORWARD_MESSAGE_TO_DLQ_FAILURE_BACKOFF_DELAY = 1000;
const RECEIVING_FLOW_CONTROL_BACKOFF_DELAY = 20;
const RECEIVING_FAILURE_BACKOFF_DELAY = 1000;
const RECEIVING_BACKOFF_DELAY_WHEN_CACHE_IS_FULL = 1000;
+const RECEIVING_BACKOFF_DELAY_WHEN_PERMITS_EXHAUSTED = 20;
export class ProcessQueue {
readonly #consumer: PushConsumer;
@@ -42,11 +44,14 @@ export class ProcessQueue {
#activityTime = Date.now();
#cacheFullTime = 0;
#aborted = false;
+ // Bounds in-flight messages per queue; mirrors Java ProcessQueueImpl
permits.
+ readonly #semaphore: Semaphore;
constructor(consumer: PushConsumer, mq: MessageQueue, filterExpression:
FilterExpression) {
this.#consumer = consumer;
this.#mq = mq;
this.#filterExpression = filterExpression;
+ this.#semaphore = new
Semaphore(consumer.getPushConsumerSettings().getConsumeConcurrentlyMax());
}
getMessageQueue(): MessageQueue {
@@ -85,17 +90,34 @@ export class ProcessQueue {
return;
}
for (const messageView of messageList) {
+ // Each in-flight message consumes one permit. The reception batch size
is
+ // already bounded by availablePermits, so tryAcquire should always
succeed;
+ // if it does not (defensive), stop caching to avoid over-committing.
+ if (!this.#semaphore.tryAcquire()) {
+ break;
+ }
this.#cachedMessages.push(messageView);
this.#cachedMessagesBytes += messageView.body.length;
}
}
+ /**
+ * Number of consumption permits still available on this process queue. When
it
+ * reaches zero the receive loop applies backpressure (see
#receiveMessageImmediately).
+ */
+ availablePermits(): number {
+ return this.#semaphore.availablePermits;
+ }
+
#getReceptionBatchSize(): number {
const bufferSize = Math.max(
this.#consumer.cacheMessageCountThresholdPerQueue() -
this.cachedMessagesCount(),
1,
);
- return Math.min(bufferSize,
this.#consumer.getPushConsumerSettings().getReceiveBatchSize());
+ // Bound the batch by the remaining consumption permits, mirroring Java's
+ // Math.max(consumeConcurrentlyMax - inflight, 1).
+ const permitSize = Math.max(this.#semaphore.availablePermits, 1);
+ return Math.min(bufferSize, permitSize,
this.#consumer.getPushConsumerSettings().getReceiveBatchSize());
}
fetchMessageImmediately(): void {
@@ -140,6 +162,13 @@ export class ProcessQueue {
#receiveMessageImmediately(attemptId?: string): void {
if (this.#aborted) return;
+ // Apply backpressure: when every consumption permit is in use (i.e. the
+ // maximum number of messages is in-flight), stop fetching until messages
are
+ // settled and permits are released. Mirrors Java's tryAcquirePermit gate.
+ if (this.#semaphore.availablePermits <= 0) {
+
this.#receiveMessageLater(RECEIVING_BACKOFF_DELAY_WHEN_PERMITS_EXHAUSTED,
attemptId);
+ return;
+ }
attemptId = attemptId ?? this.#generateAttemptId();
try {
const batchSize = this.#getReceptionBatchSize();
@@ -376,6 +405,9 @@ export class ProcessQueue {
if (index !== -1) {
this.#cachedMessages.splice(index, 1);
this.#cachedMessagesBytes -= messageView.body.length;
+ // Return the consumption permit held by this now-settled message, so the
+ // receive loop can fetch again (backpressure release).
+ this.#semaphore.release();
}
}
diff --git a/nodejs/src/consumer/PushConsumer.ts
b/nodejs/src/consumer/PushConsumer.ts
index 301d0bc1..31effdf9 100644
--- a/nodejs/src/consumer/PushConsumer.ts
+++ b/nodejs/src/consumer/PushConsumer.ts
@@ -54,6 +54,12 @@ export interface PushConsumerOptions extends ConsumerOptions
{
maxCacheMessageSizeInBytes?: number;
longPollingTimeout?: number;
enableFifoConsumeAccelerator?: boolean;
+ /**
+ * Max number of messages in-flight per process queue. Mirrors Java's
+ * consumeConcurrentlyMax (default 32) and backs the ProcessQueue semaphore
+ * that applies consumption backpressure.
+ */
+ consumeConcurrentlyMax?: number;
}
class ConsumeMetrics {
@@ -108,6 +114,7 @@ export class PushConsumer extends Consumer {
this.#pushSubscriptionSettings = new PushSubscriptionSettings(
options.namespace, this.clientId, this.getClientType(), this.endpoints,
this.consumerGroup, this.requestTimeout, this.#subscriptionExpressions,
+ options.longPollingTimeout, options.consumeConcurrentlyMax,
);
}
diff --git a/nodejs/src/consumer/PushSubscriptionSettings.ts
b/nodejs/src/consumer/PushSubscriptionSettings.ts
index 2f3b1a8c..a69dfbe1 100644
--- a/nodejs/src/consumer/PushSubscriptionSettings.ts
+++ b/nodejs/src/consumer/PushSubscriptionSettings.ts
@@ -34,6 +34,7 @@ export class PushSubscriptionSettings extends Settings {
#fifo = false;
#receiveBatchSize = 32;
#longPollingTimeout = 30000; // ms
+ #consumeConcurrentlyMax = 32; // mirrors Java PushConsumerSettings default
constructor(
namespace: string,
@@ -44,6 +45,7 @@ export class PushSubscriptionSettings extends Settings {
requestTimeout: number,
subscriptionExpressions: Map<string, FilterExpression>,
longPollingTimeout?: number,
+ consumeConcurrentlyMax?: number,
) {
super(namespace, clientId, clientType, accessPoint, requestTimeout);
this.#group = consumerGroup;
@@ -51,6 +53,9 @@ export class PushSubscriptionSettings extends Settings {
if (longPollingTimeout !== undefined) {
this.#longPollingTimeout = longPollingTimeout;
}
+ if (consumeConcurrentlyMax !== undefined) {
+ this.#consumeConcurrentlyMax = consumeConcurrentlyMax;
+ }
}
isFifo(): boolean {
@@ -61,6 +66,16 @@ export class PushSubscriptionSettings extends Settings {
return this.#receiveBatchSize;
}
+ /**
+ * Maximum number of messages that may be in-flight (fetched but not yet
+ * settled) per process queue. The Node client previously had no equivalent
of
+ * the Java {@code ProcessQueueImpl} permit, so consumption could grow
+ * unbounded; this bounds it and feeds the {@link ProcessQueue} semaphore.
+ */
+ getConsumeConcurrentlyMax(): number {
+ return this.#consumeConcurrentlyMax;
+ }
+
getLongPollingTimeout(): number {
return this.#longPollingTimeout;
}
@@ -100,6 +115,12 @@ export class PushSubscriptionSettings extends Settings {
this.#longPollingTimeout = longPollingTimeout.getSeconds() * 1000 +
Math.floor(longPollingTimeout.getNanos() / 1000000);
}
+ // consumeConcurrentlyMax is absent from the generated proto in this
fork;
+ // guard the accessor so we keep the default (32) until the field lands.
+ const consumeConcurrentlyMax = (subscription as {
getConsumeConcurrentlyMax?: () => number | null
}).getConsumeConcurrentlyMax?.();
+ if (typeof consumeConcurrentlyMax === 'number' && consumeConcurrentlyMax
> 0) {
+ this.#consumeConcurrentlyMax = consumeConcurrentlyMax;
+ }
}
const backoffPolicy = settings.getBackoffPolicy();
if (backoffPolicy) {
diff --git a/nodejs/src/util/Semaphore.ts b/nodejs/src/util/Semaphore.ts
new file mode 100644
index 00000000..4ab2ee69
--- /dev/null
+++ b/nodejs/src/util/Semaphore.ts
@@ -0,0 +1,82 @@
+/**
+ * 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.
+ */
+
+/**
+ * A counting semaphore used to throttle the number of messages that can be
+ * in-flight (fetched but not yet settled) on a single {@link ProcessQueue}.
+ *
+ * This mirrors the permit model of the Java client's
+ * {@code ProcessQueueImpl#acquirePermit()}/{@code releasePermit()}, where the
+ * permit count is bounded by {@code consumeConcurrentlyMax} and exhaustion
+ * applies backpressure by stopping further message fetches until permits are
+ * released (i.e. messages are acked / nacked).
+ */
+export class Semaphore {
+ #permits: number;
+ readonly #waiters: Array<() => void> = [];
+
+ constructor(permits: number) {
+ if (!Number.isInteger(permits) || permits < 0) {
+ throw new RangeError(`permits must be a non-negative integer, got
${permits}`);
+ }
+ this.#permits = permits;
+ }
+
+ /** Current number of available permits. */
+ get availablePermits(): number {
+ return this.#permits;
+ }
+
+ /**
+ * Non-blocking acquisition: takes one permit if available and returns true,
+ * otherwise leaves the semaphore unchanged and returns false.
+ */
+ tryAcquire(): boolean {
+ if (this.#permits > 0) {
+ this.#permits--;
+ return true;
+ }
+ return false;
+ }
+
+ /**
+ * Acquire a single permit. Resolves immediately when one is available;
+ * otherwise queues the caller until a permit is released.
+ */
+ acquire(): Promise<void> {
+ if (this.#permits > 0) {
+ this.#permits--;
+ return Promise.resolve();
+ }
+ return new Promise<void>(resolve => {
+ this.#waiters.push(resolve);
+ });
+ }
+
+ /**
+ * Release a single permit. Hands it to a pending acquirer if any, otherwise
+ * increments the available count.
+ */
+ release(): void {
+ const waiter = this.#waiters.shift();
+ if (waiter) {
+ waiter();
+ } else {
+ this.#permits++;
+ }
+ }
+}
diff --git a/nodejs/test/consumer/ProcessQueue.test.ts
b/nodejs/test/consumer/ProcessQueue.test.ts
new file mode 100644
index 00000000..dbd194f1
--- /dev/null
+++ b/nodejs/test/consumer/ProcessQueue.test.ts
@@ -0,0 +1,88 @@
+/**
+ * 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 { ProcessQueue } from '../../src/consumer/ProcessQueue';
+import { FilterExpression } from '../../src/consumer/FilterExpression';
+import { Code } from '../../proto/apache/rocketmq/v2/definition_pb';
+
+function makeConsumer(consumeConcurrentlyMax: number): any {
+ return {
+ getPushConsumerSettings: () => ({ getConsumeConcurrentlyMax: () =>
consumeConcurrentlyMax }),
+ getRetryPolicy: () => ({ getNextAttemptDelay: () => 0 }),
+ requestTimeoutValue: 3000,
+ changeInvisibleDurationViaRpc: () => Promise.resolve({
+ getStatus: () => ({ toObject: () => ({ code: Code.OK }) }),
+ }),
+ wrapChangeInvisibleDurationRequest: () => ({}),
+ getConsumerGroup: () => 'cg',
+ clientId: 'client',
+ logger: { info: () => {}, debug: () => {}, warn: () => {}, error: () => {}
},
+ getConsumeService: () => ({ consume: () => {} }),
+ cacheMessageCountThresholdPerQueue: () => 1024,
+ cacheMessageBytesThresholdPerQueue: () => 1024 * 1024,
+ };
+}
+
+function makeMessage(id: string): any {
+ return {
+ messageId: id,
+ body: '',
+ endpoints: {},
+ deliveryAttempt: 1,
+ topic: { name: 't' },
+ liteTopic: '',
+ };
+}
+
+const mq: any = { topic: { name: 't' }, broker: { name: 'b' }, queueId: 0 };
+
+describe('ProcessQueue consumption permit (backpressure)', () => {
+ it('caps cached messages by consumeConcurrentlyMax and reports
availablePermits', () => {
+ const pq = new ProcessQueue(makeConsumer(2), mq, new
FilterExpression('*'));
+ pq.cacheMessages([makeMessage('a'), makeMessage('b'), makeMessage('c')]);
+ // Only two permits exist, so only two messages are cached.
+ assert.strictEqual(pq.cachedMessagesCount(), 2);
+ assert.strictEqual(pq.availablePermits(), 0);
+ });
+
+ it('releases a permit when a message is settled (discarded/nacked)', async
() => {
+ const pq = new ProcessQueue(makeConsumer(2), mq, new
FilterExpression('*'));
+ const a = makeMessage('a');
+ pq.cacheMessages([a, makeMessage('b')]);
+ assert.strictEqual(pq.availablePermits(), 0);
+ // discardMessage -> nack -> (async) evict -> release permit
+ pq.discardMessage(a);
+ await new Promise(r => setTimeout(r, 30));
+ assert.strictEqual(pq.availablePermits(), 1);
+ });
+
+ it('releases nothing for an unknown message (no double release)', async ()
=> {
+ const pq = new ProcessQueue(makeConsumer(2), mq, new
FilterExpression('*'));
+ const a = makeMessage('a');
+ pq.cacheMessages([a]);
+ assert.strictEqual(pq.availablePermits(), 1);
+ pq.discardMessage(makeMessage('z')); // not in cache
+ await new Promise(r => setTimeout(r, 30));
+ // Only the originally cached message holds a permit; the unknown one is a
no-op.
+ assert.strictEqual(pq.availablePermits(), 1);
+ pq.discardMessage(a);
+ await new Promise(r => setTimeout(r, 30));
+ assert.strictEqual(pq.availablePermits(), 2);
+ });
+});
diff --git a/nodejs/test/util/Semaphore.test.ts
b/nodejs/test/util/Semaphore.test.ts
new file mode 100644
index 00000000..4fcff641
--- /dev/null
+++ b/nodejs/test/util/Semaphore.test.ts
@@ -0,0 +1,80 @@
+/**
+ * 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 { Semaphore } from '../../src/util/Semaphore';
+
+describe('Semaphore (consumption permit primitive)', () => {
+ it('starts with the configured number of permits', () => {
+ const s = new Semaphore(3);
+ assert.strictEqual(s.availablePermits, 3);
+ });
+
+ it('tryAcquire consumes a permit and returns false when exhausted', () => {
+ const s = new Semaphore(1);
+ assert.strictEqual(s.tryAcquire(), true);
+ assert.strictEqual(s.availablePermits, 0);
+ assert.strictEqual(s.tryAcquire(), false);
+ assert.strictEqual(s.availablePermits, 0);
+ });
+
+ it('rejects non-negative integer construction', () => {
+ assert.throws(() => new Semaphore(-1), RangeError);
+ assert.throws(() => new Semaphore(1.5), RangeError);
+ });
+
+ it('acquire resolves immediately when a permit is available', async () => {
+ const s = new Semaphore(1);
+ await s.acquire();
+ assert.strictEqual(s.availablePermits, 0);
+ });
+
+ it('acquire blocks until a permit is released, then resumes', async () => {
+ const s = new Semaphore(0);
+ let resumed = false;
+ const p = s.acquire().then(() => { resumed = true; });
+ // Permit not yet available.
+ await new Promise(r => setTimeout(r, 10));
+ assert.strictEqual(resumed, false);
+ s.release();
+ await p;
+ assert.strictEqual(resumed, true);
+ });
+
+ it('serves multiple waiters in FIFO order on successive releases', async ()
=> {
+ const s = new Semaphore(0);
+ const order: number[] = [];
+ const a = s.acquire().then(() => order.push(1));
+ const b = s.acquire().then(() => order.push(2));
+ const c = s.acquire().then(() => order.push(3));
+ s.release();
+ s.release();
+ s.release();
+ await Promise.all([a, b, c]);
+ assert.deepStrictEqual(order, [1, 2, 3]);
+ assert.strictEqual(s.availablePermits, 0);
+ });
+
+ it('release hands the permit to a waiter without incrementing the pool', ()
=> {
+ const s = new Semaphore(0);
+ const p = s.acquire();
+ s.release(); // should go straight to the waiter, not the pool
+ assert.strictEqual(s.availablePermits, 0);
+ return p;
+ });
+});
