apache / apache/rocketmq-flink

Support normal-topic consumption with the gRPC SimpleConsumer

Open
#137 0 comments 0 reactions 0 assignees View on GitHub
Dominant language
Java
Stars
174
Forks
104
PR merge metrics
No merged PRs in 30d

Description

### Search before asking

- [X] I had searched in the [issues](https://github.com/apache/rocketmq-flink/issues) and found no similar issues.

### Motivation

The gRPC connector added in #135 / #136 only supports the LITE consumption model: the source binds a
**main lite topic** through `LiteSimpleConsumer` and defers acknowledgement to a downstream operator,
which is what enables fair per-sub-topic throttling.

That leaves the most common case uncovered: consuming a **plain normal topic** over gRPC. Users who
just want `topic + tag/SQL92 filter` today have to either fall back to the remoting connector or
model their topic as a lite topic, and they inherit a downstream ack operator they do not need.

### Solution

Add a SIMPLE consumption mode alongside LITE, selected by a single `rocketmq.source.mode`
(SQL: `source.mode`) option:

- Subscribe one normal topic through the SDK `SimpleConsumer` with an optional tag / SQL92 filter
expression.
- Acknowledge on the source side, aligned with checkpoints — the same approach the GCP Pub/Sub
connector uses: receipt handles of emitted records are grouped per checkpoint and acked from
`notifyCheckpointComplete`, giving at-least-once. Un-acked messages are redelivered once their
invisible duration expires.
- Issue those acks through the very consumer that received the messages, so no second gRPC
connection and no downstream ack operator are involved.
- Abstract the two SDK Pop clients (`SimpleConsumer` / `LiteSimpleConsumer`) behind one internal
interface, since they expose the same receive / ack / changeInvisibleDuration operations but share
no common type.
- Expose the mode and the filter options through the SQL connector as well.

LITE remains fully supported for the lite-topic + downstream-throttling scenario.

### Are you willing to submit a PR?

- [X] I'm willing to submit a PR!

Contributor guide

No contributing guide indexed for this repository

Research direction

Start by tracing the existing gRPC LITE source from LiteSimpleConsumer through checkpoint completion and the SQL connector’s source.mode exposure. Compare the GCP Pub/Sub connector’s notifyCheckpointComplete acknowledgement flow, then verify SIMPLE receives filtered normal-topic messages, acks through the same consumer, and leaves LITE behavior intact.

Written by the indexing model from the issue text.

Assessment

Tech stack
grpc, java, sql
Domain
distributed-systems, stream-processing
Issue type
Feature
Difficulty
4/5
Estimated time
3-5 days
Activity status
Active
Clarity
Mostly clear
Newbie friendliness
48/100

Get new issues in your inbox

A short digest of beginner-friendly GitHub issues.