[PR] [ISSUE #137] feat: add a simple consumer mode for normal-topic consumption [rocketmq-flink]

["lizhimins (via GitHub)" <gi...@apache.org>]

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: commits-unsubscribe@rocketmq.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

Re: [PR] [ISSUE #137] feat: add a simple consumer mode for normal-topic consumption [rocketmq-flink]

["lizhimins (via GitHub)" <gi...@apache.org>]

lizhimins commented on PR #138:
URL: https://github.com/apache/rocketmq-flink/pull/138#issuecomment-5315542048

   Pushed two follow-up commits:
   
   1. `chore: apply spotless formatting` — the feature commit was written on a JDK 21 toolchain, where the parent pom's `java17` profile disables spotless, so google-java-format never validated it and the `Build (JDK 11)` check failed. Both CI jobs are green now.
   
   2. `fix: drop the empty protected_branches key from .asf.yaml` — unrelated to this feature, but ASF Infra started rejecting the file:
   
      ```
      when expecting a mapping
      found a blank string
        in "rocketmq-flink.git/.asf.yaml::github", line 13, column 1
      ```
   
      `protected_branches:` has been present without a value since the file was created in 2021, and YAML parses it as `null`. Because the parse fails, the **whole `github` section is discarded** — the description, homepage, labels and the `issues` / `wiki` feature flags never took effect. Removing the empty key makes the section parse again; filling it in instead would change the repository's actual branch-protection rules, which felt like a separate decision. Happy to split this into its own PR if you prefer.


-- 
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: commits-unsubscribe@rocketmq.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

Re: [PR] [ISSUE #137] feat: add a simple consumer mode for normal-topic consumption [rocketmq-flink]

["lizhimins (via GitHub)" <gi...@apache.org>]

lizhimins merged PR #138:
URL: https://github.com/apache/rocketmq-flink/pull/138


-- 
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: commits-unsubscribe@rocketmq.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org