RockteMQ-AI commented on code in PR #1353:
URL: https://github.com/apache/rocketmq-clients/pull/1353#discussion_r3879584324
##########
nodejs/src/consumer/PushConsumer.ts:
##########
@@ -152,6 +164,30 @@ export class PushConsumer extends Consumer {
this.logger.info('Push consumer has been shutdown successfully,
clientId=%s', this.clientId);
}
+ async #waitForCachedMessagesConsumed(): Promise<void> {
Review Comment:
**[Warning]** `#waitForCachedMessagesConsumed()` has no upper-bound timeout.
If a consumption handler never resolves (e.g. user code deadlocks or a network
call hangs indefinitely), `shutdown()` will block forever.
Consider adding an overall timeout parameter or a max-wait similar to
`#waitForInflightReceiveRequestsFinished()`:
```typescript
async #waitForCachedMessagesConsumed(timeoutMs?: number): Promise<void> {
const deadline = timeoutMs ? Date.now() + timeoutMs : Infinity;
while ([ ...this.#processQueueTable.values() ].some(({ pq }) =>
pq.cachedMessagesCount() > 0)) {
if (Date.now() > deadline) {
this.logger.warn('Timed out waiting for cached messages to be
consumed, clientId=%s', this.clientId);
break;
}
await this.sleep(100);
}
}
```
This ensures `shutdown()` always completes in bounded time, matching the
Java client's drain behavior.
##########
nodejs/src/consumer/PushConsumer.ts:
##########
@@ -135,15 +136,26 @@ export class PushConsumer extends Consumer {
clearInterval(this.#scanAssignmentTimer);
this.#scanAssignmentTimer = undefined;
}
- // Drop all process queues
+ // Drop all process queues to stop scheduling new receive requests
this.logger.info('Dropping all process queues, clientId=%s, queueCount=%d',
this.clientId, this.#processQueueTable.size);
for (const { pq } of this.#processQueueTable.values()) {
pq.drop();
+ }
+ // Wait for the inflight receive requests to be finished so that fetched
messages can still be consumed
+ this.logger.info('Waiting for the inflight receive requests to be
finished, clientId=%s', this.clientId);
+ await this.#waitForInflightReceiveRequestsFinished();
+ // Wait until every cached message has been consumed and settled (acked,
+ // redelivered or discarded), i.e. all consumption chains have quiesced
+ this.logger.info('Waiting for the cached messages to be consumed,
clientId=%s', this.clientId);
+ await this.#waitForCachedMessagesConsumed();
+ // Reserve a short buffer for the ack/changeInvisibleDuration callbacks to
be sent
+ await this.sleep(1000);
Review Comment:
**[Info]** The 1000ms buffer for ack/changeInvisibleDuration callbacks is a
reasonable default. If this ever needs tuning, consider making it configurable
via `PushConsumerOptions` or deriving it from `requestTimeout`. Not blocking —
just a future consideration.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
