rocketmq-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From duhengfore...@apache.org
Subject [rocketmq-streams] branch async-ck updated: add checkpoint storage (#69)
Date Sat, 18 Sep 2021 11:39:17 GMT
This is an automated email from the ASF dual-hosted git repository.

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


The following commit(s) were added to refs/heads/async-ck by this push:
     new 2766990  add checkpoint storage (#69)
2766990 is described below

commit 276699067024103b6f35da4699c44d05848ad0f4
Author: cyril <cw68ster@gmail.com>
AuthorDate: Sat Sep 18 19:39:13 2021 +0800

    add checkpoint storage (#69)
    
    * add channel-db module
    
    * add channel-configurable
    
    * add chinese
    
    * fixed DataStream flatMap
    
    * U
    
    * stash & pre merge
    
    * add license for chackpoint/pom.xml
    
    * add license for checkpoint/pom.xml
    
    * fix license
    
    Co-authored-by: vv <zengyu.cw@alibaba-inc.com>
---
 README-chinese.md                                  | 106 +++++++++
 pom.xml                                            |   1 +
 .../pom.xml                                        |  44 +++-
 .../streams/checkpoint/db/DBCheckPointStorage.java |  65 ++++++
 .../streams/common/channel/AbstractChannel.java    |  10 +
 .../common/channel/impl/memory/MemoryChannel.java  |   5 +
 .../streams/common/channel/sink/AbstractSink.java  |   2 +-
 .../common/channel/source/AbstractSource.java      |  53 ++++-
 .../streams/common/channel/source/ISource.java     |  18 ++
 .../checkpoint/AbstractCheckPointStorage.java      | 184 +++++++++++++++
 .../streams/common/checkpoint/CheckPoint.java      |  98 ++++++--
 .../common/checkpoint/CheckPointManager.java       | 254 ++++++++-------------
 .../common/checkpoint/CheckPointMessage.java       |  10 +
 .../checkpoint/CheckPointStorageFactory.java       |  87 +++++++
 .../ICheckPointStorage.java}                       |  24 +-
 .../streams/common/checkpoint/SourceSnapShot.java  |  79 +++++++
 .../streams/common/checkpoint/SourceState.java     |  57 +++++
 .../streams/common/configure/ConfigureFileKey.java |   6 +
 .../streams/common/functions/FlatMapFunction.java  |   4 +-
 .../optimization/cachefilter/CacheFilterMeta.java  |  18 +-
 rocketmq-streams-configurable/pom.xml              |   1 +
 rocketmq-streams-state/pom.xml                     |   0
 22 files changed, 923 insertions(+), 203 deletions(-)

diff --git a/README-chinese.md b/README-chinese.md
new file mode 100644
index 0000000..5d8215f
--- /dev/null
+++ b/README-chinese.md
@@ -0,0 +1,106 @@
+# RocketMQ Streams
+## Features
+
+* 轻量级部署:可以单独部署,也支持集群部署
+* 多种类型的数据输入以及输出,source 支持 rocketmq , sink 支持db, rocketmq 等
+
+## DataStream Example
+
+```java
+import org.apache.rocketmq.streams.client.transform.DataStream;
+
+DataStreamSource source=StreamBuilder.dataStream("namespace","pipeline");
+
+    source
+    .fromFile("~/admin/data/text.txt",false)
+    .map(message->message)
+    .toPrint(1)
+    .start();
+```
+
+## Maven Repository
+
+```xml
+
+<dependency>
+    <groupId>org.apache.rocketmq</groupId>
+    <artifactId>rocketmq-streams-clients</artifactId>
+    <version>1.0.0-SNAPSHOT</version>
+</dependency>
+```
+
+# Core API
+
+rocketmq-stream 实现了一系列高级的API,可以让用户很方便的编写流计算的程序,实现自己的业务需求;
+
+## StreamBuilder
+
+StreamBuilder 用于构建流任务的源; 内部包含```dataStream()```和```tableStream()```俩个方法,分别返回DataStreamSource和TableStreamSource俩个源;
+
++ [dataStream(nameSpaceName,pipelineName)]() 返回DataStreamSource实例,用于分段编程实现流计算任务;
++ [tableStream(nameSpaceName,pipelineName)]()返回TableStreamSource实例, 用于脚本编程实现流计算任务;
+
+## DataStream API
+
+### Source
+
+DataStreamSource 是分段式编程的源头类,用于对接各种数据源, 从各大消息队列中获取数据;
+
++ ```fromFile```  从文件中读取数据, 该方法包含俩个参数
+  + ```filePath``` 文件路径,必填参数
+  + ```isJsonData```  是否json数据, 非必填参数, 默认为```true```
+
+
++ ```fromRocketmq``` 从rocketmq中获取数据,包含四个参数
+  + ```topic``` rocketmq消息队列的topic名称,必填参数
+  + ```groupName``` 消费者组的名称,必填参数
+  + ```isJson``` 是否json格式,非必填参数
+  + ```tags``` rocketmq消费的tags值,用于过滤消息,非必填参数
+
++ ```from``` 自定义的数据源, 通过实现ISource接口实现自己的数据源
+
+### transform
+
+transform 允许在流计算过程中对输入源的数据进行修改,进行下一步的操作;DataStream API中包括```DataStream```,```JoinStream```, ```SplitStream```,```WindowStream```等多个transform类;
+
+#### DataStream
+
+DataStream实现了一系列常见的流计算算子
+
++ ```map``` 通过将源的每个记录传递给函数func来返回一个新的DataStream
++ ```flatmap``` 与map类似,一个输入项对应0个或者多个输出项
++ ```filter``` 只选择func返回true的源DStream的记录来返回一个新的DStream
++ ```forEach``` 对每个记录执行一次函数func, 返回一个新的DataStream
++ ```selectFields``` 对每个记录返回对应的字段值,返回一个新的DataStream
++ ```operate```  对每个记录执行一次自定义的函数,返回一个新的DataStream
++ ```script```  针对每个记录的字段执行一段脚本,返回新的字段,生成一个新的DataStream
++ ```toPrint``` 将结果在控制台打印,生成新的DataStreamAction实例
++ ```toFile``` 将结果保存为文件,生成一个新的DataStreamAction实例
++ ```toDB``` 将结果保存到数据库
++ ```toRocketmq``` 将结果输出到rocketmq
++ ```to``` 将结果经过自定义的ISink接口输出到指定的存储
++ ```window``` 在窗口内进行相关的统计分析,一般会与```groupBy```连用, ```window()```用来定义窗口的大小, ```groupBy()```用来定义统计分析的主key,可以指定多个
+  + ```count``` 在窗口内计数
+  + ```min``` 获取窗口内统计值的最小值
+  + ```max``` 获取窗口内统计值得最大值
+  + ```avg``` 获取窗口内统计值的平均值
+  + ```sum``` 获取窗口内统计值的加和值
+  + ```reduce``` 在窗口内进行自定义的汇总运算
++ ```join``` 根据条件将将俩个流进行关联, 合并为一个大流进行相关的运算
++ ```union``` 将俩个流进行合并
++ ```split``` 将一个数据流按照标签进行拆分,分为不同的数据流供下游进行分析计算
++ ```with``` with算子用来指定计算过程中的相关策略,包括checkpoint的存储策略,state的存储策略等
+
+# Strategy
+
+策略机制主要用来控制计算引擎运行过程中的底层逻辑,如checkpoint,state的存储方式等,后续还会增加对窗口、双流join等的控制;所有的控制策略通过```with```算子传入,可以同时传入多个策略类型;
+
+```java
+//指定checkpoint的存储策略
+source
+    .fromRocketmq("TSG_META_INFO","")
+    .map(message->message+"--")
+    .toPrint(1)
+    .with(CheckpointStrategy.db("jdbc:mysql://XXXXX:3306/XXXXX","","",0L))
+    .start();
+```
diff --git a/pom.xml b/pom.xml
index 7bbe1f4..6bdd9ee 100644
--- a/pom.xml
+++ b/pom.xml
@@ -52,6 +52,7 @@
         <module>rocketmq-streams-channel-http</module>
         <module>rocketmq-streams-state</module>
         <module>rocketmq-streams-examples</module>
+        <module>rocketmq-streams-checkpoint</module>
 
     </modules>
 
diff --git a/rocketmq-streams-configurable/pom.xml b/rocketmq-streams-checkpoint/pom.xml
old mode 100755
new mode 100644
similarity index 54%
copy from rocketmq-streams-configurable/pom.xml
copy to rocketmq-streams-checkpoint/pom.xml
index f160060..40a28c0
--- a/rocketmq-streams-configurable/pom.xml
+++ b/rocketmq-streams-checkpoint/pom.xml
@@ -1,4 +1,4 @@
-<?xml version="1.0" encoding="utf-8"?>
+<?xml version="1.0" encoding="UTF-8"?>
 <!--
   Licensed to the Apache Software Foundation (ASF) under one or more
   contributor license agreements.  See the NOTICE file distributed with
@@ -18,19 +18,49 @@
 <project xmlns="http://maven.apache.org/POM/4.0.0"
          xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
          xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
-    <modelVersion>4.0.0</modelVersion>
     <parent>
-        <groupId>org.apache.rocketmq</groupId>
         <artifactId>rocketmq-streams</artifactId>
+        <groupId>org.apache.rocketmq</groupId>
         <version>1.0.0-SNAPSHOT</version>
     </parent>
-    <artifactId>rocketmq-streams-configurable</artifactId>
-    <name>ROCKETMQ STREAMS :: configurable</name>
+    <modelVersion>4.0.0</modelVersion>
+
+    <artifactId>rocketmq-streams-checkpoint</artifactId>
+    <name>ROCKETMQ STREAMS :: checkpoint</name>
     <packaging>jar</packaging>
+
+    <properties>
+        <maven.compiler.source>8</maven.compiler.source>
+        <maven.compiler.target>8</maven.compiler.target>
+    </properties>
     <dependencies>
         <dependency>
             <groupId>org.apache.rocketmq</groupId>
-            <artifactId>rocketmq-streams-serviceloader</artifactId>
+            <artifactId>rocketmq-streams-commons</artifactId>
+            <exclusions>
+                <exclusion>
+                    <groupId>com.google.auto.service</groupId>
+                    <artifactId>auto-service</artifactId>
+                </exclusion>
+            </exclusions>
+        </dependency>
+
+        <dependency>
+            <groupId>org.apache.rocketmq</groupId>
+            <artifactId>rocketmq-streams-db-operator</artifactId>
+            <exclusions>
+                <exclusion>
+                    <groupId>com.google.auto.service</groupId>
+                    <artifactId>auto-service</artifactId>
+                </exclusion>
+            </exclusions>
+        </dependency>
+
+        <dependency>
+            <groupId>com.google.auto.service</groupId>
+            <artifactId>auto-service</artifactId>
+            <optional>true</optional>
         </dependency>
     </dependencies>
-</project>
+
+</project>
\ No newline at end of file
diff --git a/rocketmq-streams-checkpoint/src/main/java/org/apache/rocketmq/streams/checkpoint/db/DBCheckPointStorage.java b/rocketmq-streams-checkpoint/src/main/java/org/apache/rocketmq/streams/checkpoint/db/DBCheckPointStorage.java
new file mode 100644
index 0000000..df1d046
--- /dev/null
+++ b/rocketmq-streams-checkpoint/src/main/java/org/apache/rocketmq/streams/checkpoint/db/DBCheckPointStorage.java
@@ -0,0 +1,65 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.rocketmq.streams.checkpoint.db;
+
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import org.apache.rocketmq.streams.common.channel.source.ISource;
+import org.apache.rocketmq.streams.common.checkpoint.AbstractCheckPointStorage;
+import org.apache.rocketmq.streams.common.checkpoint.CheckPoint;
+import org.apache.rocketmq.streams.common.checkpoint.CheckPointManager;
+import org.apache.rocketmq.streams.common.checkpoint.SourceSnapShot;
+import org.apache.rocketmq.streams.db.driver.orm.ORMUtil;
+
+import java.util.List;
+
+/**
+ * @description
+ */
+public class DBCheckPointStorage extends AbstractCheckPointStorage {
+
+    static final Log logger = LogFactory.getLog(DBCheckPointStorage.class);
+
+    static final String STORAGE_NAME = "DB";
+
+    public DBCheckPointStorage(){
+
+    }
+
+    @Override
+    public String getStorageName() {
+        return STORAGE_NAME;
+    }
+
+    @Override
+    public <T> void save(List<T> checkPointState) {
+        logger.info(String.format("save checkpoint size %d", checkPointState.size()));
+        ORMUtil.batchReplaceInto(checkPointState);
+    }
+
+    @Override
+    //todo
+    public CheckPoint recover(ISource iSource, String queueId) {
+        String sourceName = CheckPointManager.createSourceName(iSource, null);
+        String key = CheckPointManager.createCheckPointKey(sourceName, queueId);
+        String sql = "select * from source_snap_shot where `key` = " + "'" + key + "';";
+        SourceSnapShot snapShot = ORMUtil.queryForObject(sql, null, SourceSnapShot.class);
+
+        logger.info(String.format("checkpoint recover key is %s, sql is %s, recover sourceSnapShot : %s", key, sql, snapShot == null ? "null snapShot" : snapShot.toString()));
+        return new CheckPoint().fromSnapShot(snapShot);
+    }
+}
diff --git a/rocketmq-streams-commons/src/main/java/org/apache/rocketmq/streams/common/channel/AbstractChannel.java b/rocketmq-streams-commons/src/main/java/org/apache/rocketmq/streams/common/channel/AbstractChannel.java
index 74ff5f4..7765669 100644
--- a/rocketmq-streams-commons/src/main/java/org/apache/rocketmq/streams/common/channel/AbstractChannel.java
+++ b/rocketmq-streams-commons/src/main/java/org/apache/rocketmq/streams/common/channel/AbstractChannel.java
@@ -227,4 +227,14 @@ public abstract class AbstractChannel extends BasedConfigurable implements IChan
         create();
         ((AbstractSource)source).setJsonData(isJsonData);
     }
+
+    @Override
+    public String getTopic(){
+        return source.getTopic();
+    }
+
+    @Override
+    public void setTopic(String topic){
+        source.setTopic(topic);
+    }
 }
