Merge pull request #38 from caigy/main

duhenglucky · web-flow · commit c8935a65de22 · 2021-08-23T21:09:17.000+08:00
[ISSUE #37]Add Apache license and fix some naming problems
diff --git a/build_without_test.sh b/build_without_test.sh
@@ -1 +1,18 @@
+#!/bin/sh
+
+# 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.
+
 mvn clean package -Dmaven.test.skip=true
diff --git a/rocketmq-streams-channel-rocketmq/src/main/java/org/apache/rocketmq/streams/sink/RocketMQSink.java b/rocketmq-streams-channel-rocketmq/src/main/java/org/apache/rocketmq/streams/sink/RocketMQSink.java
@@ -125,7 +125,7 @@ protected boolean putMessage2Mq(IMessage fieldName2Value,AtomicInteger msgFinish
     }
 
     /**
-     * 发送metaq消息
+     * 发送RocketMQ消息
      *  @param content 消息内容
      * @param key     消息的Key字段是为了唯一标识消息的，方便运维排查问题。如果不设置Key，则无法定位消息丢失原因。
      * @param targetQueue
@@ -230,9 +230,9 @@ public List<ISplit> getSplitList() {
         try {
 
             if (messageQueues == null || messageQueues.size() == 0) {
-                List<MessageQueue> metaqQueueSet = producer.fetchPublishMessageQueues(topic);
+                List<MessageQueue> rocketmqQueueSet = producer.fetchPublishMessageQueues(topic);
                 List<ISplit> queueList = new ArrayList<>();
-                for (MessageQueue queue : metaqQueueSet) {
+                for (MessageQueue queue : rocketmqQueueSet) {
                     RocketMQMessageQueue rocketMQMessageQueue = new RocketMQMessageQueue(queue);
                     queueList.add(rocketMQMessageQueue);
 
diff --git a/rocketmq-streams-channel-rocketmq/src/main/java/org/apache/rocketmq/streams/source/RocketMQSource.java b/rocketmq-streams-channel-rocketmq/src/main/java/org/apache/rocketmq/streams/source/RocketMQSource.java
@@ -143,7 +143,7 @@ protected DefaultMQPushConsumer startConsumer() {
                         i++;
                     }
                 } catch (Exception e) {
-                    LOG.error("消费metaq报错：" + e, e);
+                    LOG.error("消费rocketmq报错：" + e, e);
                 }
 
                 return ConsumeOrderlyStatus.SUCCESS;// 返回消费成功
@@ -156,21 +156,21 @@ protected DefaultMQPushConsumer startConsumer() {
         } catch (Exception e) {
             setInitSuccess(false);
             e.printStackTrace();
-            throw new RuntimeException("start metaq channel error " + topic, e);
+            throw new RuntimeException("start rocketmq channel error " + topic, e);
         }
     }
     @Override
     public List<ISplit> getAllSplits(){
         try {
-            Set<MessageQueue> metaqQueueSet =  consumer.fetchSubscribeMessageQueues(this.topic);
+            Set<MessageQueue> rocketmqQueueSet =  consumer.fetchSubscribeMessageQueues(this.topic);
             List<ISplit> queueList = new ArrayList<>();
-            for (MessageQueue queue : metaqQueueSet) {
-                RocketMQMessageQueue metaqMessageQueue = new RocketMQMessageQueue(queue);
-                if(isNotDataSplit(metaqMessageQueue.getQueueId())){
+            for (MessageQueue queue : rocketmqQueueSet) {
+                RocketMQMessageQueue rocketmqMessageQueue = new RocketMQMessageQueue(queue);
+                if(isNotDataSplit(rocketmqMessageQueue.getQueueId())){
                     continue;
                 }
 
-                queueList.add(metaqMessageQueue);
+                queueList.add(rocketmqMessageQueue);
 
             }
             return queueList;
@@ -192,8 +192,8 @@ public Map<String,List<ISplit>> getWorkingSplitsGroupByInstances(){
             Map<org.apache.rocketmq.common.message.MessageQueue, String> queue2Instances= getMessageQueueAllocationResult(defaultMQAdminExt,this.groupName);
             Map<String,List<ISplit>> instanceOwnerQueues=new HashMap<>();
             for(org.apache.rocketmq.common.message.MessageQueue messageQueue:queue2Instances.keySet()){
-                RocketMQMessageQueue metaqMessageQueue = new RocketMQMessageQueue(new MessageQueue(messageQueue.getTopic(),messageQueue.getBrokerName(),messageQueue.getQueueId()));
-                if(isNotDataSplit(metaqMessageQueue.getQueueId())){
+                RocketMQMessageQueue rocketmqMessageQueue = new RocketMQMessageQueue(new MessageQueue(messageQueue.getTopic(),messageQueue.getBrokerName(),messageQueue.getQueueId()));
+                if(isNotDataSplit(rocketmqMessageQueue.getQueueId())){
                     continue;
                 }
                 String instanceName=queue2Instances.get(messageQueue);
@@ -202,7 +202,7 @@ public Map<String,List<ISplit>> getWorkingSplitsGroupByInstances(){
                     splits=new ArrayList<>();
                     instanceOwnerQueues.put(instanceName,splits);
                 }
-                splits.add(metaqMessageQueue);
+                splits.add(rocketmqMessageQueue);
             }
             return instanceOwnerQueues;
 
diff --git a/rocketmq-streams-clients/src/main/java/org/apache/rocketmq/streams/client/strategy/ConfiguableConnector.java b/rocketmq-streams-clients/src/main/java/org/apache/rocketmq/streams/client/strategy/ConfiguableConnector.java
@@ -1,3 +1,19 @@
+/*
+ * 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.client.strategy;
 
 import org.apache.rocketmq.streams.common.configurable.IConfigurableService;
diff --git a/rocketmq-streams-clients/src/test/java/org/apache/rocketmq/streams/client/SinkTest.java b/rocketmq-streams-clients/src/test/java/org/apache/rocketmq/streams/client/SinkTest.java
@@ -1,3 +1,19 @@
+/*
+ * 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.client;
 
 import com.alibaba.fastjson.JSONObject;
diff --git a/rocketmq-streams-clients/src/test/java/org/apache/rocketmq/streams/client/SourceTest.java b/rocketmq-streams-clients/src/test/java/org/apache/rocketmq/streams/client/SourceTest.java
@@ -1,3 +1,19 @@
+/*
+ * 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.client;
 
 import org.apache.rocketmq.streams.common.channel.impl.file.FileSource;
diff --git a/rocketmq-streams-clients/src/test/java/org/apache/rocketmq/streams/client/WindowTest.java b/rocketmq-streams-clients/src/test/java/org/apache/rocketmq/streams/client/WindowTest.java
@@ -52,33 +52,4 @@ public void foreach(JSONObject o) {
 
     }
 
-    //    @Test
-    //    public void testWindowFromMetaq() throws InterruptedException {
-    //        String topic = "TOPIC_DIPPER_SYSTEM_MSG_4";
-    //        StreamBuilder.dataStream("namespace", "name")
-    //            .fromFile("/Users/yuanxiaodong/chris/sls_100.txt", true)
-    //            .toRocketmq(topic)
-    //            .asyncStart();
-    //
-    //        StreamBuilder.dataStream("namespace", "name1")
-    //            .fromRocketmq(topic, "chris", true)
-    //            .window(TumblingWindow.of(Time.seconds(5)))
-    //            .groupby("ProjectName", "LogStore")
-    //            .setLocalStorageOnly(true)
-    //            .count("total")
-    //            .sum("OutFlow", "OutFlow")
-    //            .sum("InFlow", "inflow")
-    //            .toDataSteam()
-    //            .forEach(new ForEachFunction<JSONObject>() {
-    //                protected int sum = 0;
-    //
-    //                @Override
-    //                public void foreach(JSONObject o) {
-    //                    int total = o.getInteger("total");
-    //                    sum = sum + total;
-    //                    o.put("sum(total)", sum);
-    //                }
-    //            }).toPrint().start();
-    //    }
-
 }
diff --git a/rocketmq-streams-clients/src/test/java/org/apache/rocketmq/streams/client/windows/WindowFromRocketMQ.java b/rocketmq-streams-clients/src/test/java/org/apache/rocketmq/streams/client/windows/WindowFromRocketMQ.java
@@ -20,7 +20,7 @@
 import org.apache.rocketmq.streams.client.transform.DataStream;
 import org.junit.Test;
 
-public class WindowFromMetaq extends AbstractWindowFireModeTest {
+public class WindowFromRocketMQ extends AbstractWindowFireModeTest {
 
     String topic = "TOPIC_DIPPER_SYSTEM_MSG_5";
     @Test
@@ -44,12 +44,12 @@ public void testWindowFireMode2() {
 
 
     @Test
-    public void testWindowToMetaq() throws InterruptedException {
+    public void testWindowToRocketMQ() throws InterruptedException {
 
         long start=System.currentTimeMillis();
         StreamBuilder.dataStream("namespace", "name")
             .fromFile("/Users/yuanxiaodong/chris/sls_10.txt", true)
-            .toMetaq(topic)
+                .toRocketmq(topic, "chris1", "", "", "", "", "")
             .start();
     }
 
diff --git a/rocketmq-streams-commons/src/main/java/org/apache/rocketmq/streams/common/channel/sink/AbstractSupportShuffleSink.java b/rocketmq-streams-commons/src/main/java/org/apache/rocketmq/streams/common/channel/sink/AbstractSupportShuffleSink.java
@@ -27,7 +27,7 @@ public abstract class AbstractSupportShuffleSink extends AbstractSink {
     protected volatile transient boolean hasCreated = false;
 
     /**
-     * 获取sink的主题，在sls中是logStore，metaq是topic
+     * 获取sink的主题，在sls中是logStore，RocketMQ是topic
      *
      * @return
      */
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
@@ -123,14 +123,14 @@ static String getDipperJdbcPassword() {
     }
 
     /**
-     * shuffle相关配置，如果后面加上.namespace,则只对某个namespace生效，如window.shuffle.channel.type.namespace=metaq,相当于只对这个namespace配置
+     * shuffle相关配置，如果后面加上.namespace,则只对某个namespace生效，如window.shuffle.channel.type.namespace=rocketmq,相当于只对这个namespace配置
      */
 
     String WINDOW_SHUFFLE_CHANNEL_TYPE = "window.shuffle.channel.type";//window 做shuffle中转需要的消息队列类型
-    //比如metaq，需要topic，tags和group,属性值和字段名保持一致即可。配置如下:window.shuffle.channel.topic=abdc    window.shuffle.channel.tag=fdd
+    //比如rocketmq，需要topic，tags和group,属性值和字段名保持一致即可。配置如下:window.shuffle.channel.topic=abdc    window.shuffle.channel.tag=fdd
 
     String WINDOW_SHUFFLE_CHANNEL_PROPERTY_PREFIX = "window.shuffle.channel.";
-    String WINDOW_SYSTEM_MESSAGE_CHENNEL_OWNER = "window.system.message.channel.owner";//如果能做消息过滤，只过滤本window的消息，可以配置这个属性，如metaq的tags.不支持的会做客户端过滤
+    String WINDOW_SYSTEM_MESSAGE_CHENNEL_OWNER = "window.system.message.channel.owner";//如果能做消息过滤，只过滤本window的消息，可以配置这个属性，如rocketmq的tags.不支持的会做客户端过滤
 
     /**
      * 通知相关
diff --git a/rocketmq-streams-window/src/main/java/org/apache/rocketmq/streams/window/debug/DebugWriter.java b/rocketmq-streams-window/src/main/java/org/apache/rocketmq/streams/window/debug/DebugWriter.java
@@ -1,3 +1,19 @@
+/*
+ * 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.window.debug;
 
 import com.alibaba.fastjson.JSONObject;
diff --git a/rocketmq-streams-window/src/main/java/org/apache/rocketmq/streams/window/fire/EventTimeManager.java b/rocketmq-streams-window/src/main/java/org/apache/rocketmq/streams/window/fire/EventTimeManager.java
@@ -1,3 +1,19 @@
+/*
+ * 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.window.fire;
 
 import java.util.HashMap;
diff --git a/rocketmq-streams-window/src/main/java/org/apache/rocketmq/streams/window/fire/SplitEventTimeManager.java b/rocketmq-streams-window/src/main/java/org/apache/rocketmq/streams/window/fire/SplitEventTimeManager.java
@@ -1,3 +1,19 @@
+/*
+ * 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.window.fire;
 
 import java.util.ArrayList;
diff --git a/rocketmq-streams-window/src/main/java/org/apache/rocketmq/streams/window/fire/WindowFireManager.java b/rocketmq-streams-window/src/main/java/org/apache/rocketmq/streams/window/fire/WindowFireManager.java
@@ -1,3 +1,19 @@
+/*
+ * 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.window.fire;
 
 import java.util.concurrent.ConcurrentHashMap;
diff --git a/rocketmq-streams-window/src/main/java/org/apache/rocketmq/streams/window/state/WindowBaseValue.java b/rocketmq-streams-window/src/main/java/org/apache/rocketmq/streams/window/state/WindowBaseValue.java
@@ -42,7 +42,7 @@ public class WindowBaseValue extends BasedConfigurable implements Serializable {
     protected String windowInstanceId;
 
     /**
-     * 分片信息（metaQ里是queue）
+     * 分片信息（RocketMQ里是queue）
      */
     protected String partition;
 
diff --git a/rocketmq-streams-window/src/main/resources/dipper.properties b/rocketmq-streams-window/src/main/resources/dipper.properties
@@ -8,7 +8,7 @@ dipper.rds.table.name=dipper_sql_configure
 
 #window 异步消息分发需要的消息队列类型
 dipper.window.shuffle.rocketmq.dispatch.channel.type=rocketmq
-#如果是metaq，统计和join分别用的topoc
+#如果是RocketMQ，统计和join分别用的topoc
 dipper.window.shuffle.rocketmq.dispatch.channel.topic=TOPIC_DIPPER_WINDOW_STATISTICS
 #动态生成的消息队列的属性的key值，这个值会被动态赋值
 dipper.window.shuffle.rocketmq.dispatch.channel.dynamic.property=tag

Original file line number	Diff line number	Diff line change
`@@ -27,7 +27,7 @@ public abstract class AbstractSupportShuffleSink extends AbstractSink {`
`27`	`27`	`protected volatile transient boolean hasCreated = false;`
`28`	`28`
`29`	`29`	`/**`
`30`		`- * 获取sink的主题，在sls中是logStore，metaq是topic`
	`30`	`+ * 获取sink的主题，在sls中是logStore，RocketMQ是topic`
`31`	`31`	`*`
`32`	`32`	`* @return`
`33`	`33`	`*/`