lizhimins opened a new pull request, #138:
URL: https://github.com/apache/rocketmq-flink/pull/138
Closes #137
## What is the purpose of the change
The gRPC connector added in #136 only supports the LITE model: bind a **main
lite topic** through
`LiteSimpleConsumer` and defer acknowledgement to a downstream operator.
Consuming a plain
**normal topic** over gRPC — `topic` plus an optional tag / SQL92 filter —
was not possible.
This PR adds a SIMPLE consumption mode next to LITE, selected by one option.
## Brief change log
- **`rocketmq.source.mode`** (SQL: `source.mode`) selects the mode.
`ConsumerMode.SIMPLE` is the
default; LITE is opt-in.
- **SIMPLE mode** subscribes one normal topic through the SDK
`SimpleConsumer` with an optional
`source.filter-expression` / `source.filter-type` (TAG or SQL92).
- **Checkpoint-aligned ack, Pub/Sub style.** The receipt handles of emitted
records are grouped per
checkpoint by `CheckpointAckTracker`, snapshotted in `snapshotState` and
acked from
`notifyCheckpointComplete`. Un-acked messages are redelivered once their
invisible duration
expires — that redelivery is the at-least-once guarantee, so the tracker
keeps no
failure-survivable state. Ack failures are logged and tolerated, so one
stale handle can never
fail the job.
- **Acks go through the receiving consumer**, not through a second client:
the split reader's
consumer already targets the same group and topic and is thread-safe, so
SIMPLE mode opens no
extra gRPC connection and resolves no credentials of its own.
- **`PopConsumer`** abstracts the two SDK Pop clients (`SimpleConsumer` /
`LiteSimpleConsumer`),
which expose the same `receive` / `ack` / `changeInvisibleDuration`
operations but share no common
interface, keeping `RocketMQGrpcSourceSplitReader` mode-agnostic.
- **`RocketMQLiteAckClient` → `RocketMQAckClient`**; it now serves only the
LITE downstream-ack path,
where credentials and per-TaskManager pooling are genuinely needed.
- **Docs**: `docs/grpc-connector.md` covers both modes, the checkpoint-ack
semantics, and how to size
`invisible-duration` with headroom (a failed checkpoint defers the ack to
the next successful one).
LITE behaviour is unchanged: no ack client is acquired by the reader and no
ack is issued by it; the
downstream ack / throttle operators are untouched.
### Breaking change
The source now consumes a normal topic with a checkpoint-aligned
`SimpleConsumer` **by default**.
Jobs relying on the main lite topic with downstream acknowledgement must
select LITE explicitly:
```java
builder.setMode(ConsumerMode.LITE).setMainTopic("LiteMainTopic")
```
```sql
'source.mode' = 'lite'
```
## Verifying this change
- Unit tests: 67 tests green in `flink-connector-rocketmq-grpc`, including
new
`CheckpointAckTrackerTest`, `PopConsumerProviderTest`,
`RocketMQGrpcSourceSplitReaderTest`, plus
builder / SQL-factory coverage for the new options.
- `RocketMQGrpcITCase#testSimpleModeConsumesNormalTopic` verifies the full
ack loop: 50 messages are
received, the job is kept alive until a checkpoint completes, and after
the invisible duration
expires a fresh `SimpleConsumer` receives **zero** redelivered messages.
- The ITCase class now uses
`@EnabledIfEnvironmentVariable("ROCKETMQ_GRPC_ENDPOINTS")` instead of
`@Disabled`, so it runs automatically once the `ROCKETMQ_GRPC_*` variables
are exported and is
skipped otherwise.
- Executed against a self-hosted open-source RocketMQ 5.x cluster
(nameserver + broker + proxy) and
against an Alibaba Cloud RocketMQ 5.x instance over its public gRPC
endpoint; both passed,
including the zero-redelivery assertion.
## Does this pull request potentially affect one of the following parts
- Dependencies (does it add or upgrade a dependency): **no**
- The public API: **yes** — new `ConsumerMode` / source options / builder
methods, the
`RocketMQAckClient` rename, and the default-mode change described above.
- The serializers: **no**
- The runtime per-record code paths: **yes** — SIMPLE mode registers a
receipt handle per emitted
record; LITE is unchanged.
- Anything that affects deployment: **no**
## Documentation
- Does this pull request introduce a new feature: **yes**
- If yes, how is the feature documented: `docs/grpc-connector.md`
(consumption model, delivery
semantics, source examples, options table, SQL DDL, known limitations).
--
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]
