Apache Flink RocketMQ Connector

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

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

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/

Building the Apache Flink RocketMQ Connector from Source

Prerequisites:

• Unix-like environment (we use Linux, Mac OS X)
• Git
• Maven (we recommend version 3.8.6)
• Java 11

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

Quick Start

DataStream source & sink (remoting)

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();

SQL

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'
);

Documentation

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.

Developing

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.

Support

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.

Fork and Contribute

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.

About

Apache Flink and Apache RocketMQ are open source projects of The Apache Software Foundation (ASF).