ROCKETMQ-82: RocketMQ-Flink Integration #45

vesense · 2018-01-27T11:11:38Z

What is the purpose of the change

RocketMQ integration for Apache Flink. This module includes the RocketMQ source and sink that allows a flink job to either write messages into a topic or read from topics in a flink job.
https://issues.apache.org/jira/browse/ROCKETMQ-82

Brief changelog

RocketMQSource - The RocketMQSource is based on RocketMQ pull consumer mode, and provides exactly once reliability guarantees when checkpoints are enabled.
Otherwise, the source doesn't provide any reliability guarantees.
RocketMQSink - The RocketMQSink provides at-least-once reliability guarantees when checkpoints are enabled and withBatchFlushOnCheckpoint(true) is set.
Otherwise, the sink reliability guarantees depends on rocketmq producer's retry policy, for this case, the messages sending way is sync by default,
but you can change it by invoking withAsync(true).
KeyValueDeserializationSchema - The main API for deserializing topic and tags is the org.apache.rocketmq.flink.common.serialization.KeyValueDeserializationSchema interface.
rocketmq-flink includes general purpose KeyValueDeserializationSchema implementations called SimpleKeyValueDeserializationSchema.
KeyValueSerializationSchema - The main API for serializing topic and tags is the org.apache.rocketmq.flink.common.serialization.KeyValueSerializationSchema interface.
rocketmq-flink includes general purpose KeyValueSerializationSchema implementations called SimpleKeyValueSerializationSchema.
TopicSelector - The main API for selecting topic and tags is the org.apache.rocketmq.flink.common.selector.TopicSelector interface.
rocketmq-flink includes general purpose TopicSelector implementations called DefaultTopicSelector and SimpleTopicSelector.
RocketMQFlinkExample - which receive messages from RocketMQ brokers and send messages to broker after processing.

Verifying this change

local & unit tests check passed ✅
apache-rat check passed ✅
checkstyle check passed ✅

vesense · 2018-01-27T11:15:51Z

@vongosling @zhouxinyu Do yo have time to take a look?

vesense · 2018-02-01T02:46:14Z

@vongosling @zhouxinyu ping..

vesense · 2018-02-06T14:58:54Z

Any comments are welcome..

vesense · 2018-02-23T03:20:34Z

Will add more unit tests later.

vesense · 2018-03-09T05:28:07Z

Added unit tests.

zhouxinyu · 2018-03-09T05:46:09Z

Cool, let's move this PR forward.

vesense · 2018-03-20T14:03:22Z

I'm glad to see that some users have used this patch in some scenes. But I would suggest to fork the code from official repo, not my personal.
And I will always keep my eyes on the RocketMQ streaming integration.

vesense · 2018-03-20T14:09:53Z

Also, I will update the status of integrating and SQL modules in community mailing list periodically.

zhouxinyu

I added some comments, please check.

zhouxinyu · 2018-03-20T14:11:51Z

rocketmq-flink/README.md

+| consumer.topic | consumer topic *Required*       |    null |
+| consumer.tag | consumer topic tag      |    * |
+| consumer.offset.reset.to | what to do when there is no initial offset on the server      |   latest/earliest/timestamp |
+| consumer.offset.from.timestamp | the timestamp when `consumer.offset.reset.to=timestamp` was set   |   $TIMESTAMP |


The default value $TIMESTAMP and $UUID are confusing, especially the $ symbol, a short description may be better.

Good catch. Will fix.

zhouxinyu · 2018-03-20T14:14:20Z

rocketmq-flink/pom.xml

+        <developer>
+            <id>vesense</id>
+            <name>Xin Wang</name>
+            <email>xinwang@apache.org</email>


Not recommended to use personal user info, see: https://github.com/apache/rocketmq/blob/master/pom.xml

Will remove.

zhouxinyu · 2018-03-20T14:15:22Z

rocketmq-flink/pom.xml

+        <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
+        <!--maven properties -->
+        <maven.test.skip>true</maven.test.skip>
+        <maven.javadoc.skip>true</maven.javadoc.skip>


IMO, it's not a good practice that skip test and javadoc.

zhouxinyu · 2018-03-20T14:16:36Z

rocketmq-flink/pom.xml

+                    <!--<groupId>ch.qos.logback</groupId>-->
+                    <!--<artifactId>logback-classic</artifactId>-->
+                <!--</exclusion>-->
+            <!--</exclusions>-->


