[rocketmq-flink] branch main updated: Persist consumer offset to broker when new msg found from queues. (#37)

[duhengforever]

This is an automated email from the ASF dual-hosted git repository.

duhengforever pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/rocketmq-flink.git


The following commit(s) were added to refs/heads/main by this push:
     new 2d5c3ec  Persist consumer offset to broker when new msg found from 
queues. (#37)
2d5c3ec is described below

commit 2d5c3ec2b0a9fcd0259d5d0b64830a9ee48c96c0
Author: dinglei <libya_...@163.com>
AuthorDate: Mon Jul 11 11:44:09 2022 +0800

    Persist consumer offset to broker when new msg found from queues. (#37)
    
    Co-authored-by: shannon.dl <shannon...@alibaba-inc.com>
---
 .../java/org/apache/rocketmq/flink/legacy/RocketMQSourceFunction.java    | 1 +
 1 file changed, 1 insertion(+)

diff --git 
a/src/main/java/org/apache/rocketmq/flink/legacy/RocketMQSourceFunction.java 
b/src/main/java/org/apache/rocketmq/flink/legacy/RocketMQSourceFunction.java
index d4b59b3..6fb7875 100644
--- a/src/main/java/org/apache/rocketmq/flink/legacy/RocketMQSourceFunction.java
+++ b/src/main/java/org/apache/rocketmq/flink/legacy/RocketMQSourceFunction.java
@@ -392,6 +392,7 @@ public class RocketMQSourceFunction<OUT> extends 
RichParallelSourceFunction<OUT>
         offsetTable.put(mq, offset);
         if (!enableCheckpoint) {
             consumer.updateConsumeOffset(mq, offset);
+            consumer.getOffsetStore().persist(consumer.queueWithNamespace(mq));
         }
     }