Merge pull request apache#93 from cyril68/version1.0.2

add dim&script&filter&kafka

pom.xml

@@ -310,6 +310,11 @@
                 <artifactId>rocketmq-streams-channel-rocketmq</artifactId>
                 <version>${project.version}</version>
             </dependency>
+            <dependency>
+                <groupId>org.apache.rocketmq</groupId>
+                <artifactId>rocketmq-streams-channel-kafka</artifactId>
+                <version>${project.version}</version>
+            </dependency>
             <dependency>
                 <groupId>org.apache.rocketmq</groupId>
                 <artifactId>rocketmq-streams-connectors</artifactId>

rocketmq-streams-channel-es/pom.xml

@@ -1,5 +1,20 @@
 <?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
+  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.
+  -->
 <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">

rocketmq-streams-channel-es/src/test/java/org/apache/rocketmq/streams/db/sink/es/EsChannelTest.java

@@ -46,8 +46,8 @@ private ISink createChannel() {
         esChannel.setPort(port + "");
         esChannel.setEsIndex("es_index_test");
         esChannel.setNeedAuth(true);
-        esChannel.setAuthUsername("elastic");
-        esChannel.setAuthPassword("Admin123");
+        esChannel.setAuthUsername("*");
+        esChannel.setAuthPassword("*");
         esChannel.init();
         return esChannel;
     }

rocketmq-streams-channel-kafka/pom.xml

@@ -0,0 +1,48 @@
+<?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
+  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.
+  -->
+<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">
+    <parent>
+        <artifactId>rocketmq-streams</artifactId>
+        <groupId>org.apache.rocketmq</groupId>
+        <version>1.0.0-SNAPSHOT</version>
+    </parent>
+    <modelVersion>4.0.0</modelVersion>
+
+    <artifactId>rocketmq-streams-channel-kafka</artifactId>
+    <name>ROCKETMQ STREAMS :: channel-kafka</name>
+
+    <properties>
+        <maven.compiler.source>8</maven.compiler.source>
+        <maven.compiler.target>8</maven.compiler.target>
+    </properties>
+
+
+    <dependencies>
+        <dependency>
+            <groupId>org.apache.kafka</groupId>
+            <artifactId>kafka_2.12</artifactId>
+        </dependency>
+
+        <dependency>
+            <groupId>org.apache.rocketmq</groupId>
+            <artifactId>rocketmq-streams-commons</artifactId>
+        </dependency>
+    </dependencies>
+</project>

rocketmq-streams-channel-kafka/src/main/java/org/apache/rocketmq/streams/kafka/KafkaSplit.java

@@ -0,0 +1,55 @@
+/*
+ * 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.kafka;
+
+import com.alibaba.fastjson.JSONObject;
+import org.apache.kafka.common.PartitionInfo;
+import org.apache.rocketmq.streams.common.channel.split.ISplit;
+import org.apache.rocketmq.streams.common.configurable.BasedConfigurable;
+
+public class KafkaSplit extends BasedConfigurable implements ISplit<KafkaSplit, PartitionInfo> {
+    protected PartitionInfo partitionInfo;
+    protected String topic;
+    protected int partition;
+
+    @Override
+    public String getQueueId() {
+        return partition + "";
+    }
+
+
+    @Override
+    public PartitionInfo getQueue() {
+        return partitionInfo;
+    }
+
+    @Override
+    public int compareTo(KafkaSplit o) {
+        return partition - o.partition;
+    }
+
+    @Override
+    protected void getJsonObject(JSONObject jsonObject) {
+        super.getJsonObject(jsonObject);
+        this.partitionInfo = new PartitionInfo(topic, partition, null, null, null);
+    }
+
+    public KafkaSplit(PartitionInfo partitionInfo) {
+        this.partitionInfo = partitionInfo;
+        this.partition = partitionInfo.partition();
+    }
+}

rocketmq-streams-channel-kafka/src/main/java/org/apache/rocketmq/streams/kafka/KakfaChannelBuilder.java

@@ -0,0 +1,89 @@
+/*
+ * 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.kafka;
+
+import com.alibaba.fastjson.JSONObject;
+import com.google.auto.service.AutoService;
+import java.util.Properties;
+import org.apache.rocketmq.streams.common.channel.builder.AbstractSupportShuffleChannelBuilder;
+import org.apache.rocketmq.streams.common.channel.builder.IChannelBuilder;
+import org.apache.rocketmq.streams.common.channel.sink.ISink;
+import org.apache.rocketmq.streams.common.channel.source.ISource;
+import org.apache.rocketmq.streams.common.metadata.MetaData;
+import org.apache.rocketmq.streams.common.model.ServiceName;
+import org.apache.rocketmq.streams.common.utils.ConfigurableUtil;
+import org.apache.rocketmq.streams.kafka.sink.KafkaSink;
+import org.apache.rocketmq.streams.kafka.source.KafkaSource;
+
+@AutoService(IChannelBuilder.class)
+@ServiceName(value = KakfaChannelBuilder.TYPE, aliasName = "KafkaSource")
+public class KakfaChannelBuilder extends AbstractSupportShuffleChannelBuilder {
+    public static final String TYPE = "kakfa";
+    /**
+     * @param namespace
+     * @param name
+     * @param properties
+     * @return
+     */
+    @Override
+    public ISource createSource(String namespace, String name, Properties properties, MetaData metaData) {
+
+        KafkaSource kafkaSource = (KafkaSource) ConfigurableUtil.create(KafkaSource.class.getName(), namespace, name, createFormatProperty(properties), null);
+        return kafkaSource;
+    }
+
+    @Override
+    public String getType() {
+        return TYPE;
+    }
+
+    @Override
+    public ISink createSink(String namespace, String name, Properties properties, MetaData metaData) {
+        KafkaSink kafkaSink = (KafkaSink)ConfigurableUtil.create(KafkaSink.class.getName(), namespace, name, createFormatProperty(properties), null);
+        return kafkaSink;
+    }
+
+    /**
+     * 创建标准的属性文件
+     *
+     * @param properties
+     * @return
+     */
+    protected JSONObject createFormatProperty(Properties properties) {
+        JSONObject formatProperties = new JSONObject();
+        for (Object object : properties.keySet()) {
+            String key = (String)object;
+            if ("type".equals(key)) {
+                continue;
+            }
+            formatProperties.put(key, properties.getProperty(key));
+        }
+        IChannelBuilder.formatPropertiesName(formatProperties, properties, "groupName", "group.id");
+        IChannelBuilder.formatPropertiesName(formatProperties, properties, "groupName", "consumerGroup");
+        IChannelBuilder.formatPropertiesName(formatProperties, properties, "endpoint", "bootstrap.servers");
+        IChannelBuilder.formatPropertiesName(formatProperties, properties, "maxThread", "thread.max.count");
+        return formatProperties;
+    }
+
+    @Override
+    public ISink createBySource(ISource pipelineSource) {
+        KafkaSource kafkaSource = (KafkaSource)pipelineSource;
+        String topic = kafkaSource.getTopic();
+        KafkaSink sink = new KafkaSink(kafkaSource.getEndpoint(),topic);
+        return sink;
+    }
+}