diff --git a/rocketmq-streams-commons/src/main/java/org/apache/rocketmq/streams/common/channel/impl/memory/MemoryChannel.java b/rocketmq-streams-commons/src/main/java/org/apache/rocketmq/streams/common/channel/impl/memory/MemoryChannel.java
index 06dff1c..f3d06ee 100644
--- a/rocketmq-streams-commons/src/main/java/org/apache/rocketmq/streams/common/channel/impl/memory/MemoryChannel.java
+++ b/rocketmq-streams-commons/src/main/java/org/apache/rocketmq/streams/common/channel/impl/memory/MemoryChannel.java
@@ -74,4 +74,9 @@ public class MemoryChannel extends AbstractChannel {
             }
         };
     }
+
+    @Override
+    public String createCheckPointName() {
+        return "memory-source";
+    }
 }
diff --git a/rocketmq-streams-commons/src/main/java/org/apache/rocketmq/streams/common/channel/sink/AbstractSink.java b/rocketmq-streams-commons/src/main/java/org/apache/rocketmq/streams/common/channel/sink/AbstractSink.java
index 82218c2..aa7c1bc 100644
--- a/rocketmq-streams-commons/src/main/java/org/apache/rocketmq/streams/common/channel/sink/AbstractSink.java
+++ b/rocketmq-streams-commons/src/main/java/org/apache/rocketmq/streams/common/channel/sink/AbstractSink.java
@@ -32,8 +32,8 @@ import org.apache.rocketmq.streams.common.channel.sinkcache.impl.MultiSplitMessa
 import org.apache.rocketmq.streams.common.channel.source.ISource;
 import org.apache.rocketmq.streams.common.channel.split.ISplit;
 import org.apache.rocketmq.streams.common.checkpoint.CheckPointManager;
