Add a new connector ' flink-connector-rocketmq'

flink-connectors/flink-connector-rocketmq/README.md

@@ -0,0 +1,51 @@
+<!--
+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.
+-->
+
+# Flink Connector for RocketMQ
+
+This module is a implementation for the message queue middleware 'RocketMQ'.The MQ now is belongs to ASF and born in Alibaba,You can refer to this module to make connection between Flink and RocketMQ cluster
+
+## Features
+* Connect RocketMQ cluster as a `DataSink` from Flink to produce message to RocketMQ cluster
+* Connect RocketMQ cluster as a `DataSource` from Flink to pull and consumer message from RocketMQ cluster
+
+## Quick Start
+
+The `Example.java` can be used to learn how to use Flink to connect RocketMQ cluster.
+
+There are the steps:
+1. Create a `StreamExecutionEnvironment` obejct likes the common starting
+2. Create a `ConsumerConfig` obejct and fill all the needed configuration
+3. Create a `RocketMQSource` as the fink source base on the previous `ConsumerConfig`
+4. Create a `ProducerConfig` and new a `RocketMQSink` as flink sink
+5. Integrate the `RocketMQSource` and `RocketMQSink` into flink DataStreamSource
+6. run `StreamExecutionEnvironment.execute()`
+
+
+## Road Map
+
+* For this first commit is the initialize version for this module,and just implement some entry level functions.and the version for rocketmq just only support the `version 3.2.6`
+* The next step will ehance the connector to support options configuration,enhance the unit test
+* Finally will support multiple rocketmq version and support ASF RocketMQ as well.
+
+## Contact
+
+Any suggestion and question are welcome.and we can improve this project together.
+* GitHub `mikiaichiyu`
+* Mail <13560102795@139.com>

flink-connectors/flink-connector-rocketmq/pom.xml

@@ -0,0 +1,77 @@
+<?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">
+
+	<modelVersion>4.0.0</modelVersion>
+
+	<parent>
+		<artifactId>flink-connectors</artifactId>
+		<groupId>org.apache.flink</groupId>
+		<version>1.10-SNAPSHOT</version>
+		<relativePath>..</relativePath>
+	</parent>
+
+
+	<artifactId>flink-connector-rocketmq_${scala.binary.version}</artifactId>
+	<name>flink-connector-rocketmq</name>
+
+	<packaging>jar</packaging>
+
+	<properties>
+		<rocketmq.version>3.2.6</rocketmq.version>
+	</properties>
+
+	<dependencies>
+
+		<dependency>
+			<groupId>org.apache.flink</groupId>
+			<artifactId>flink-streaming-java_${scala.binary.version}</artifactId>
+			<version>${project.version}</version>
+		</dependency>
+
+		<dependency>
+			<groupId>com.alibaba.rocketmq</groupId>
+			<artifactId>rocketmq-client</artifactId>
+			<version>${rocketmq.version}</version>
+		</dependency>
+
+		<dependency>
+			<groupId>org.apache.flink</groupId>
+			<artifactId>flink-streaming-java_${scala.binary.version}</artifactId>
+			<version>${project.version}</version>
+			<scope>test</scope>
+			<type>test-jar</type>
+		</dependency>
+
+		<dependency>
+			<groupId>org.apache.flink</groupId>
+			<artifactId>flink-runtime_${scala.binary.version}</artifactId>
+			<version>${project.version}</version>
+			<type>test-jar</type>
+			<scope>test</scope>
+		</dependency>
+
+	</dependencies>
+
+
+</project>

flink-connectors/flink-connector-rocketmq/src/main/java/org/apache/flink/streaming/connectors/rocketmq/RocketMQSink.java

@@ -0,0 +1,118 @@
+/*
+ * 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.flink.streaming.connectors.rocketmq;
+
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;
+import org.apache.flink.streaming.connectors.rocketmq.common.ProducerConfig;
+import org.apache.flink.streaming.connectors.rocketmq.common.RocketMQMessage;
+
+import com.alibaba.rocketmq.client.exception.MQClientException;
+import com.alibaba.rocketmq.client.producer.DefaultMQProducer;
+import com.alibaba.rocketmq.client.producer.SendResult;
+import com.alibaba.rocketmq.client.producer.SendStatus;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+
+/**
+ * A flink Data Sink for RocketMQ {@link RocketMQSink}.
+ */
+public class RocketMQSink extends RichSinkFunction<RocketMQMessage> {
+
+	private static final Logger LOG = LoggerFactory.getLogger(RocketMQSink.class);
+
+	/**
+	 * define the RocketMQ producer client here to send flink message to RocketMQ cluster.
+	 */
+	private transient DefaultMQProducer producer = null;
+
+	/**
+	 * define the ProducerConfig for producer init.
+	 */
+	private ProducerConfig producerConfig = null;
+
+	/**
+	 * Constructor for RocketMQSink,need the ProducerConfig.
+	 *
+	 * @param producerConfig
+	 * @throws MQClientException
+	 */
+	public RocketMQSink(ProducerConfig producerConfig) throws MQClientException {
+
+		if (null == this.producerConfig) {
+			LOG.error("error producerConfig,exit");
+			return;
+		} else {
+			this.producerConfig = producerConfig;
+			if (null == this.producer) {
+				producer = new DefaultMQProducer(producerConfig.getProducerGroupName());
+			}
+		}
+	}
+
+	/**
+	 * Flink sink life cycle open().
+	 *
+	 * @param parameters
+	 * @throws Exception
+	 */
+	@Override
+	public void open(Configuration parameters) throws Exception {
+
+		LOG.info("RocketMQSink open called");
+
+		if (null != producer) {
+			producer.setNamesrvAddr(producerConfig.getNameServerAddress());
+			producer.start();
+			LOG.info("RocketMQSink started,{},{}", producer.getProducerGroup(), producer.getNamesrvAddr());
+		}
+	}
+
+	/**
+	 * Flink sink life cycle invoke(),all the logical processing are here.
+	 *
+	 * @param value   The input record.
+	 * @param context Additional context about the input record.
+	 * @throws Exception
+	 */
+	@Override
+	public void invoke(RocketMQMessage value, Context context) throws Exception {
+
+		SendResult sendResult = producer.send(value);
+		if (null != sendResult && !sendResult.getSendStatus().equals(SendStatus.SEND_OK)) {
+			LOG.error("RocketMQSink send message failed, error = {}", sendResult.toString());
+		} else {
+			LOG.debug("RocketMQSink send message {}", sendResult.toString());
+		}
+	}
+
+	/**
+	 * Flink sink life cycle close().
+	 *
+	 * @throws Exception
+	 */
+	@Override
+	public void close() throws Exception {
+
+		if (producer != null) {
+			producer.shutdown();
+		}
+		LOG.info("RocketMQSink shutdown");
+	}
+}