lizhimins opened a new issue, #135: URL: https://github.com/apache/rocketmq-flink/issues/135
## 中文描述 ### 背景与动机 在 AI 等场景中，常见的形态是一个**主 lite topic** 下挂载大量子 topic（每个租户/会话一个子 topic）。这类场景对 Flink 接入有两个诉求： 1. **下游 ack**：一条消息是否消费成功，只有在下游业务处理（可能跨 keyBy/shuffle）之后才知道，source 端无法就地确认； 2. **按子 topic 公平限流**：某个热点子 topic 占用过多资源时，希望推迟它的消息重投，把拉取预算让给其它子 topic，而不是全局限流。 现有 remoting 连接器基于 offset 模型，无法表达"单条消息延迟重投"；RocketMQ 5.x 的 Pop 消费模型（`invisibleDuration` + 自包含 receipt handle）恰好匹配这两个诉求。 ### 提议 新增独立模块 `flink-connector-rocketmq-grpc`（基于 `rocketmq-client-java` 5.2.1+ 的 `LiteSimpleConsumer`）： - **Source**：`bindTopic` 主 lite topic 做泛化（wildcard）订阅，broker 做消息级负载均衡，不做 split 分区；每 subtask 单个共享 consumer + `fetch-concurrency` 个并发长轮询线程；**source 永不 ack**； - **产出类型 `AckableMessage<OUT>`**：payload + 可序列化、免凭据的 `RocketMQReceiptHandle`（路由三元组 endpoint/namespace/consumerGroup + 消息标识），可跨 shuffle 传输； - **下游 ack / 限流**：每 TM 引用计数池化的同组 ack client；两种接线——继承 `RocketMQAckProcessFunction` 直接 ack，或实现 `MessageThrottlePolicy<T>` 配合 `RocketMQThrottleProcessFunction`（内置 maxDeliveryAttempt=16、maxInvisibleDuration=30min 安全阀，防止反复 defer 进 DLQ）； - **invisible-duration 续期**：背压时对尚未 emit 的队内消息按策略续期，防止提前重投； - **Sink / SQL**：gRPC Producer sink 与 `rocketmq-grpc` SQL connector（SQL 路径只消费 value，ack/限流为 DataStream-only）； - **语义**：at-least-once，提交点由用户业务显式 ack 决定。 详细设计见 PR 中的 `docs/grpc-connector.md`（动机、架构、决策与取舍、备选方案对比）。 --- ## English Description ### Motivation A common shape in AI-style workloads is one **main lite topic** carrying a large number of sub topics (one per tenant/session). Two requirements follow: 1. **Downstream acknowledgement**: whether a message is successfully consumed is only known after downstream processing, possibly across keyBy/shuffle — the source cannot ack in place. 2. **Fair per-sub-topic throttling**: when a hot sub topic consumes too many resources, its redelivery should be deferred so the fetch budget goes to other sub topics, instead of throttling globally. The offset-based remoting connector cannot express "defer this single message". The RocketMQ 5.x Pop model (`invisibleDuration` + self-contained receipt handles) matches both requirements. ### Proposal Add a separate module `flink-connector-rocketmq-grpc` built on `LiteSimpleConsumer` (`rocketmq-client-java` 5.2.1+): - **Source**: binds one main lite topic with a wildcard subscription; the broker performs message-level load balancing so there is no split partitioning. Each subtask runs one shared consumer plus `fetch-concurrency` concurrent long-poll workers. **The source never acks.** - **Output type `AckableMessage<OUT>`**: the payload plus a serializable, credential-free `RocketMQReceiptHandle` (routing triple endpoint/namespace/consumerGroup + message identity) that can cross shuffle boundaries. - **Downstream ack / throttling**: a per-TaskManager, reference-counted pool of same-group ack consumers; two wirings — extend `RocketMQAckProcessFunction` and ack directly, or implement `MessageThrottlePolicy<T>` with the one-stop `RocketMQThrottleProcessFunction` (built-in safety valves maxDeliveryAttempt=16 and maxInvisibleDuration=30min to avoid DLQ from repeated defers). - **Invisible-duration renewal**: under backpressure, queue-resident (not yet emitted) messages are renewed ahead of expiry to prevent premature redelivery. - **Sink / SQL**: a gRPC producer sink and a `rocketmq-grpc` SQL connector (the SQL path consumes values only; ack/throttling APIs are DataStream-only). - **Semantics**: at-least-once; the commit point is the user's explicit ack. See `docs/grpc-connector.md` in the PR for the full design (motivation, architecture, decisions, trade-offs and rejected alternatives). -- 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]
