zhaohai666 opened a new pull request, #1379:
URL: https://github.com/apache/rocketmq-clients/pull/1379
# feat(nodejs): implement LiteSimpleConsumer
## Summary
This PR ports the **LiteSimpleConsumer** from the reference Java client to
the Node.js client
(RocketMQ 5.x gRPC protocol), branch `feat/nodejs-lite-simple-consumer`
based on `master` (`07697f75`).
It lands **1 commit** touching **11 files (+1,154 / −40)**.
The lite consumer model lets one subscription group consume many short-lived
"lite topics" that all
map onto a single parent topic. The client binds to the parent topic, then
dynamically subscribes /
unsubscribes individual lite topics via `SyncLiteSubscription`, receiving
their messages through the
parent topic route without per-topic route management.
**What is included**
- `LiteSimpleConsumer` public interface + builder with validation, mirroring
`LiteSimpleConsumerImpl` / `LiteSimpleConsumerBuilderImpl` on the Java
side.
- `LiteSimpleConsumerImpl` with `subscribeLite` (with `OffsetOption`),
`unsubscribeLite`,
`receive`, `ack`, `changeInvisibleDuration`, and full startup / shutdown
flow
(`LITE_SIMPLE_CONSUMER` client type, settings sync, telemetry, route
pruning to the first
readable master queue).
- Reworked `LiteSubscriptionManager`: host-agnostic (works for both lite
consumer variants),
multi-endpoint sync (`FULL` / `PARTIAL_ADD` / `PARTIAL_REMOVE`) with
per-endpoint failure
aggregation, quota enforcement (`maxLiteTopicSize` adopted from server
settings), and
`NOTIFY_UNSUBSCRIBE_LITE_COMMAND` handling.
- `NOTIFY_UNSUBSCRIBE_LITE_COMMAND` dispatch added to `BaseClient` /
`TelemetrySession`, with the
callback signature aligned to `(endpoints, command)` across consumers.
- A defect fix on the lite ack path discovered during real-cluster testing
(see below).
---
## Defect fix: missing `liteTopic` in ack / change-invisible-duration
requests
Real-cluster testing showed messages were received fine, but every `ack`
failed with
`50001 INTERNAL_SERVER_ERROR`. The proxy resolves LMQ receipt handles by the
`liteTopic` carried in
`AckMessageEntry` / `ChangeInvisibleDurationRequest`; the `SimpleConsumer`
base path never set it
(and `isLiteConsumer()` only recognized `LITE_PUSH_CONSUMER`), so the proxy
could not map the
handle back to the parent topic.
The fix mirrors `ConsumerImpl.wrapAckMessageRequest` in the reference
implementation:
- `Consumer#isLiteConsumer()` now covers `LITE_SIMPLE_CONSUMER` as well.
- `Consumer#ackMessage` and `Consumer#invisibleDuration` set `liteTopic` on
the request entries
when the message view carries one (same pattern the `LitePushConsumer`
path already used).
Without this, no lite consumer based on the simple-consumer path can ack
messages.
---
## Tests
**Offline unit tests** — 22/22 passing (`node --test`, no cluster required):
- Builder: option chaining, blank-topic / group / await-duration validation,
required options.
- Impl constructor: bind-topic guard, `LITE_SIMPLE_CONSUMER` client type,
default `SUB_ALL`
subscription on the bind topic, empty lite topic set before startup.
- Route pruning: keep only the first readable master queue; empty route
fallback.
- `LiteSubscriptionManager`: `PARTIAL_ADD`/`PARTIAL_REMOVE` fan-out to every
sync endpoint,
duplicate/unknown-topic short-circuit, name/length validation, quota
exceeded, settings-driven
`maxLiteTopicSize`, rejected-sync failure isolation, notify-unsubscribe
(including blank topic),
clean shutdown.
**Real-cluster integration tests** — 4/4 passing against RocketMQ 5.5.1
(namesrv + broker with `enableLmq=true` + `enableMultiDispatch=true`,
standalone proxy in CLUSTER
mode, LITE topic + consumer group pre-created with `+lite.bind.topic`):
- End-to-end: send 5 lite messages → `subscribeLite` with
`OffsetOption.MIN_OFFSET` → receive →
body-by-body verification → ack all.
- Delivery of messages sent **after** `subscribeLite` (default offset
behavior).
- No delivery after `unsubscribeLite`.
- Blank lite topic names rejected.
Known server-side race handled in the test harness: right after topic
creation, the proxy may
serve a cached topic route without the `message.type` attribute, making the
producer reject the
send client-side; the integration test refreshes the route cache and retries
within a bounded
window (same workaround used by the reference client tests).
Also adds `tsconfig.test.json` (copied from an in-flight client hardening
branch) so the test
sources compile under the same two-config layout used by recent Node.js work.
---
## Checklist
- [x] Reference implementation (`java/client/.../LiteSimpleConsumerImpl`)
reviewed and mirrored
- [x] `tsc` passes for both `tsconfig.prod.json` and `tsconfig.test.json`
- [x] ESLint clean (no new warnings beyond existing baseline)
- [x] Offline unit tests pass (22/22)
- [x] Real-cluster integration tests pass (4/4)
- [x] No changes to generated proto files required (existing bindings
already carry
`SyncLiteSubscription`, `AckMessageEntry#liteTopic`,
`MessageType.LITE`)
--
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]