-import org.apache.rocketmq.streams.common.checkpoint.CheckPointManager.SourceState;
 import org.apache.rocketmq.streams.common.checkpoint.CheckPointMessage;
+import org.apache.rocketmq.streams.common.checkpoint.SourceState;
 import org.apache.rocketmq.streams.common.configurable.BasedConfigurable;
 import org.apache.rocketmq.streams.common.configurable.IConfigurableIdentification;
 import org.apache.rocketmq.streams.common.context.IMessage;
diff --git a/rocketmq-streams-commons/src/main/java/org/apache/rocketmq/streams/common/channel/source/AbstractSource.java b/rocketmq-streams-commons/src/main/java/org/apache/rocketmq/streams/common/channel/source/AbstractSource.java
index f5de98b..bf45e7e 100644
--- a/rocketmq-streams-commons/src/main/java/org/apache/rocketmq/streams/common/channel/source/AbstractSource.java
+++ b/rocketmq-streams-commons/src/main/java/org/apache/rocketmq/streams/common/channel/source/AbstractSource.java
@@ -42,6 +42,7 @@ import org.apache.rocketmq.streams.common.context.MessageHeader;
 import org.apache.rocketmq.streams.common.context.UserDefinedMessage;
 import org.apache.rocketmq.streams.common.interfaces.IStreamOperator;
 import org.apache.rocketmq.streams.common.topology.builder.PipelineBuilder;
+import org.apache.rocketmq.streams.common.utils.MapKeyUtil;
 import org.apache.rocketmq.streams.common.utils.StringUtil;
 
 /**
@@ -75,6 +76,7 @@ public abstract class AbstractSource extends BasedConfigurable implements ISourc
 
     protected List<String> logFingerprintFields;//log fingerprint to filter msg quickly
 
+
     /**
      * 数据源投递消息的算子,此算子用来接收source的数据,做处理
      */
@@ -223,18 +225,22 @@ public abstract class AbstractSource extends BasedConfigurable implements ISourc
         return createJson(message);
     }
 
