Merge pull request #293 from ni-ze/develop

[ISSUE #294]Discard data because of delay

Author: ni-ze

core/src/main/java/org/apache/rocketmq/streams/core/function/supplier/JoinWindowAggregateSupplier.java

@@ -111,9 +111,7 @@ public void process(Object data) throws Throwable {
             long watermark = this.watermark(time - allowDelay, stateTopicMessageQueue);
 
             if (time < watermark) {
-                //已经触发，丢弃数据
-                logger.warn("discard data:[{}], window has been fired. maxFiredWindowEnd:{}, time of data:{}, watermark:{}",
-                        data, watermark, watermark, time);
+                logger.warn("discard delay data:[{}]. time of data:{}, watermark:{}", data, time, watermark);
                 return;
             }
             WindowInfo.JoinStream stream = (WindowInfo.JoinStream) header.get(Constant.STREAM_TAG);

core/src/main/java/org/apache/rocketmq/streams/core/function/supplier/WindowAccumulatorSupplier.java

@@ -122,9 +122,8 @@ public void process(V data) throws Throwable {
 
             long watermark = this.watermark(time - allowDelay, stateTopicMessageQueue);
             if (time < watermark) {
-                //已经触发，丢弃数据
-                logger.warn("discard data:[{}], window has been fired. time of data:{}, watermark:{}",
-                        data, time, watermark);
+                //delay data.
+                logger.warn("discard delay data:[{}]. time of data:{}, watermark:{}", data, time, watermark);
                 return;
             }
 
@@ -207,9 +206,7 @@ public void process(V data) throws Throwable {
 
             long watermark = this.watermark(time - allowDelay, stateTopicMessageQueue);
             if (time < watermark) {
-                //已经触发，丢弃数据
-                logger.warn("discard data:[{}], window has been fired. time of data:{}, watermark:{}",
-                        data, time, watermark);
+                logger.warn("discard delay data:[{}]. time of data:{}, watermark:{}", data, time, watermark);
                 return;
             }
             //本地存储里面搜索下

core/src/main/java/org/apache/rocketmq/streams/core/function/supplier/WindowAggregateSupplier.java

@@ -122,9 +122,7 @@ public void process(V data) throws Throwable {
 
             long watermark = this.watermark(time - allowDelay, stateTopicMessageQueue);
             if (time < watermark) {
-                //已经触发，丢弃数据
-                logger.warn("discard data:[{}], window has been fired. time of data:{}, watermark:{}",
-                        data, time, watermark);
+                logger.warn("discard delay data:[{}]. time of data:{}, watermark:{}", data, time, watermark);
                 return;
             }
 
@@ -212,9 +210,7 @@ public void process(V data) throws Throwable {
 
             long watermark = this.watermark(time - allowDelay, stateTopicMessageQueue);
             if (time < watermark) {
-                //已经触发，丢弃数据
-                logger.warn("discard data:[{}], window has been fired. time of data:{}, watermark:{}",
-                        data, time, watermark);
+                logger.warn("discard delay data:[{}]. time of data:{}, watermark:{}", data, time, watermark);
                 return;
             }
 

core/src/main/java/org/apache/rocketmq/streams/core/metadata/StreamConfig.java

@@ -22,6 +22,7 @@ public class StreamConfig {
     public static final String IDLE_TIME_TO_FIRE_WINDOW = "idleTimeToFireWindow";
     public static final String ROCKETMQ_STREAMS_CONSUMER_GROUP = "__source_shuffle_group";
     public static final String ROCKETMQ_STREAMS_STATE_CONSUMER_GROUP = "__state_group";
+    public static final String COMMIT_STATE_INTERNAL_MS = "commitStateIntervalMillisecond";
 
     public static Integer STREAMS_PARALLEL_THREAD_NUM = 1;
     public static Integer SHUFFLE_TOPIC_QUEUE_NUM = 8;

core/src/main/java/org/apache/rocketmq/streams/core/running/WorkerThread.java

@@ -125,8 +125,6 @@ class PlanetaryEngine<K, V> {
         private final IdleWindowScaner idleWindowScaner;
         private volatile boolean stop = false;
 
-        private long lastCommit = 0;
-        private int commitInterval = 10 * 1000;
         private final HashSet<MessageQueue> mq2Commit = new HashSet<>();
 
 
@@ -147,14 +145,15 @@ public PlanetaryEngine(DefaultLitePullConsumer unionConsumer, DefaultMQProducer
                 }
             });
             Integer idleTime = (Integer) WorkerThread.this.properties.getOrDefault(StreamConfig.IDLE_TIME_TO_FIRE_WINDOW, 2000);
+            int commitInterval = (Integer) WorkerThread.this.properties.getOrDefault(StreamConfig.COMMIT_STATE_INTERNAL_MS, 2 * 1000);
             this.idleWindowScaner = new IdleWindowScaner(idleTime, executor);
             WorkerThread.this.executor.scheduleAtFixedRate(() -> {
                 try {
                     doCommit(mq2Commit);
                 } catch (Throwable t) {
                     logger.error("commit offset and state error.", t);
                 }
-            }, 10, 10, TimeUnit.SECONDS);
+            }, 1000, commitInterval, TimeUnit.MILLISECONDS);
         }
 
 
@@ -227,15 +226,15 @@ void runInLoop() throws Throwable {
         }
 
         void doCommit(HashSet<MessageQueue> set) throws Throwable {
-            if ((System.currentTimeMillis() - lastCommit > commitInterval) && set.size() != 0) {
+            if (set != null && set.size() != 0) {
 
                 this.stateStore.persist(set);
                 this.unionConsumer.commit(set, true);
 
                 for (MessageQueue messageQueue : set) {
                     logger.debug("committed messageQueue: [{}]", messageQueue);
                 }
-                lastCommit = System.currentTimeMillis();
+
                 set.clear();
             }
         }

core/src/main/java/org/apache/rocketmq/streams/core/state/RocketMQStore.java

@@ -181,7 +181,7 @@ public void persist(Set<MessageQueue> messageQueues) throws Throwable {
             Set<byte[]> keySet = super.getInCalculating(stateTopicQueueKey);
 
             if (keySet == null || keySet.size() == 0) {
-                return;
+                continue;
             }
 
             String stateTopic = stateTopicQueue.getTopic();

example/src/main/java/org/apache/rocketmq/streams/examples/WordCount.java

@@ -37,7 +37,7 @@ public static void main(String[] args) {
                     return new Pair<>(null, value);
                 })
                 .flatMap((ValueMapperAction<String, List<String>>) value -> {
-                    String[] splits = value.toLowerCase().split("\\W+");
+                    String[] splits = value.toLowerCase().split(",");
                     return Arrays.asList(splits);
                 })
                 .keyBy(value -> value)