If the exclusions section isn't necessary, just remove it and make code clean.

Will remove.

zhouxinyu · 2018-03-20T14:17:21Z

rocketmq-flink/src/main/java/org/apache/rocketmq/flink/RocketMqConfig.java

+/**
+ * RocketMqConfig for Consumer/Producer.
+ */
+public class RocketMqConfig {


Use RocketMQConfig

Also RocketMqUtils -> RocketMQUtils.

zhouxinyu · 2018-03-20T14:20:19Z

rocketmq-flink/src/main/java/org/apache/rocketmq/flink/RocketMQSink.java

+                producer.send(msg, new SendCallback() {
+                    @Override
+                    public void onSuccess(SendResult sendResult) {
+                        LOG.debug("Async send message success!");


how about appending the message id in send result to debug?

I think we can print the SendResult here following the way of sync sending.

zhouxinyu · 2018-03-20T14:21:07Z

rocketmq-flink/src/main/java/org/apache/rocketmq/flink/RocketMQSink.java

+            // sync sending, will return a SendResult
+            try {
+                SendResult result = producer.send(msg);
+                if (LOG.isDebugEnabled()) {


No need to check whether debug is enabled.

Yes. Will remove.

zhouxinyu · 2018-03-20T14:25:50Z

rocketmq-flink/src/main/java/org/apache/rocketmq/flink/RocketMQSource.java

+                                // output and state update are atomic
+                                synchronized (lock) {
+                                    context.collectWithTimestamp(data, msg.getBornTimestamp());
+                                    putMessageQueueOffset(mq, pullResult.getNextBeginOffset());


Do we need use the next begin offset in other PullStatus?

Yes. We need the nextBeginOffset in the PullResult to update the offset table which will be checkpointing when snapshotState invoked .

I mean we also should update the offset when the pullresult is OFFSET_ILLEGAL or NO_MATCHED_MSG

Got it. Will fix.

vesense · 2018-03-20T15:06:38Z

@zhouxinyu Addressed your comments.

vesense · 2018-03-21T11:34:23Z

@zhouxinyu Addressed all of your comments. Please take a look again.

zhouxinyu · 2018-03-21T14:02:19Z

rocketmq-flink/src/main/java/org/apache/rocketmq/flink/RocketMQSink.java

+        Message msg = prepareMessage(input);
+
+        if (batchFlushOnCheckpoint) {
+            batchList.add(msg);


Add action isn't protected by the lock batchList， is this a concurrent issue?

Not an issue. Each RocketMQSink instance belong to a thread. We just need to sync batchList when snapshotState invoked.

zhouxinyu · 2018-03-22T03:11:47Z

LGTM now~

vesense · 2018-03-22T06:30:02Z

Thanks @zhouxinyu

vesense · 2018-03-23T03:11:38Z

@zhouxinyu Can we merge this in?

vongosling · 2018-03-23T11:32:40Z

LGTM

ROCKETMQ-82: RocketMQ-Flink Integration

f95b989

vesense mentioned this pull request Jan 27, 2018

We need flink-rocketmq-connector apache/rocketmq#175

Closed

Added unit tests

6080c63

zhouxinyu reviewed Mar 20, 2018

View reviewed changes

Addressed review comments

fb08756

Addressed review comments

09d9cd9

zhouxinyu reviewed Mar 21, 2018

View reviewed changes

vongosling approved these changes Mar 23, 2018

View reviewed changes

zhouxinyu merged commit 45efa39 into apache:master Mar 23, 2018

ROCKETMQ-82: RocketMQ-Flink Integration #45

ROCKETMQ-82: RocketMQ-Flink Integration #45

Conversation

vesense commented Jan 27, 2018 • edited Loading

What is the purpose of the change

Brief changelog

Verifying this change

vesense commented Jan 27, 2018

vesense commented Feb 1, 2018

vesense commented Feb 6, 2018

vesense commented Feb 23, 2018

vesense commented Mar 9, 2018

zhouxinyu commented Mar 9, 2018

vesense commented Mar 20, 2018

vesense commented Mar 20, 2018

zhouxinyu left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

vesense commented Mar 20, 2018

vesense commented Mar 21, 2018

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

zhouxinyu commented Mar 22, 2018

vesense commented Mar 22, 2018

vesense commented Mar 23, 2018

vongosling commented Mar 23, 2018

vesense commented Jan 27, 2018 •

edited

Loading