-
+    /**
+     * 交给receiver执行后续逻辑
+     *
+     * @param channelMessage
+     * @return
+     */
     public AbstractContext executeMessage(Message channelMessage) {
         AbstractContext context = new Context(channelMessage);
-        if (!channelMessage.getHeader().isSystemMessage()) {
-            messageQueueChangedCheck(channelMessage.getHeader());
-        }
-
         if (isSplitInRemoving(channelMessage)) {
             return context;
         }
+        if (!channelMessage.getHeader().isSystemMessage()) {
+            messageQueueChangedCheck(channelMessage.getHeader());
+        }
 
-        boolean needFlush = !channelMessage.getHeader().isSystemMessage() && channelMessage.getHeader().isNeedFlush();
+        boolean needFlush = channelMessage.getHeader().isSystemMessage() == false && channelMessage.getHeader().isNeedFlush();
 
         if (receiver != null) {
             receiver.doMessage(channelMessage, context);
@@ -277,6 +283,9 @@ public abstract class AbstractSource extends BasedConfigurable implements ISourc
      * @param header
      */
     protected void messageQueueChangedCheck(MessageHeader header) {
+        if (supportNewSplitFind() && supportRemoveSplitFind()) {
+            return;
+        }
         Set<String> queueIds = new HashSet<>();
         String msgQueueId = header.getQueueId();
         if (StringUtil.isNotEmpty(msgQueueId)) {
@@ -287,7 +296,7 @@ public abstract class AbstractSource extends BasedConfigurable implements ISourc
             queueIds.addAll(checkpointQueueIds);
         }
         Set<String> newQueueIds = new HashSet<>();
-
+        Set<String> removeQueueIds = new HashSet<>();
         for (String queueId : queueIds) {
             if (isNotDataSplit(queueId)) {
                 continue;
@@ -536,4 +545,34 @@ public abstract class AbstractSource extends BasedConfigurable implements ISourc
         return isBatchMessage;
     }
 
+    @Override
+    public String createCheckPointName(){
+
+        ISource source = this;
+
+        String namespace = source.getNameSpace();
+        String name = source.getConfigureName();
+        String groupName = source.getGroupName();
+
+
+        if(StringUtil.isEmpty(namespace)){
+            namespace = "default_namespace";
+        }
+
+        if(StringUtil.isEmpty(name)){
+            name = "default_name";
+        }
+
+        if(StringUtil.isEmpty(groupName)){
+            groupName = "default_groupName";
+        }
+        String topic = source.getTopic();
+        if(topic == null || topic.trim().length() == 0){
+            topic = "default_topic";
+        }
+        return MapKeyUtil.createKey(namespace, groupName, topic, name);
+
+    }
+
+
 }
diff --git a/rocketmq-streams-commons/src/main/java/org/apache/rocketmq/streams/common/channel/source/ISource.java b/rocketmq-streams-commons/src/main/java/org/apache/rocketmq/streams/common/channel/source/ISource.java
index 4b9f55f..d3f6802 100644
--- a/rocketmq-streams-commons/src/main/java/org/apache/rocketmq/streams/common/channel/source/ISource.java
+++ b/rocketmq-streams-commons/src/main/java/org/apache/rocketmq/streams/common/channel/source/ISource.java
@@ -74,4 +74,22 @@ public interface ISource<T extends ISource> extends IConfigurable, IStageBuilder
      */
     long getCheckpointTime();
 
+    /**
+     *
+     * @param topic
+     */
+    void setTopic(String topic);
+
+    /**
+     *
+     * @return
+     */
+    String getTopic();
+
+    /**
+     * 创建checkpoint名字
+     * @return checkpoint key name
+     */
+    String createCheckPointName();
+
 }
diff --git a/rocketmq-streams-commons/src/main/java/org/apache/rocketmq/streams/common/checkpoint/AbstractCheckPointStorage.java b/rocketmq-streams-commons/src/main/java/org/apache/rocketmq/streams/common/checkpoint/AbstractCheckPointStorage.java
new file mode 100644
index 0000000..fad0ae1
--- /dev/null
+++ b/rocketmq-streams-commons/src/main/java/org/apache/rocketmq/streams/common/checkpoint/AbstractCheckPointStorage.java
@@ -0,0 +1,184 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.rocketmq.streams.common.checkpoint;
+
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import org.apache.rocketmq.streams.common.channel.sinkcache.IMessageCache;
+import org.apache.rocketmq.streams.common.channel.sinkcache.IMessageFlushCallBack;
+import org.apache.rocketmq.streams.common.channel.sinkcache.impl.MessageCache;
+import org.apache.rocketmq.streams.common.context.MessageOffset;
+
+import java.util.*;
+
+public abstract class AbstractCheckPointStorage implements ICheckPointStorage{
+
+    static final Log logger = LogFactory.getLog(AbstractCheckPointStorage.class);
+
+    protected transient IMessageCache<CheckPointMessage> messageCache;
+
+    public AbstractCheckPointStorage(){
+        messageCache = new MessageCache<>(new IMessageFlushCallBack<CheckPointMessage>() {
+            @Override
+            public boolean flushMessage(List<CheckPointMessage> messages) {
+                //合并最近的checkpoint,只存储一次
+                // key 为 sourceName, Value中的map, k : v = queueid : offset
+                Map<String, SourceState> sourceStateMap = mergeSourceState(messages);
+
+                logger.info(String.format("flushMessage raw size %d, merge size %d", messages.size(), sourceStateMap.size()));
+                logger.info("flushMessage : " + messages.get(0).getCheckPointStates().get(0).getQueueIdAndOffset().toString());
+
+                saveCheckPoint(sourceStateMap);
+                return true;
+            }
+        });
+        ((MessageCache)messageCache).setAutoFlushSize(10);
+        ((MessageCache)messageCache).setAutoFlushTimeGap(500);
+        messageCache.openAutoFlush();
+    }
+
+    public void flush(){
+        messageCache.flush();
+    }
+
+
+    /**
+     * 可能有多次的offset合并在一起,对offset合并
+     * 合并包含两个逻辑:1.同1个CheckPointMessage中,选择最小的作为本次的offset
+     * @param messages
+     */
+    protected  Map<String, SourceState> mergeSourceState(List<CheckPointMessage> messages) {
+        Map<String,SourceState> sourceStateMap = new HashMap<>();
+        for(CheckPointMessage checkPointMessage:messages){
+            SourceState sourceState = createSourceState(checkPointMessage);
+            if(sourceState == null){
+                continue;
+            }
+            String sourceName = sourceState.getSourceName();
+            SourceState existSourceState = sourceStateMap.get(sourceName);
+            SourceState lastSourceState = sourceState;
+            if(existSourceState != null){
+                lastSourceState = merge(sourceState,existSourceState);
+            }
+            sourceStateMap.put(sourceName,lastSourceState);
+        }
+        return sourceStateMap;
+    }
+
+    /**
+     * 如果多次的checkpoint在一起,先合并再保存
+     * @param sourceState
+     * @param existSourceState
+     * @return
+     */
+    protected SourceState merge(SourceState sourceState, SourceState existSourceState) {
+        Iterator<Map.Entry<String, MessageOffset>> it = sourceState.getQueueId2Offsets().entrySet()
+            .iterator();
+        while (it.hasNext()){
+            Map.Entry<String, MessageOffset> entry=it.next();
+            String queueId = entry.getKey();
+            MessageOffset offset = entry.getValue();
+            MessageOffset existOffset = existSourceState.getQueueId2Offsets().get(queueId);
+            if(existOffset == null){
+                existSourceState.getQueueId2Offsets().put(queueId,offset);
+            }else {
+                boolean isGreateThan = offset.greateThan(existOffset.getOffsetStr());
+                if(isGreateThan){
+                    existSourceState.getQueueId2Offsets().put(queueId,offset);
+                }
+            }
+        }
+        return existSourceState;
+    }
+
+    /**
+     * 一个pipeline流程中,找最小的offset提交保存
+     * @param checkPointMessage
+     * @return
+     */
+    protected SourceState createSourceState(CheckPointMessage checkPointMessage) {
+        SourceState sourceState = new SourceState();
+        String pipelineName = checkPointMessage.getPipelineName();
+
+        Map<String, MessageOffset> queueId2Offsets = new HashMap<>();
+        sourceState.setSourceName(CheckPointManager.createSourceName(checkPointMessage.getSource(),pipelineName));
+        sourceState.setQueueId2Offsets(queueId2Offsets);
+
+        for(CheckPointState checkPointState:checkPointMessage.getCheckPointStates()){
+
+            if(checkPointState.isReplyAnyOny()){
+                continue;
+            }
+            if(checkPointState.isReplyRefuse()){
+                return null;
+            }
+            Iterator<Map.Entry<String, MessageOffset>> it = checkPointState.getQueueIdAndOffset().entrySet()
+                .iterator();
+            while (it.hasNext()){
+                Map.Entry<String, MessageOffset> entry=it.next();
+                String queueId=entry.getKey();
+                MessageOffset offset = entry.getValue();
+                MessageOffset existOffset = queueId2Offsets.get(queueId);
+                if(existOffset==null){
+                    queueId2Offsets.put(queueId,offset);
+                }else {
+                    boolean isGreateThan=existOffset.greateThan(offset.getOffsetStr());
+                    if(isGreateThan){
+                        queueId2Offsets.put(queueId,offset);
+                    }else {
+                        queueId2Offsets.put(queueId,existOffset);
+                    }
+                }
+            }
+        }
+        return sourceState;
+    }
+
+    /**
+     * 先查询现在数据源的分片,如果已经不处理的分片,不做保存
+     * 否则把结果保存到db中
+     * @param sourceStateMap
+     */
+    protected void saveCheckPoint(Map<String, SourceState> sourceStateMap) {
+
+        List<SourceSnapShot> checkPoints = new ArrayList<>();
+
+        for(SourceState sourceState:sourceStateMap.values()){
+            for(Map.Entry<String, MessageOffset> entry : sourceState.getQueueId2Offsets().entrySet()){
+                CheckPoint checkPoint = new CheckPoint();
+                checkPoint.setSourceName(sourceState.getSourceName());
+                checkPoint.setQueueId(entry.getKey());
+                checkPoint.setData(entry.getValue().getMainOffset());
+                checkPoint.setGmtCreate(new Date());
+                checkPoint.setGmtModified(new Date());
+                SourceSnapShot object = checkPoint.toSnapShot();
+                checkPoints.add(object);
+
+            }
+        }
+        save(checkPoints);
+    }
+
+    public void addCheckPointMessage(CheckPointMessage message){
+        List<CheckPointState> states = message.getCheckPointStates();
+        for(CheckPointState state : states){
+            logger.debug(String.format("addCheckPointMessage states %s", state.getQueueIdAndOffset().toString()));
+        }
+        messageCache.addCache(message);
+    }
+
+}
diff --git a/rocketmq-streams-commons/src/main/java/org/apache/rocketmq/streams/common/checkpoint/CheckPoint.java b/rocketmq-streams-commons/src/main/java/org/apache/rocketmq/streams/common/checkpoint/CheckPoint.java
index 4eb7211..8bcedbd 100644
--- a/rocketmq-streams-commons/src/main/java/org/apache/rocketmq/streams/common/checkpoint/CheckPoint.java
+++ b/rocketmq-streams-commons/src/main/java/org/apache/rocketmq/streams/common/checkpoint/CheckPoint.java
@@ -16,28 +16,24 @@
  */
 package org.apache.rocketmq.streams.common.checkpoint;
 
+import com.alibaba.fastjson.JSONObject;
 import org.apache.rocketmq.streams.common.channel.source.ISource;
 import org.apache.rocketmq.streams.common.model.Entity;
 
 /**
  * model for checkpoint,need save in store
  */
-public class CheckPoint extends Entity {
+public class CheckPoint<T> extends Entity {
+
     protected String sourceNamespace;
+    protected String pipelineName;
     protected String sourceName;
     protected String queueId;
-    protected String offset;
+    protected String topic;
 
-    public static String loadOffset(ISource source, String splitId) {
-        return null;
-        //Set<String> splits=new HashSet<>();
-        //splits.add(splitId);
-        //Map<String,String> queueId2Offset=loadOffset(source,splits);
-        //if(queueId2Offset==null||queueId2Offset.containsKey(splitId)==false){
-        //    return null;
-        //}
-        //return queueId2Offset.get(splitId);
-    }
+    protected T data;
+
+    protected JSONObject content;
 
     public String getQueueId() {
         return queueId;
@@ -47,13 +43,6 @@ public class CheckPoint extends Entity {
         this.queueId = queueId;
     }
 
-    public String getOffset() {
-        return offset;
-    }
-
-    public void setOffset(String offset) {
-        this.offset = offset;
-    }
 
     public String getSourceNamespace() {
         return sourceNamespace;
@@ -70,4 +59,73 @@ public class CheckPoint extends Entity {
     public void setSourceName(String sourceName) {
         this.sourceName = sourceName;
     }
-}
+
+    public String getPipelineName() {
+        return pipelineName;
+    }
+
+    public void setPipelineName(String pipelineName) {
+        this.pipelineName = pipelineName;
+    }
+
+    public JSONObject getContent() {
+        return content;
+    }
+
+    public void setContent(JSONObject content) {
+        this.content = content;
+    }
+
+    public T getData() {
+        return data;
+    }
+
+    public void setData(T data) {
+        this.data = data;
+    }
+
+    public String getTopic() {
+        return topic;
+    }
+
+    public void setTopic(String topic) {
+        this.topic = topic;
+    }
+
+    public SourceSnapShot toSnapShot(){
+        SourceSnapShot snapShot = new SourceSnapShot();
+        snapShot.setGmtCreate(gmtCreate);
+        snapShot.setGmtModified(gmtModified);
+        snapShot.setKey(CheckPointManager.createCheckPointKey(sourceName, queueId));
+        if(content == null){
+            content = new JSONObject();
+        }
+        content.put("offset", data);
+
+        snapShot.setValue(content.toString());
+        return snapShot;
+
+    }
+
+    public CheckPoint fromSnapShot(SourceSnapShot sourceSnapShot){
+
+        if(sourceSnapShot == null){
+            return null;
+        }
+
+        String key = sourceSnapShot.getKey();
+        String value = sourceSnapShot.getValue();
+        CheckPoint<String> checkPoint = new CheckPoint<>();
+        String[] tmp1 = CheckPointManager.parseCheckPointKey(key);
+        String[] tmp2 = tmp1[0].split(";");
+        checkPoint.setSourceNamespace(tmp2[0]);
+        checkPoint.setPipelineName(tmp2[1]);
+        checkPoint.setSourceName(tmp2[2]);
+        checkPoint.setQueueId(tmp1[1]);
+        checkPoint.setData(value);
+        checkPoint.setGmtCreate(sourceSnapShot.getGmtCreate());
+        checkPoint.setGmtModified(sourceSnapShot.getGmtModified());
+        return checkPoint;
+
+    }
+}
\ No newline at end of file
diff --git a/rocketmq-streams-commons/src/main/java/org/apache/rocketmq/streams/common/checkpoint/CheckPointManager.java b/rocketmq-streams-commons/src/main/java/org/apache/rocketmq/streams/common/checkpoint/CheckPointManager.java
index e29f7d7..f57bcc2 100644
--- a/rocketmq-streams-commons/src/main/java/org/apache/rocketmq/streams/common/checkpoint/CheckPointManager.java
+++ b/rocketmq-streams-commons/src/main/java/org/apache/rocketmq/streams/common/checkpoint/CheckPointManager.java
@@ -16,164 +16,84 @@
  */
 package org.apache.rocketmq.streams.common.checkpoint;
 
-import org.apache.rocketmq.streams.common.channel.sinkcache.IMessageCache;
-import org.apache.rocketmq.streams.common.channel.sinkcache.IMessageFlushCallBack;
-import org.apache.rocketmq.streams.common.channel.sinkcache.impl.MessageCache;
 import org.apache.rocketmq.streams.common.channel.source.ISource;
-import org.apache.rocketmq.streams.common.configurable.IConfigurableIdentification;
+import org.apache.rocketmq.streams.common.channel.split.ISplit;
+import org.apache.rocketmq.streams.common.component.ComponentCreator;
+import org.apache.rocketmq.streams.common.configurable.BasedConfigurable;
+import org.apache.rocketmq.streams.common.configure.ConfigureFileKey;
 import org.apache.rocketmq.streams.common.context.MessageOffset;
 import org.apache.rocketmq.streams.common.utils.MapKeyUtil;
+import org.apache.rocketmq.streams.common.utils.ReflectUtil;
 import org.apache.rocketmq.streams.common.utils.StringUtil;
 
-import java.util.HashMap;
-import java.util.Iterator;
-import java.util.List;
-import java.util.Map;
+import java.util.*;
 import java.util.Map.Entry;
-import java.util.Set;
 
-public class CheckPointManager {
-    protected IMessageCache<CheckPointMessage> messageCache;
+public class CheckPointManager extends BasedConfigurable {
+
     protected transient Map<String, Long> currentSplitAndLastUpdateTime = new HashMap<>();//保存这个实例处理的分片数
 
     protected transient Map<String, Long> removingSplits = new HashMap<>();//正在删除的分片
 
-    public CheckPointManager() {
-        messageCache = new MessageCache<>(new IMessageFlushCallBack<CheckPointMessage>() {
-            @Override
-            public boolean flushMessage(List<CheckPointMessage> messages) {
-                //合并最近的checkpoint,只存储一次,为了
-                Map<String, SourceState> sourceStateMap = mergeSourceState(messages);
-                saveCheckPoint(sourceStateMap);
-                return true;
-            }
-        });
-        messageCache.openAutoFlush();
-    }
+    protected transient ICheckPointStorage iCheckPointStorage;
 
-    public void flush() {
-        messageCache.flush();
+    public CheckPointManager(){
+        String name = ComponentCreator.getProperties().getProperty(ConfigureFileKey.CHECKPOINT_STORAGE_NAME);
+        iCheckPointStorage = CheckPointStorageFactory.getInstance().getStorage(name);
     }
 
-    public synchronized void addSplit(String splitId) {
-        this.currentSplitAndLastUpdateTime.put(splitId, System.currentTimeMillis());
-    }
 
-    public synchronized void removeSplit(String splitId) {
+    public synchronized void addSplit(String splitId){
+        this.currentSplitAndLastUpdateTime.put(splitId,System.currentTimeMillis());
+    }
+    public synchronized void removeSplit(String splitId){
         this.currentSplitAndLastUpdateTime.remove(splitId);
     }
 
-    public boolean contains(String splitId) {
+    public boolean contains(String splitId){
         return this.currentSplitAndLastUpdateTime.containsKey(splitId);
     }
 
-    /**
-     * 可能有多次的offset合并在一起,对offset合并 合并包含两个逻辑:1.同1个CheckPointMessage中,选择最小的作为本次的offset
-     *
-     * @param messages
-     */
-    protected Map<String, SourceState> mergeSourceState(List<CheckPointMessage> messages) {
-        Map<String, SourceState> sourceStateMap = new HashMap<>();
-        for (CheckPointMessage checkPointMessage : messages) {
-            SourceState sourceState = createSourceState(checkPointMessage);
-            if (sourceState == null) {
-                continue;
-            }
-            String sourceName = sourceState.getSourceName();
-            SourceState existSourceState = sourceStateMap.get(sourceName);
-            if (existSourceState != null) {
-                SourceState lastSourceState = merge(sourceState, existSourceState);
-                sourceStateMap.put(sourceName, lastSourceState);
+    private final List<CheckPoint> fromSourceState(Map<String, SourceState> sourceStateMap){
+
+        List<CheckPoint> checkPoints = new ArrayList<>();
+
+        for(Entry<String, SourceState> entry : sourceStateMap.entrySet()){
+            String key = entry.getKey();
+            SourceState value = entry.getValue();
+            String[] ss = key.split("\\;");
+            assert ss.length == 3 : "key length must be three. format is namespace;pipelineName;sourceName" + key;
+            for(Entry<String, MessageOffset> tmpEntry : value.getQueueId2Offsets().entrySet()){
+                String queueId = tmpEntry.getKey();
+                String offset = tmpEntry.getValue().getMainOffset();
+                CheckPoint checkPoint = new CheckPoint();
+                checkPoint.setSourceNamespace(ss[0]);
+                checkPoint.setPipelineName(ss[1]);
+                checkPoint.setSourceName(ss[2]);
+                checkPoint.setQueueId(queueId);
+                checkPoint.setData(offset);
+                checkPoints.add(checkPoint);
             }
         }
-        return sourceStateMap;
-    }
 
-    /**
-     * 一个pipeline流程中,找最小的offset提交保存
-     *
-     * @param checkPointMessage
-     * @return
-     */
-    protected SourceState createSourceState(CheckPointMessage checkPointMessage) {
-        SourceState sourceState = new SourceState();
-        String pipelineName = null;
-        if (checkPointMessage.getStreamOperator() instanceof IConfigurableIdentification) {
-            IConfigurableIdentification configurable = (IConfigurableIdentification)checkPointMessage.getCheckPointStates();
-            pipelineName = configurable.getConfigureName();
-        }
-        Map<String, MessageOffset> queueId2Offsets = new HashMap<>();
-        sourceState.setSourceName(createSourceName(checkPointMessage.getSource(), pipelineName));
-        sourceState.setQueueId2Offsets(queueId2Offsets);
+        return checkPoints;
 
-        for (CheckPointState checkPointState : checkPointMessage.getCheckPointStates()) {
-            if (checkPointState.isReplyAnyOny()) {
-                continue;
-            }
-            if (checkPointState.isReplyRefuse()) {
-                return null;
-            }
-            for (Entry<String, MessageOffset> entry : checkPointState.getQueueIdAndOffset().entrySet()) {
-                String queueId = entry.getKey();
-                MessageOffset offset = entry.getValue();
-                MessageOffset existOffset = queueId2Offsets.get(queueId);
-                if (existOffset == null) {
-                    queueId2Offsets.put(queueId, offset);
-                } else {
-                    boolean isGreateThan = existOffset.greateThan(offset.getOffsetStr());
-                    if (isGreateThan) {
-                        queueId2Offsets.put(queueId, offset);
-                    } else {
-                        queueId2Offsets.put(queueId, existOffset);
-                    }
-                }
-            }
-        }
-        return sourceState;
     }
 
-    /**
-     * 先查询现在数据源的分片,如果已经不处理的分片,不做保存 否则把结果保存到db中
-     *
-     * @param sourceStateMap
-     */
-    protected void saveCheckPoint(Map<String, SourceState> sourceStateMap) {
-
+    public void addCheckPointMessage(CheckPointMessage message){
+        this.iCheckPointStorage.addCheckPointMessage(message);
     }
 
-    /**
-     * 如果多次的checkpoint在一起,先合并再保存
-     *
-     * @param sourceState
-     * @param existSourceState
-     * @return
-     */
-    protected SourceState merge(SourceState sourceState, SourceState existSourceState) {
-        Iterator<Entry<String, MessageOffset>> it = sourceState.getQueueId2Offsets().entrySet()
-            .iterator();
-        while (it.hasNext()) {
-            Entry<String, MessageOffset> entry = it.next();
-            String queueId = entry.getKey();
-            MessageOffset offset = entry.getValue();
-            MessageOffset existOffset = existSourceState.getQueueId2Offsets().get(queueId);
-            if (existOffset == null) {
-                existSourceState.getQueueId2Offsets().put(queueId, offset);
-            } else {
-                boolean isGreaterThan = offset.greateThan(existOffset.getOffsetStr());
-                if (isGreaterThan) {
-                    existSourceState.getQueueId2Offsets().put(queueId, offset);
-                }
-            }
+    public CheckPoint recover(ISource iSource, ISplit iSplit){
+        String isRecover =  ComponentCreator.getProperties().getProperty(ConfigureFileKey.IS_RECOVER_MODE);
+        if(isRecover != null && Boolean.valueOf(isRecover)){
+            String queueId = iSplit.getQueueId();
+            return iCheckPointStorage.recover(iSource, queueId);
         }
-        return existSourceState;
-    }
-
-    public void addCheckPointMessage(CheckPointMessage message) {
-        this.messageCache.addCache(message);
+        return null;
     }
 
 
-
     public void updateLastUpdate(String queueId) {
         addSplit(queueId);
     }
@@ -182,54 +102,29 @@ public class CheckPointManager {
         return this.currentSplitAndLastUpdateTime.keySet();
     }
 
-    public static class SourceState {
-        protected String sourceName;
-        protected Map<String, MessageOffset> queueId2Offsets = new HashMap<>();
-
-        public String getSourceName() {
-            return sourceName;
-        }
+    public void flush(){
+        iCheckPointStorage.flush();
+    }
 
-        public void setSourceName(String sourceName) {
-            this.sourceName = sourceName;
-        }
 
-        public Map<String, MessageOffset> getQueueId2Offsets() {
-            return queueId2Offsets;
-        }
 
-        public void setQueueId2Offsets(
-            Map<String, MessageOffset> queueId2Offsets) {
-            this.queueId2Offsets = queueId2Offsets;
-        }
-    }
 
     /**
      * 根据source进行划分,主要是针对双流join的场景
-     *
      * @param source
      * @return
      */
-    public static String createSourceName(ISource source, String piplineName) {
-        String namespace = null;
-        String name = null;
-        if (source != null) {
-            namespace = source.getNameSpace();
-            name = source.getConfigureName();
-        }
+    public static String createSourceName(ISource source, String pipelineName){
 
-        if (StringUtil.isEmpty(namespace)) {
-            namespace = "default_namespace";
-        }
-        if (StringUtil.isEmpty(name)) {
-            name = "default_name";
+        if(StringUtil.isNotEmpty(pipelineName)){
+            return MapKeyUtil.createKey(source.createCheckPointName(), pipelineName);
         }
-        if (StringUtil.isEmpty(piplineName)) {
-            piplineName = "default_piplineName";
-        }
-        return MapKeyUtil.createKey(namespace, piplineName, name);
+        return source.createCheckPointName();
     }
 
+
+
+
     public Map<String, Long> getCurrentSplitAndLastUpdateTime() {
         return currentSplitAndLastUpdateTime;
     }
@@ -260,4 +155,43 @@ public class CheckPointManager {
         }
         return true;
     }
+
+    public static final String createCheckPointKey(String key, String queueId){
+        return key + "^^^" + queueId;
+    }
+
+    public static final String[] parseCheckPointKey(String checkPointKey){
+        return checkPointKey.split("\\^\\^\\^");
+    }
+
+    public static final String getNameSpaceFromCheckPointKey(String checkPointKey){
+        return parseCheckPointKey(checkPointKey)[0].split("\\;")[0];
+    }
+
+    public static final String getGroupNameFromCheckPointKey(String checkPointKey){
+        return parseCheckPointKey(checkPointKey)[0].split("\\;")[1];
+    }
+
+    public static final String getNameFromCheckPointKey(String checkPointKey){
+        return parseCheckPointKey(checkPointKey)[0].split("\\;")[2];
+    }
+
+    public static final String getTopicFromCheckPointKey(String checkPointKey){
+        return parseCheckPointKey(checkPointKey)[0].split("\\;")[3];
+    }
+
+    public static final String getQueueIdFromCheckPointKey(String checkPointKey){
+        return parseCheckPointKey(checkPointKey)[1];
+    }
+
+    public static void main(String[] args){
+        SourceSnapShot snapShot = new SourceSnapShot();
+        snapShot.setId(1L);
+        snapShot.setGmtCreate(new Date());
+        snapShot.setGmtModified(new Date());
+        snapShot.setKey("key");
+        snapShot.setValue("value");
+        System.out.println(ReflectUtil.serializeObject(snapShot));
+
+    }
 }
diff --git a/rocketmq-streams-commons/src/main/java/org/apache/rocketmq/streams/common/checkpoint/CheckPointMessage.java b/rocketmq-streams-commons/src/main/java/org/apache/rocketmq/streams/common/checkpoint/CheckPointMessage.java
index 96c510c..ee91d9c 100644
--- a/rocketmq-streams-commons/src/main/java/org/apache/rocketmq/streams/common/checkpoint/CheckPointMessage.java
+++ b/rocketmq-streams-commons/src/main/java/org/apache/rocketmq/streams/common/checkpoint/CheckPointMessage.java
@@ -27,6 +27,8 @@ public class CheckPointMessage implements ISystemMessage {
     protected IStreamOperator streamOperator;//当前的pipline
     protected List<CheckPointState> checkPointStates = new ArrayList<>();
     protected boolean isValidate = true;
+    protected String pipelineName;
+
 
     public ISource getSource() {
         return source;
@@ -68,4 +70,12 @@ public class CheckPointMessage implements ISystemMessage {
     public boolean isValidate() {
         return isValidate;
     }
+
+    public String getPipelineName() {
+        return pipelineName;
+    }
+
+    public void setPipelineName(String pipelineName) {
+        this.pipelineName = pipelineName;
+    }
 }
diff --git a/rocketmq-streams-commons/src/main/java/org/apache/rocketmq/streams/common/checkpoint/CheckPointStorageFactory.java b/rocketmq-streams-commons/src/main/java/org/apache/rocketmq/streams/common/checkpoint/CheckPointStorageFactory.java
new file mode 100644
index 0000000..747b5be
--- /dev/null
+++ b/rocketmq-streams-commons/src/main/java/org/apache/rocketmq/streams/common/checkpoint/CheckPointStorageFactory.java
@@ -0,0 +1,87 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.rocketmq.streams.common.checkpoint;
+
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+
+import java.net.URL;
+import java.net.URLClassLoader;
+import java.util.Iterator;
+import java.util.ServiceLoader;
+
+/**
+ * @description
+ */
+public class CheckPointStorageFactory {
+
+    private static final Log logger = LogFactory.getLog(CheckPointStorageFactory.class);
+
+    public static final String DEFAULT_CHECKPOINT_TYPE_NAME = "DB";
+
+    private static CheckPointStorageFactory instance;
+
+    private ServiceLoader<ICheckPointStorage> loader;
+
+    private CheckPointStorageFactory() {
+        URLClassLoader classLoader = (URLClassLoader)Thread.currentThread().getContextClassLoader();
+        URL[] urls = classLoader.getURLs();
+        for(URL u : urls){
+            String s = u.toString();
+            if(s.contains("rocketmq-streams")){
+                logger.info(String.format("list class : %s", s));
+            }
+        }
+        loader = ServiceLoader.load(ICheckPointStorage.class);
+    }
+
+    public static CheckPointStorageFactory getInstance() {
+        if (null == instance) {
+            synchronized (CheckPointStorageFactory.class) {
+                if (null == instance) {
+                    instance = new CheckPointStorageFactory();
+                }
+            }
+        }
+        return instance;
+    }
+
+    public ICheckPointStorage getStorage(String name){
+
+        Iterator<ICheckPointStorage> it = loader.iterator();
+        ICheckPointStorage storage = null;
+
+        ICheckPointStorage defaultStorage = null;
+        while(it.hasNext()){
+            ICheckPointStorage local = it.next();
+            if(local.getStorageName().equalsIgnoreCase(name)){
+                return local;
+            }
+            if(local.getStorageName().equalsIgnoreCase(DEFAULT_CHECKPOINT_TYPE_NAME)){
+                defaultStorage = local;
+            }
+        }
+
+        if(storage == null){
+            logger.error(String.format("checkpoint storage name config error, name is %s. use default checkpoint type db.", name));
+            return defaultStorage;
+        }
+        return null;
+    }
+
+}
diff --git a/rocketmq-streams-commons/src/main/java/org/apache/rocketmq/streams/common/functions/FlatMapFunction.java b/rocketmq-streams-commons/src/main/java/org/apache/rocketmq/streams/common/checkpoint/ICheckPointStorage.java
similarity index 63%
copy from rocketmq-streams-commons/src/main/java/org/apache/rocketmq/streams/common/functions/FlatMapFunction.java
copy to rocketmq-streams-commons/src/main/java/org/apache/rocketmq/streams/common/checkpoint/ICheckPointStorage.java
index 5319f41..bf60fff 100644
--- a/rocketmq-streams-commons/src/main/java/org/apache/rocketmq/streams/common/functions/FlatMapFunction.java
+++ b/rocketmq-streams-commons/src/main/java/org/apache/rocketmq/streams/common/checkpoint/ICheckPointStorage.java
@@ -14,12 +14,28 @@
  * See the License for the specific language governing permissions and
  * limitations under the License.
  */
-package org.apache.rocketmq.streams.common.functions;
+package org.apache.rocketmq.streams.common.checkpoint;
+
+
+import org.apache.rocketmq.streams.common.channel.source.ISource;
 
-import java.io.Serializable;
 import java.util.List;
 
-public interface FlatMapFunction <T, O> extends Function{
+/**
+ * @description 负责checkpoint的保存、恢复
+ */
+public interface ICheckPointStorage {
+
+    String TYPE = "checkpoint_storage";
+
+    String getStorageName();
+
+    <T> void save(List<T> checkPointState);
+
+    <T> T recover(ISource iSource, String queueID);
+
+    void flush();
+
+    void addCheckPointMessage(CheckPointMessage message);
 
-    List<T> flatMap(O message) throws Exception;
 }
diff --git a/rocketmq-streams-commons/src/main/java/org/apache/rocketmq/streams/common/checkpoint/SourceSnapShot.java b/rocketmq-streams-commons/src/main/java/org/apache/rocketmq/streams/common/checkpoint/SourceSnapShot.java
new file mode 100644
index 0000000..43293d1
--- /dev/null
+++ b/rocketmq-streams-commons/src/main/java/org/apache/rocketmq/streams/common/checkpoint/SourceSnapShot.java
@@ -0,0 +1,79 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.rocketmq.streams.common.checkpoint;
+
+import com.alibaba.fastjson.JSONObject;
+import org.apache.rocketmq.streams.common.metadata.MetaData;
+import org.apache.rocketmq.streams.common.model.Entity;
+
+import java.io.Serializable;
+
+/**
+ * @create 2021-08-06 16:21:30
+ * @description
+ */
+public class SourceSnapShot extends Entity implements Serializable {
+
+    private static final long serialVersionUID = 4449170945607357658L;
+
+    public final static MetaData snapshotTable = new MetaData();
+
+    static{
+        snapshotTable.setTableName("checkpoint_snapshot");
+        snapshotTable.addMetaDataField("id", "long", false);
+        snapshotTable.addMetaDataField("gmt_create", "DATE", false);
+        snapshotTable.addMetaDataField("gmt_modified", "DATE", false);
+        snapshotTable.addMetaDataField("key", "string", false);
+        snapshotTable.addMetaDataField("value", "string", false);
+    }
+
+    String key;
+
+    String value;
+
+    public String getKey() {
+        return key;
+    }
+
+    public void setKey(String key) {
+        this.key = key;
+    }
+
+    public String getValue() {
+        return value;
+    }
+
+    public void setValue(String value) {
+        this.value = value;
+    }
+
+
+    @Override
+    public String toString() {
+        return "CheckPointSnapShot{" +
+            "key='" + key + '\'' +
+            ", value='" + value + '\'' +
+            '}';
+    }
+
+    public JSONObject toJson(){
+        JSONObject object = new JSONObject();
+        object.put("key", key);
+        object.put("value", value);
+        return object;
+    }
+}
diff --git a/rocketmq-streams-commons/src/main/java/org/apache/rocketmq/streams/common/checkpoint/SourceState.java b/rocketmq-streams-commons/src/main/java/org/apache/rocketmq/streams/common/checkpoint/SourceState.java
new file mode 100644
index 0000000..a239fdc
--- /dev/null
+++ b/rocketmq-streams-commons/src/main/java/org/apache/rocketmq/streams/common/checkpoint/SourceState.java
@@ -0,0 +1,57 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.rocketmq.streams.common.checkpoint;
+
+import org.apache.rocketmq.streams.common.context.MessageOffset;
+
+import java.util.HashMap;
+import java.util.Map;
+
+/**
+ * @create 2021-08-11 15:51:50
+ * @description
+ */
+public class SourceState {
+
+    protected String sourceName;
+    protected Map<String, MessageOffset> queueId2Offsets = new HashMap<>();
+
+    public String getSourceName() {
+        return sourceName;
+    }
+
+    public void setSourceName(String sourceName) {
+        this.sourceName = sourceName;
+    }
+
+    public Map<String, MessageOffset> getQueueId2Offsets() {
+        return queueId2Offsets;
+    }
+
+    public void setQueueId2Offsets(
+        Map<String, MessageOffset> queueId2Offsets) {
+        this.queueId2Offsets = queueId2Offsets;
+    }
+
+    @Override
+    public String toString() {
+        return "SourceState{" +
+            "sourceName='" + sourceName + '\'' +
+            ", queueId2Offsets=" + queueId2Offsets +
+            '}';
+    }
+}
diff --git a/rocketmq-streams-commons/src/main/java/org/apache/rocketmq/streams/common/configure/ConfigureFileKey.java b/rocketmq-streams-commons/src/main/java/org/apache/rocketmq/streams/common/configure/ConfigureFileKey.java
index cbdb3e4..f060a7a 100644
--- a/rocketmq-streams-commons/src/main/java/org/apache/rocketmq/streams/common/configure/ConfigureFileKey.java
+++ b/rocketmq-streams-commons/src/main/java/org/apache/rocketmq/streams/common/configure/ConfigureFileKey.java
@@ -140,4 +140,10 @@ public interface ConfigureFileKey {
 
     String LEASE_STORAGE_NAME = "DB";//通过这个配置,可以修改lease 的底层存储
 
+    String CHECKPOINT_STORAGE_NAME = "checkPointStorageName";
+
+    String IS_RECOVER_MODE = "isRecover";
+
+    String IS_ATOMIC_DB_SINK = "isAtomicDbSink";
+
 }
diff --git a/rocketmq-streams-commons/src/main/java/org/apache/rocketmq/streams/common/functions/FlatMapFunction.java b/rocketmq-streams-commons/src/main/java/org/apache/rocketmq/streams/common/functions/FlatMapFunction.java
index 5319f41..87802b9 100644
--- a/rocketmq-streams-commons/src/main/java/org/apache/rocketmq/streams/common/functions/FlatMapFunction.java
+++ b/rocketmq-streams-commons/src/main/java/org/apache/rocketmq/streams/common/functions/FlatMapFunction.java
@@ -19,7 +19,7 @@ package org.apache.rocketmq.streams.common.functions;
 import java.io.Serializable;
 import java.util.List;
 
-public interface FlatMapFunction <T, O> extends Function{
+public interface FlatMapFunction <OUT, IN> extends Function, Serializable {
 
-    List<T> flatMap(O message) throws Exception;
+    List<OUT> flatMap(IN message) throws Exception;
 }
diff --git a/rocketmq-streams-commons/src/main/java/org/apache/rocketmq/streams/common/optimization/cachefilter/CacheFilterMeta.java b/rocketmq-streams-commons/src/main/java/org/apache/rocketmq/streams/common/optimization/cachefilter/CacheFilterMeta.java
index ab4612b..bacb7b8 100644
--- a/rocketmq-streams-commons/src/main/java/org/apache/rocketmq/streams/common/optimization/cachefilter/CacheFilterMeta.java
+++ b/rocketmq-streams-commons/src/main/java/org/apache/rocketmq/streams/common/optimization/cachefilter/CacheFilterMeta.java
@@ -1,6 +1,20 @@
 package org.apache.rocketmq.streams.common.optimization.cachefilter;
-
-import java.util.BitSet;
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
 import java.util.Map;
 import org.apache.rocketmq.streams.common.cache.compress.BitSetCache;
 import org.apache.rocketmq.streams.common.context.AbstractContext;
diff --git a/rocketmq-streams-configurable/pom.xml b/rocketmq-streams-configurable/pom.xml
index f160060..baacd27 100755
--- a/rocketmq-streams-configurable/pom.xml
+++ b/rocketmq-streams-configurable/pom.xml
@@ -1,4 +1,5 @@
 <?xml version="1.0" encoding="utf-8"?>
+
 <!--
   Licensed to the Apache Software Foundation (ASF) under one or more
   contributor license agreements.  See the NOTICE file distributed with
diff --git a/rocketmq-streams-state/pom.xml b/rocketmq-streams-state/pom.xml
old mode 100755
new mode 100644

Mime
View raw message