apache · duhenglucky · Sep 2, 2021 · Sep 1, 2021 · Sep 1, 2021 · Sep 2, 2021
diff --git a/...mq-streams-channel-rocketmq/src/main/java/org/apache/rocketmq/streams/RocketMQOffset.java b/...mq-streams-channel-rocketmq/src/main/java/org/apache/rocketmq/streams/RocketMQOffset.java
@@ -20,6 +20,7 @@
 import java.util.HashSet;
 import java.util.Map;
 import java.util.Set;
+import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.concurrent.ConcurrentMap;
 import java.util.concurrent.atomic.AtomicLong;
 import org.apache.rocketmq.client.consumer.store.OffsetStore;
@@ -36,9 +37,12 @@
 public class RocketMQOffset implements OffsetStore {
     protected OffsetStore offsetStore;
     protected AbstractSupportOffsetResetSource source;
+    private AtomicBoolean starting;
+
     public RocketMQOffset(OffsetStore offsetStore, AbstractSupportOffsetResetSource source){
         this.offsetStore=offsetStore;
         this.source=source;
+        this.starting = new AtomicBoolean(true);
     }
     @Override
     public void load() throws MQClientException {
@@ -67,9 +71,16 @@ public void persist(MessageQueue mq) {
 
     @Override
     public void removeOffset(MessageQueue mq) {
-        Set<String> splitIds = new HashSet<>();
-        splitIds.add(new RocketMQMessageQueue(mq).getQueueId());
-        source.removeSplit(splitIds);
+        //todo 启动时第一次做rebalance时source中也没有原有消费mq，不做移除，做了会有副作用
+        //后续整个checkpoint机制都会调整成异步，整块代码都不会保留，目前为了整体跑通，不做修改。
+        if (starting.get()) {
+            starting.set(false);
+        } else {
+            Set<String> splitIds = new HashSet<>();
+            splitIds.add(new RocketMQMessageQueue(mq).getQueueId());
+            source.removeSplit(splitIds);
+        }
+
         offsetStore.removeOffset(mq);
     }
 

diff --git a/...ams-channel-rocketmq/src/main/java/org/apache/rocketmq/streams/source/RocketMQSource.java b/...ams-channel-rocketmq/src/main/java/org/apache/rocketmq/streams/source/RocketMQSource.java
diff --git a/...treams-clients/src/main/java/org/apache/rocketmq/streams/client/transform/DataStream.java b/...treams-clients/src/main/java/org/apache/rocketmq/streams/client/transform/DataStream.java
@@ -19,6 +19,8 @@
 
 import com.alibaba.fastjson.JSONObject;
 import com.google.common.collect.Sets;
+
+import java.nio.charset.StandardCharsets;
 import java.util.ArrayList;
 import java.util.List;
 import org.apache.rocketmq.streams.client.DataStreamAction;
@@ -131,7 +133,7 @@ protected <T> T operate(IMessage message, AbstractContext context) {
                     for(T t:result){
                         Message subMessage=null;
                         if (result instanceof JSONObject) {
-                            subMessage=new Message((JSONObject)result);
+                            subMessage=new Message((JSONObject)t);
                         } else {
                             subMessage=new Message(new UserDefinedMessage(result));
                         }

diff --git a/...mmons/src/main/java/org/apache/rocketmq/streams/common/channel/source/AbstractSource.java b/...mmons/src/main/java/org/apache/rocketmq/streams/common/channel/source/AbstractSource.java
@@ -18,13 +18,15 @@
 
 import com.alibaba.fastjson.JSONArray;
 import com.alibaba.fastjson.JSONObject;
+
 import java.util.HashSet;
 import java.util.Iterator;
 import java.util.List;
 import java.util.Map;
 import java.util.Map.Entry;
 import java.util.Set;
 import java.util.concurrent.atomic.AtomicBoolean;
+
 import org.apache.rocketmq.streams.common.channel.source.systemmsg.NewSplitMessage;
 import org.apache.rocketmq.streams.common.channel.source.systemmsg.RemoveSplitMessage;
 import org.apache.rocketmq.streams.common.channel.split.ISplit;
@@ -219,22 +221,18 @@ public JSONObject create(String message) {
         return createJson(message);
     }
 
-    /**
-     * 交给receiver执行后续逻辑
-     *
-     * @param channelMessage
-     * @return
-     */
+
     public AbstractContext executeMessage(Message channelMessage) {
         AbstractContext context = new Context(channelMessage);
-        if (isSplitInRemoving(channelMessage)) {
-            return context;
-        }
         if (!channelMessage.getHeader().isSystemMessage()) {
             messageQueueChangedCheck(channelMessage.getHeader());
         }
 
-        boolean needFlush = channelMessage.getHeader().isSystemMessage() == false && channelMessage.getHeader().isNeedFlush();
+        if (isSplitInRemoving(channelMessage)) {
+            return context;
+        }
+
+        boolean needFlush = !channelMessage.getHeader().isSystemMessage() && channelMessage.getHeader().isNeedFlush();
 
         if (receiver != null) {
             receiver.doMessage(channelMessage, context);
@@ -277,9 +275,6 @@ protected boolean isSplitInRemoving(Message channelMessage) {
      * @param header
      */
     protected void messageQueueChangedCheck(MessageHeader header) {
-        if (supportNewSplitFind() && supportRemoveSplitFind()) {
-            return;
-        }
         Set<String> queueIds = new HashSet<>();
         String msgQueueId = header.getQueueId();
         if (StringUtil.isNotEmpty(msgQueueId)) {
@@ -290,7 +285,7 @@ protected void messageQueueChangedCheck(MessageHeader header) {
             queueIds.addAll(checkpointQueueIds);
         }
         Set<String> newQueueIds = new HashSet<>();
-        Set<String> removeQueueIds = new HashSet<>();
+
         for (String queueId : queueIds) {
             if (isNotDataSplit(queueId)) {
                 continue;
@@ -306,22 +301,12 @@ protected void messageQueueChangedCheck(MessageHeader header) {
                 } else {
 
                     this.checkPointManager.updateLastUpdate(queueId);
-                    //if(this.checkPointManager.isRemovedSplit(queueId)){
-                    //    this.checkPointManager.removeSplit(queueId);
-                    //    removeQueueIds.add(queueId);
-                    //}else {
-                    //
-                    //}
+
                 }
             }
         }
-        //if(!supportRemoveSplitFind()){
-        //    removeSplit(removeQueueIds);
-        //}
-        if (!supportNewSplitFind()) {
-            addNewSplit(newQueueIds);
-        }
 
+        addNewSplit(newQueueIds);
     }
 
     protected abstract boolean isNotDataSplit(String queueId);
@@ -343,12 +328,15 @@ public void removeSplit(Set<String> splitIds) {
 
         }
     }
-    public List<ISplit> getAllSplits(){
+
+    public List<ISplit> getAllSplits() {
         return null;
     }
-    public Map<String,List<ISplit>> getWorkingSplitsGroupByInstances(){
+
+    public Map<String, List<ISplit>> getWorkingSplitsGroupByInstances() {
         return null;
     }
+
     public void addNewSplit(Set<String> splitIds) {
         if (splitIds == null || splitIds.size() == 0) {
             return;

diff --git a/...ommons/src/main/java/org/apache/rocketmq/streams/common/checkpoint/CheckPointManager.java b/...ommons/src/main/java/org/apache/rocketmq/streams/common/checkpoint/CheckPointManager.java
@@ -16,13 +16,6 @@
  */
 package org.apache.rocketmq.streams.common.checkpoint;
 
-import java.util.ArrayList;
-import java.util.HashMap;
-import java.util.Iterator;
-import java.util.List;
-import java.util.Map;
-import java.util.Map.Entry;
-import java.util.Set;
 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;
@@ -32,6 +25,13 @@
 import org.apache.rocketmq.streams.common.utils.MapKeyUtil;
 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.Map.Entry;
+import java.util.Set;
+
 public class CheckPointManager {
     protected IMessageCache<CheckPointMessage> messageCache;
     protected transient Map<String, Long> currentSplitAndLastUpdateTime = new HashMap<>();//保存这个实例处理的分片数
@@ -138,11 +138,7 @@ protected SourceState createSourceState(CheckPointMessage checkPointMessage) {
      * @param sourceStateMap
      */
     protected void saveCheckPoint(Map<String, SourceState> sourceStateMap) {
-        List<CheckPoint> checkPoints = new ArrayList<>();
-        for (SourceState sourceState : sourceStateMap.values()) {
-            CheckPoint checkPoint = new CheckPoint();
-            //  checkPoint.setOffset();
-        }
+
     }
 
     /**
@@ -176,16 +172,7 @@ public void addCheckPointMessage(CheckPointMessage message) {
         this.messageCache.addCache(message);
     }
 
-    //public boolean isRemovedSplit(String queueId) {
-    //    Long lastUpdateTime=this.currentSplitAndLastUpdateTime.get(queueId);
-    //    if(lastUpdateTime==null){
-    //        return false;
-    //    }
-    //    if(System.currentTimeMillis()-lastUpdateTime>10000*1000){
-    //        return true;
-    //    }
-    //    return false;
-    //}
+
 
     public void updateLastUpdate(String queueId) {
         addSplit(queueId);

diff --git a/...-commons/src/main/java/org/apache/rocketmq/streams/common/configurable/IConfigurable.java b/...-commons/src/main/java/org/apache/rocketmq/streams/common/configurable/IConfigurable.java
@@ -28,12 +28,12 @@ public interface IConfigurable extends IJsonable, IConfigurableIdentification, S
     /**
      * 把toJson的结果当作一个特殊属性
      */
-    static final String JSON_PROPERTY = "configurable_json";
+    String JSON_PROPERTY = "configurable_json";
 
     /**
      * 把status当作configurable 的一个特殊属性
      */
-    static final String STATUS_PROPERTY = "configurable_status";
+    String STATUS_PROPERTY = "configurable_status";
 
     /**
      * 每个配置有一个独立的名字

diff --git a/...ams-commons/src/main/java/org/apache/rocketmq/streams/common/context/AbstractContext.java b/...ams-commons/src/main/java/org/apache/rocketmq/streams/common/context/AbstractContext.java
@@ -17,6 +17,7 @@
 package org.apache.rocketmq.streams.common.context;
 
 import java.util.ArrayList;
+import java.util.Arrays;
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
@@ -128,9 +129,7 @@ public void addSplitMessages(T... splitMessages) {
         if (splitMessages == null) {
             return;
         }
-        for (T t : splitMessages) {
-            this.splitMessages.add(t);
-        }
+        this.splitMessages.addAll(Arrays.asList(splitMessages));
     }
 
     public void removeSpliteMessage(T message) {

diff --git a/...reams-commons/src/main/java/org/apache/rocketmq/streams/common/functions/MapFunction.java b/...reams-commons/src/main/java/org/apache/rocketmq/streams/common/functions/MapFunction.java
@@ -16,7 +16,9 @@
  */
 package org.apache.rocketmq.streams.common.functions;
 
-public interface MapFunction<T, O> extends Function {
+import java.io.Serializable;
+
+public interface MapFunction<T, O> extends Function, Serializable {
 
     T map(O message) throws Exception;
 }
diff --git a/...ns/src/main/java/org/apache/rocketmq/streams/common/topology/builder/PipelineBuilder.java b/...ns/src/main/java/org/apache/rocketmq/streams/common/topology/builder/PipelineBuilder.java
@@ -16,7 +16,6 @@
  */
 package org.apache.rocketmq.streams.common.topology.builder;
 
-import java.io.Serializable;
 import org.apache.rocketmq.streams.common.channel.sink.ISink;
 import org.apache.rocketmq.streams.common.channel.source.ISource;
 import org.apache.rocketmq.streams.common.configurable.AbstractConfigurable;
@@ -30,13 +29,15 @@
 import org.apache.rocketmq.streams.common.utils.NameCreatorUtil;
 import org.apache.rocketmq.streams.common.utils.StringUtil;
 
+import java.io.Serializable;
 import java.util.ArrayList;
 import java.util.Collection;
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
 
 public class PipelineBuilder implements Serializable {
+    private static final long serialVersionUID = 1L;
 
     /**
      * 最终产出的pipeline

diff --git a/...ommons/src/main/java/org/apache/rocketmq/streams/common/topology/model/AbstractStage.java b/...ommons/src/main/java/org/apache/rocketmq/streams/common/topology/model/AbstractStage.java
@@ -24,6 +24,7 @@
 import org.apache.rocketmq.streams.common.configurable.BasedConfigurable;
 import org.apache.rocketmq.streams.common.context.AbstractContext;
 import org.apache.rocketmq.streams.common.context.IMessage;
+import org.apache.rocketmq.streams.common.context.Message;
 import org.apache.rocketmq.streams.common.interfaces.IStreamOperator;
 import org.apache.rocketmq.streams.common.interfaces.ISystemMessageProcessor;
 import org.apache.rocketmq.streams.common.optimization.SQLLogFingerprintFilter;
@@ -33,7 +34,7 @@
 public abstract class AbstractStage<T extends IMessage> extends BasedConfigurable
     implements IStreamOperator<T, T>, ISystemMessageProcessor {
 
-    private static final long serialVersionUID = -143202547707927632L;
+
 
     private static final Log LOG = LogFactory.getLog(AbstractStage.class);
 

diff --git a/...ns/src/main/java/org/apache/rocketmq/streams/common/topology/stages/udf/StageBuilder.java b/...ns/src/main/java/org/apache/rocketmq/streams/common/topology/stages/udf/StageBuilder.java
@@ -26,11 +26,13 @@
 import org.apache.rocketmq.streams.common.topology.model.IStageHandle;
 import org.apache.rocketmq.streams.common.topology.stages.AbstractStatelessChainStage;
 
+import java.io.Serializable;
+
 /**
  * 给用户提供自定义的抽象类
  */
-public abstract class StageBuilder extends AbstractStatelessChainStage<IMessage> implements IStageBuilder<ChainStage>, IAfterConfigurableRefreshListener {
-
+public abstract class StageBuilder extends AbstractStatelessChainStage<IMessage> implements IStageBuilder<ChainStage>, Serializable, IAfterConfigurableRefreshListener {
+    private static final long serialVersionUID = 1L;
     @Override
     protected boolean initConfigurable() {
         return true;

diff --git a/...main/java/org/apache/rocketmq/streams/examples/rocketmqsource/RocketMQSourceExample2.java b/...main/java/org/apache/rocketmq/streams/examples/rocketmqsource/RocketMQSourceExample2.java
@@ -22,19 +22,18 @@
 
 public class RocketMQSourceExample2 {
     public static final String NAMESRV_ADDRESS = "127.0.0.1:9876";
-    public static final String RMQ_TOPIC = "topic_tiger_0822_01";
-    public static final String RMQ_CONSUMER_GROUP_NAME = "consumer_tiger_0822_01";
+    public static final String RMQ_TOPIC = "topic_tiger_0901_01";
+    public static final String RMQ_CONSUMER_GROUP_NAME = "test-group-10";
     public static final String TAGS = "*";
 
     public static void main(String[] args) {
         DataStreamSource source = StreamBuilder.dataStream("namespace", "pipeline");
 
-        source.from(new RocketMQSource(
+        source.fromRocketmq(
                 RMQ_TOPIC,
-                TAGS,
                 RMQ_CONSUMER_GROUP_NAME,
-                NAMESRV_ADDRESS
-        ))
+                false,
+                NAMESRV_ADDRESS)
                 .map(message -> message)
                 .toPrint(1)
                 .start();

diff --git a/...eams-filter/src/test/java/org/apache/rocketmq/streams/filter/operator/ExpressionTest.java b/...eams-filter/src/test/java/org/apache/rocketmq/streams/filter/operator/ExpressionTest.java
@@ -41,8 +41,7 @@ public void testRelationExpression() {
         jsonObject.put("vmip", "1.1.1.1");
 
         boolean value =
-            ExpressionBuilder.executeExecute("namespace", "(ip,=,1.2.2.3)&((uid,=,1224)|(vmip,=,1.1.11.1))",
-                jsonObject);
+            ExpressionBuilder.executeExecute("namespace", "(ip,=,1.2.2.3)&((uid,=,1224)|(vmip,=,1.1.11.1))", jsonObject);
         assertTrue(value);
     }
 }
diff --git a/...script/src/main/java/org/apache/rocketmq/streams/script/operator/impl/FunctionScript.java b/...script/src/main/java/org/apache/rocketmq/streams/script/operator/impl/FunctionScript.java
@@ -89,7 +89,7 @@ protected boolean initConfigurable() {
             if (scriptOptimization.supportOptimize()) {
                 expressions = scriptOptimization.getScriptOptimizeExprssions();
             }
-            FunctionScript functionScript = this;
+
             //转化成istreamoperator 接口
             for (IScriptExpression scriptExpression : expressions) {
                 receivers.add((message, context) -> {