zhaohai666 opened a new pull request, #1380:
URL: https://github.com/apache/rocketmq-clients/pull/1380
# feat(csharp): implement LiteSimpleConsumer
## Summary
This PR ports the **LiteSimpleConsumer** from the reference Java client to
the .NET client
(RocketMQ 5.x gRPC protocol), branch `feat/csharp-lite-simple-consumer`
based on `master`
(`fc088d4a`). It lands **1 commit** touching **14 files (+1,124 / −37)**.
The lite consumer model lets one consumer group consume many short-lived
"lite topics" that all map
onto a single parent topic. The consumer binds to the parent topic, then
dynamically subscribes /
unsubscribes individual lite topics through the `SyncLiteSubscription` RPC,
and pulls their messages
explicitly with `Receive` + `Ack`.
**What is included**
- `LiteSimpleConsumer` + `ILiteSimpleConsumer`: `SubscribeLite(liteTopic)`,
`SubscribeLite(liteTopic, OffsetOption)`, `UnsubscribeLite(liteTopic)`,
`GetLiteTopicSet()`,
`GetBindTopic()`, plus the inherited `SimpleConsumer` receive / ack /
change-invisible-duration
APIs, and a `Builder` with the same validation rules as the Java builder.
- `LiteSimpleSubscriptionSettings`: simple-consumer settings that report
`ClientType.LITE_SIMPLE_CONSUMER`, so the server derives `LITE_SELECTIVE`
consumption mode.
- `ClientType`: new `LiteSimpleConsumer` member mapped to the protobuf enum.
- `LiteSubscriptionManager`: generalized from `PushConsumer` to the
`Consumer` base class so both
lite consumer kinds share it, and given `Shutdown()` to dispose the
periodic sync timer.
- `Consumer`: shared `CheckRunning()` / `GetRequestTimeout()` / `Namespace` /
`SyncLiteSubscription(request, timeout)`; `PushConsumer` now overrides
them instead of duplicating
the code. `IsLiteConsumer()` also recognizes
`LiteSimpleSubscriptionSettings`.
- `MessageView`: exposes `LiteTopic` parsed from the message system
properties.
- `SubscriptionLoadBalancer`: throws `NotFoundException` instead of dividing
by zero when a route
carries no queue.
- A defect fix on the lite ack path (see below).
---
## Defect fix: missing `liteTopic` in ack / change-invisible-duration
requests
Real-cluster testing showed that messages were received correctly but every
`Ack` failed with
`50001 INTERNAL_SERVER_ERROR`. The proxy resolves LMQ receipt handles
through the `liteTopic`
carried in `AckMessageEntry` / `ChangeInvisibleDurationRequest`; the plain
`SimpleConsumer` path
never set it, so the proxy could not map the receipt handle back to the
parent topic.
The fix mirrors `ConsumerImpl.wrapAckMessageRequest` on the Java side:
`LiteSimpleConsumer` overrides `WrapAckMessageRequest` and
`WrapChangeInvisibleDuration` and sets
`LiteTopic` whenever `MessageView.LiteTopic` is present. Without it, no lite
consumer built on the
simple-consumer path can acknowledge messages.
---
## Tests
**Offline unit tests** — 20/20 passing (`dotnet test --filter
FullyQualifiedName~LiteSimpleConsumerTest`):
- Builder / constructor: missing client config, invalid or blank consumer
group, blank bind topic,
non-positive await duration, missing required options.
- Protocol: settings and heartbeat carry `LITE_SIMPLE_CONSUMER`;
`SubscribeLite` issues a
`PARTIAL_ADD` sync; duplicate subscribe short-circuits without an RPC;
`UnsubscribeLite` issues a
`PARTIAL_REMOVE`; unknown topic unsubscribe is a no-op; blank lite topic
rejected;
`NOTIFY_UNSUBSCRIBE_LITE` command removes the topic locally.
- Ack path: `Ack` and `ChangeInvisibleDuration` requests carry `liteTopic`,
and omit it for a
non-lite message.
- Routing: only the first readable master queue is kept; an empty route
raises `NotFoundException`.
- Life cycle: `SubscribeLite` / `UnsubscribeLite` before startup fail fast.
**Real-cluster integration tests** — 3/3 passing against a local RocketMQ
5.5.1 deployment
(namesrv + broker with `enableLmq=true` and `enableMultiDispatch=true`,
standalone **proxy in
CLUSTER mode**, parent topic created with `message.type=LITE`, consumer
group pre-created with
`+lite.bind.topic=<parentTopic>`):
- `TestReceiveAndAckLiteMessages`: send 5 lite messages → `SubscribeLite`
with
`OffsetOption.MinOffset` → receive → assert `liteTopic` on every message →
ack all.
- `TestDeliverMessagesSentAfterSubscribeLite`: messages sent after the
subscription are delivered.
- `TestStopDeliveringAfterUnsubscribeLite`: no delivery after
`UnsubscribeLite`.
The integration suite is guarded by the `ROCKETMQ_CSHARP_LITE_ENDPOINTS`
environment variable and
reports `Assert.Inconclusive` when it is not set, so CI without a cluster
stays green.
The broker log confirms the consumer registers as `CONSUME_ACTIVELY
LITE_SELECTIVE`.
---
## Example
`csharp/examples/LiteSimpleConsumerExample.cs` (wired into
`examples/QuickStart.cs`): produces 5
lite messages with `Message.Builder.SetLiteTopic`, subscribes with
`OffsetOption.MinOffset`,
receives and acks them, then unsubscribes to release the quota.
---
## Checklist
- [x] Reference implementation (`java/client/.../LiteSimpleConsumerImpl`)
reviewed and mirrored
- [x] `dotnet build` clean, no new compiler warnings
- [x] Offline unit tests pass (20/20)
- [x] Real-cluster integration tests pass (3/3)
- [x] No changes to generated protobuf bindings required
(`SyncLiteSubscription`,
`AckMessageEntry#lite_topic`, `ClientType.LITE_SIMPLE_CONSUMER`
already present)
- [x] Example added and executed against a local cluster
--
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]
