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 666dfbce fix(nodejs): floor seconds when constructing protobuf
Duration (#1352)
666dfbce is described below
commit 666dfbceb39a95bf2d6b6f612bb944bf5b499f2e
Author: Quan <[email protected]>
AuthorDate: Fri Aug 28 15:49:51 2026 +0800
fix(nodejs): floor seconds when constructing protobuf Duration (#1352)
- createDuration(): Math.floor(ms / 1000) for the int64 seconds field
- log the swallowed exception in ProcessQueue ack/changeInvisibleDuration
retries
- ConsumeResultSuspend.of() rejects non-integer milliseconds
- add unit tests for createDuration and ConsumeResultSuspend
---
nodejs/src/consumer/ConsumeResult.ts | 7 +++++--
nodejs/src/consumer/ProcessQueue.ts | 12 ++++++++++--
nodejs/src/util/index.ts | 7 +++++--
nodejs/test/consumer/ConsumeResult.test.ts | 14 ++++++++++++++
nodejs/test/util/index.test.ts | 18 ++++++++++++++++++
5 files changed, 52 insertions(+), 6 deletions(-)
diff --git a/nodejs/src/consumer/ConsumeResult.ts
b/nodejs/src/consumer/ConsumeResult.ts
index 9dc70a9e..0c0db07b 100644
--- a/nodejs/src/consumer/ConsumeResult.ts
+++ b/nodejs/src/consumer/ConsumeResult.ts
@@ -57,6 +57,9 @@ export class ConsumeResultSuspend extends ConsumeResult {
private constructor(suspendTimeMs: number) {
super('SUSPEND');
+ if (!Number.isInteger(suspendTimeMs)) {
+ throw new Error(`suspend time must be an integer number of milliseconds,
got ${suspendTimeMs}`);
+ }
if (suspendTimeMs < MIN_SUSPEND_TIME_MS) {
throw new Error(`suspend time cannot be less than
${MIN_SUSPEND_TIME_MS}ms, got ${suspendTimeMs}ms`);
}
@@ -66,9 +69,9 @@ export class ConsumeResultSuspend extends ConsumeResult {
/**
* Create a suspend result with the given suspend time in milliseconds.
*
- * @param suspendTimeMs - Suspend time in milliseconds
+ * @param suspendTimeMs - Suspend time in milliseconds, does not need to be
a whole second
* @return {ConsumeResultSuspend} ConsumeResultSuspend instance
- * @throws {Error} if suspendTimeMs is less than 50ms
+ * @throws {Error} if suspendTimeMs is not an integer or is less than 50ms
*/
static of(suspendTimeMs: number): ConsumeResultSuspend {
return new ConsumeResultSuspend(suspendTimeMs);
diff --git a/nodejs/src/consumer/ProcessQueue.ts
b/nodejs/src/consumer/ProcessQueue.ts
index 5fc83387..e4e2c7fd 100644
--- a/nodejs/src/consumer/ProcessQueue.ts
+++ b/nodejs/src/consumer/ProcessQueue.ts
@@ -218,7 +218,11 @@ export class ProcessQueue {
if (status?.code !== Code.OK) {
await this.#ackMessageLater(messageView, attempt + 1);
}
- } catch {
+ } catch (err) {
+ (this.#consumer as any).logger?.warn(
+ 'Failed to ack message, attempt=%d, messageId=%s, error=%s',
+ attempt, messageView.messageId, err,
+ );
await this.#ackMessageLater(messageView, attempt + 1);
}
}
@@ -251,7 +255,11 @@ export class ProcessQueue {
if (status?.code !== Code.OK) {
await this.#changeInvisibleDurationLater(messageView, duration,
attempt + 1);
}
- } catch {
+ } catch (err) {
+ (this.#consumer as any).logger?.warn(
+ 'Failed to change invisible duration, attempt=%d, duration=%dms,
messageId=%s, error=%s',
+ attempt, duration, messageView.messageId, err,
+ );
await this.#changeInvisibleDurationLater(messageView, duration, attempt
+ 1);
}
}
diff --git a/nodejs/src/util/index.ts b/nodejs/src/util/index.ts
index eb2af3a8..df68c20f 100644
--- a/nodejs/src/util/index.ts
+++ b/nodejs/src/util/index.ts
@@ -45,9 +45,12 @@ export function sign(accessSecret: string, dateTime: string)
{
}
export function createDuration(ms: number) {
- const nanos = ms % 1000 * 1000000;
+ // Duration.seconds is a protobuf int64 field, so it must be an integer;
+ // the remainder is carried by nanos. Non-integer seconds would fail
+ // serialization with 'Assertion failed'.
+ const nanos = Math.floor(ms % 1000 * 1000000);
return new Duration()
- .setSeconds(ms / 1000)
+ .setSeconds(Math.floor(ms / 1000))
.setNanos(nanos);
}
diff --git a/nodejs/test/consumer/ConsumeResult.test.ts
b/nodejs/test/consumer/ConsumeResult.test.ts
index 25db9ed5..dab83d07 100644
--- a/nodejs/test/consumer/ConsumeResult.test.ts
+++ b/nodejs/test/consumer/ConsumeResult.test.ts
@@ -35,9 +35,23 @@ describe('ConsumeResult', () => {
assert.strictEqual(suspend.toString(), 'SUSPEND(100ms)');
});
+ it('should accept non-whole-second suspend time', () => {
+ const suspend = ConsumeResultSuspend.of(2173);
+ assert.strictEqual(suspend.suspendTimeMs, 2173);
+ });
+
it('should reject suspend time less than 50ms', () => {
assert.throws(() => {
ConsumeResultSuspend.of(49);
}, /suspend time cannot be less than 50ms/);
});
+
+ it('should reject non-integer suspend time', () => {
+ assert.throws(() => {
+ ConsumeResultSuspend.of(100.5);
+ }, /suspend time must be an integer number of milliseconds/);
+ assert.throws(() => {
+ ConsumeResultSuspend.of(NaN);
+ }, /suspend time must be an integer number of milliseconds/);
+ });
});
diff --git a/nodejs/test/util/index.test.ts b/nodejs/test/util/index.test.ts
index 1ee50348..854969c3 100644
--- a/nodejs/test/util/index.test.ts
+++ b/nodejs/test/util/index.test.ts
@@ -17,6 +17,7 @@
import { strict as assert } from 'node:assert';
import {
+ createDuration,
getTimestamp,
calculateStringSipHash24,
} from '../../src/util';
@@ -30,6 +31,23 @@ describe('test/util/index.test.ts', () => {
});
});
+ describe('createDuration()', () => {
+ it('should split whole seconds correctly', () => {
+ const duration = createDuration(2000);
+ assert.equal(duration.getSeconds(), 2);
+ assert.equal(duration.getNanos(), 0);
+ assert(duration.serializeBinary().length > 0);
+ });
+
+ it('should split non-whole-second milliseconds correctly', () => {
+ const duration = createDuration(2173);
+ assert.equal(duration.getSeconds(), 2);
+ assert.equal(duration.getNanos(), 173000000);
+ // protobuf Duration.seconds is int64, serialization must not throw
+ assert(duration.serializeBinary().length > 0);
+ });
+ });
+
describe('calculateStringSipHash24()', () => {
it('should work', () => {
assert.equal(calculateStringSipHash24('foo哈哈😄2222哈哈'),
11716758754047899126n);
