[ISSUE #133] refactor: restructure to multi-module layout and harden the remoting connector (#134) Split the single-module project into flink-connector-rocketmq + flink-sql-connector-rocketmq, move legacy classes under streaming.connectors.rocketmq, and fix a batch of issues found in review and verified against an Aliyun commercial instance: - implement the no-op SQL row deserialization (source rows were empty) - close the producer in RocketMQWriter, rework AT_LEAST_ONCE to async send with pending counter and flush barrier, call serializationSchema.open() on construction - propagate fetch()/fetchOffsets failures instead of swallowing them - add consumer-side fallback for min/max/committed/timestamp offset lookup when admin broker-name resolution fails on cloud instances - implement sanityCheck() in source/sink builders - apply parameter passthrough fixes (send timeout/retry, batch size, namespace, filters, scan.mode auto-infer, transaction timeout unit) - wire USE_TLS, MQ_CLIENT_API_TIMEOUT, CLIENT_CALLBACK_EXECUTOR_THREADS and the pull flow-control options; drop dead options and unused code Tests: migrate unit tests to the new layout and add manual integration tests against a real commercial instance (FLIP-27 datastream pipeline, EXACTLY_ONCE two-phase commit, SQL DDL end-to-end); credentials are read from the environment. CI and docs: add GitHub Actions (spotless:check + clean package on a JDK 11/17 matrix, replacing the stale .travis.yml), rewrite the README following the official Flink connector layout, split detailed configuration into docs/ (remoting-connector.md, sql-connector.md, legacy-connector.md) and add url/licenses/mailing lists to the pom.
This repository contains the Apache Flink RocketMQ connector. It provides Flink sources and sinks that read from and write to Apache RocketMQ, for both the DataStream API and the Table/SQL API.
Two connector tracks are shipped:
| Track | Module | RocketMQ client | Best for |
|---|---|---|---|
| Remoting | flink-connector-rocketmq | rocketmq-client (remoting protocol) | Classic topics, FLIP-27 source + SinkV2 sink, SQL |
The SQL fat-jar is packaged by flink-sql-connector-rocketmq.
Apache Flink is an open source stream processing framework with powerful stream- and batch-processing capabilities.
Learn more about Flink at https://flink.apache.org/
Apache RocketMQ is a cloud native messaging and streaming platform, making it simple to build event-driven applications.
Learn more about RocketMQ at https://rocketmq.apache.org/
Prerequisites:
git clone https://github.com/apache/rocketmq-flink.git cd rocketmq-flink mvn clean package -DskipTests
The resulting jars can be found in the target directory of the respective module.
Before opening a pull request, format the code and run the tests:
mvn spotless:apply clean package
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(); env.enableCheckpointing(3000); RocketMQSource<String> source = RocketMQSource.<String>builder() .setEndpoints("127.0.0.1:9876") .setGroupId("GID-flink") .setTopics("flink-source") .setMinOffsets(OffsetsSelector.latest()) .setBodyOnlyDeserializer(new SimpleStringSchema()) .build(); RocketMQSink<String> sink = RocketMQSink.<String>builder() .setEndpoints("127.0.0.1:9876") .setGroupId("PID-flink") .setDeliveryGuarantee(DeliveryGuarantee.AT_LEAST_ONCE) .setSerializer(new MySerializationSchema("flink-sink")) .build(); env.fromSource(source, WatermarkStrategy.noWatermarks(), "rocketmq-source") .sinkTo(sink); env.execute();
CREATE TABLE rocketmq_source ( `id` BIGINT, `message` STRING ) WITH ( 'connector' = 'rocketmq', 'rocketmq.client.endpoints' = '127.0.0.1:9876', 'rocketmq.source.topic' = 'flink-source', 'rocketmq.source.group' = 'GID-flink' );
Connector documentation is located in the docs/ directory of this repository:
| Document | Content |
|---|---|
| docs/remoting-connector.md | Remoting DataStream connector: usage and all configuration options |
| docs/sql-connector.md | Table/SQL connector: DDL, metadata columns, fat-jar notes |
| docs/legacy-connector.md | Legacy RocketMQSourceFunction / RocketMQSink (deprecated) |
| docs/connector-overview.md | Feature comparison with other messaging connectors |
The documentation of Apache Flink is located on the website https://flink.apache.org, and the documentation of Apache RocketMQ on https://rocketmq.apache.org.
The Flink committers use IntelliJ IDEA to develop the Flink codebase. IntelliJ supports Maven out of the box. Check out the Setting up IntelliJ guide for details.
Run mvn spotless:apply before committing to keep the code style consistent.
Don't hesitate to ask!
Contact the developers and community on the Apache Flink mailing lists or the Apache RocketMQ mailing lists if you need any help.
Open an issue if you find a bug in the connector.
This is an active open-source project. We are always open to people who want to use the system or contribute to it. If you find a bug or want a new feature, open an issue first and discuss it with the community. See how to contribute to Apache RocketMQ and how to contribute to Apache Flink.
Apache Flink and Apache RocketMQ are open source projects of The Apache Software Foundation (ASF).