lizhimins opened a new pull request, #136: URL: https://github.com/apache/rocketmq-flink/pull/136
Closes #135 ## 中文描述 ### 内容 新增独立模块 `flink-connector-rocketmq-grpc`（及 SQL fat-jar `flink-sql-connector-rocketmq-grpc`），基于 `rocketmq-client-java` 5.2.1+ 的 `LiteSimpleConsumer`（Pop 模型）： **Source** - `bindTopic` 主 lite topic 泛化订阅，broker 做消息级负载均衡，不做 split 分区； - 每 subtask 单个共享 consumer + `fetch-concurrency` 个并发 `receive()` 长轮询线程，接收失败指数退避； - source 永不 ack，产出 `AckableMessage<OUT>`（payload + 可序列化、免凭据的 `RocketMQReceiptHandle`），配套专用 TypeInformation/TypeSerializer（不走 Kryo）。 **下游 ack / 限流（DataStream-only）** - `RocketMQLiteAckClient`：每 TM 引用计数池化，按 handle 的路由三元组 (endpoint, namespace, consumerGroup) 维护同组 consumer 池；ack RPC 有界指数退避重试，`INVALID_RECEIPT_HANDLE` 容忍为 warn（消息重投，at-least-once）； - 两种接线：继承 `RocketMQAckProcessFunction` 直接 `ack()`/`changeInvisibleDuration()`，或 `MessageThrottlePolicy<T>` + `RocketMQThrottleProcessFunction<T>`（内置 maxDeliveryAttempt=16、maxInvisibleDuration=30min 安全阀）； - 凭据不入数据流：静态 AK/SK 或可插拔 `CredentialsResolver`（TM 本地按 endpoint 解析）； - SDK internal 依赖收敛到唯一的 `@Internal` 适配类 `RocketMQReceiptHandleCodec`。 **其它** - invisible-duration 续期策略（背压时对未 emit 的队内消息续期，emit 后冻结 handle）； - gRPC producer sink 与 `rocketmq-grpc` SQL connector； - 设计文档 `docs/grpc-connector.md`：动机、架构、语义、决策与被否决的备选方案。 ### 验证 - 模块单测 50 个全部通过（serializer round-trip、引用计数、codec、throttle policy 分支等）； - 真实 RocketMQ 5.x 集群 E2E：限流 defer（5 条消息 ~10s 后重投并 ack）与背压续期（renewals=102、duplicates=0/60）均 PASS； - `mvn spotless:check clean package`（JDK 11/17）全部 BUILD SUCCESS。 --- ## English Description ### What Adds a separate module `flink-connector-rocketmq-grpc` (plus the SQL fat-jar `flink-sql-connector-rocketmq-grpc`) built on `rocketmq-client-java` 5.2.1+ `LiteSimpleConsumer` (Pop model): **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 with `fetch-concurrency` concurrent `receive()` long-poll workers; receive failures back off exponentially. - The source never acks. It emits `AckableMessage<OUT>` — the payload plus a serializable, credential-free `RocketMQReceiptHandle` — with dedicated TypeInformation/TypeSerializer (no Kryo fallback). **Downstream ack / throttling (DataStream-only)** - `RocketMQLiteAckClient`: per-TaskManager reference-counted pool keyed by the handle routing triple (endpoint, namespace, consumerGroup); ack RPCs retry with bounded exponential backoff, and `INVALID_RECEIPT_HANDLE` is tolerated as a warning (redelivery, at-least-once). - Two wirings: extend `RocketMQAckProcessFunction` and call `ack()`/`changeInvisibleDuration()` directly, or implement `MessageThrottlePolicy<T>` with `RocketMQThrottleProcessFunction<T>` (built-in safety valves: maxDeliveryAttempt=16, maxInvisibleDuration=30min). - Credentials never travel in the stream: static AK/SK or a pluggable `CredentialsResolver` resolved locally per TaskManager and endpoint. - SDK-internal usage is confined to a single `@Internal` adapter, `RocketMQReceiptHandleCodec`. **Other** - Invisible-duration renewal policy (renews queue-resident messages under backpressure; handles are frozen once emitted). - gRPC producer sink and the `rocketmq-grpc` SQL connector. - Design document `docs/grpc-connector.md`: motivation, architecture, semantics, decisions and rejected alternatives. ### Verification - 50 module unit tests pass (serializer round-trips, reference counting, codec, throttle policy branches, etc.). - E2E against a real RocketMQ 5.x cluster: throttle defer (5 messages redelivered after ~10s then acked) and backpressure renewal (renewals=102, duplicates=0/60) both PASS. - `mvn spotless:check clean package` succeeds on JDK 11 and 17. -- 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]
