[FLINK-8468][RabbitMQ] Make RabbitMQ connector to take advantage of A…

…MQP features (routing key, exchange and message properties)

[FLINK-8468] Make the connector to take advantage of AMQP features (routing key, exchange and message properties) and Test

[FLINK-8468] Make the connector to take advantage of AMQP features (routing key, exchange and message properties) and Test

[FLINK-8468] Make the connector to take advantage of AMQP features (routing key, exchange and message properties) and Test

[FLINK-8468] Make the connector to take advantage of AMQP features (routing key, exchange and message properties) and Test

[FLINK-8468] Make the connector to take advantage of AMQP features (routing key, exchange and message properties) and Test

[FLINK-8468] Make the connector to take advantage of AMQP features (routing key, exchange and message properties) and Test

Fix test assertions and imports

Modified according to code reviews.

[FLINK-8468] Take advantage of AMQP features (routing key, exchange, flags, message properties and returnListener)

Update README.md

Update Readme.md

[FLINK-8468] Take advantage of AMQP features (routing key, exchange, flags, message properties and returnListener)

[FLINK-8468] Take advantage of AMQP features (routing key, exchange, flags, message properties and returnListener)

[FLINK-8468] Take advantage of AMQP features (routing key, exchange, flags, message properties and returnListener)

[FLINK-8468] Take advantage of AMQP features (routing key, exchange, flags, message properties and returnListener)

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

@@ -9,3 +9,7 @@ nor packages binaries from the "RabbitMQ AMQP Java Client".
 Users that create and publish derivative work based on Flink's
 RabbitMQ connector (thereby re-distributing the "RabbitMQ AMQP Java Client")
 must be aware that this may be subject to conditions declared in the Mozilla Public License 1.1 ("MPL"), the GNU General Public License version 2 ("GPL") and the Apache License version 2 ("ASL").
+
+# This version provides a mechanism to handle AMQ Messaging features
+
+One of its Constructor uses an implemented interface object with five methods and an optionnal returned message handler. See RMQSinkFeatureTest class to get a sample of the methods to implement. The returned message handler is an implementation of the standard com.rabbitmq.client.ReturnListener interface. As this mechasnism uses RoutingKeys, queueName is null then the queue can not be declared to RabbitMQ during start. 

flink-connectors/flink-connector-rabbitmq/src/main/java/org/apache/flink/streaming/connectors/rabbitmq/RMQSink.java

@@ -17,6 +17,7 @@
 
 package org.apache.flink.streaming.connectors.rabbitmq;
 
+import org.apache.flink.annotation.PublicEvolving;
 import org.apache.flink.api.common.serialization.SerializationSchema;
 import org.apache.flink.configuration.Configuration;
 import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;
@@ -25,6 +26,7 @@
 import com.rabbitmq.client.Channel;
 import com.rabbitmq.client.Connection;
 import com.rabbitmq.client.ConnectionFactory;
+import com.rabbitmq.client.ReturnListener;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -47,15 +49,58 @@ public class RMQSink<IN> extends RichSinkFunction<IN> {
 	protected SerializationSchema<IN> schema;
 	private boolean logFailuresOnly = false;
 
+	private final RMQSinkPublishOptions<IN> publishOptions;
+	private final ReturnListener returnListener;
+
 	/**
 	 * @param rmqConnectionConfig The RabbitMQ connection configuration {@link RMQConnectionConfig}.
 	 * @param queueName The queue to publish messages to.
 	 * @param schema A {@link SerializationSchema} for turning the Java objects received into bytes
+	 * @param publishOptions A {@link RMQSinkPublishOptions} for providing message's routing key and/or properties
+	 * @param returnListener A ReturnListener implementation object to handle returned message event
      */
-	public RMQSink(RMQConnectionConfig rmqConnectionConfig, String queueName, SerializationSchema<IN> schema) {
+	private RMQSink(RMQConnectionConfig rmqConnectionConfig, String queueName, SerializationSchema<IN> schema,
+			RMQSinkPublishOptions<IN> publishOptions, ReturnListener returnListener) {
 		this.rmqConnectionConfig = rmqConnectionConfig;
 		this.queueName = queueName;
 		this.schema = schema;
+		this.publishOptions = publishOptions;
+		this.returnListener = returnListener;
+	}
+
+	/**
+	 * @param rmqConnectionConfig The RabbitMQ connection configuration {@link RMQConnectionConfig}.
+	 * @param queueName The queue to publish messages to.
+	 * @param schema A {@link SerializationSchema} for turning the Java objects received into bytes
+     */
+	@PublicEvolving
+	public RMQSink(RMQConnectionConfig rmqConnectionConfig, String queueName, SerializationSchema<IN> schema) {
+		this(rmqConnectionConfig, queueName, schema, null, null);
+	}
+
+	/**
+	 * @param rmqConnectionConfig The RabbitMQ connection configuration {@link RMQConnectionConfig}.
+	 * @param schema A {@link SerializationSchema} for turning the Java objects received into bytes
+	 * @param publishOptions A {@link RMQSinkPublishOptions} for providing message's routing key and/or properties
+	 * In this case the computeMandatoy or computeImmediate MUST return false otherwise an
+	 * IllegalStateException is raised during runtime.
+     */
+	@PublicEvolving
+	public RMQSink(RMQConnectionConfig rmqConnectionConfig, SerializationSchema<IN> schema,
+			RMQSinkPublishOptions<IN> publishOptions) {
+		this(rmqConnectionConfig, null, schema, publishOptions, null);
+	}
+
+	/**
+	 * @param rmqConnectionConfig The RabbitMQ connection configuration {@link RMQConnectionConfig}.
+	 * @param schema A {@link SerializationSchema} for turning the Java objects received into bytes
+	 * @param publishOptions A {@link RMQSinkPublishOptions} for providing message's routing key and/or properties
+	 * @param returnListener A ReturnListener implementation object to handle returned message event
+     */
+	@PublicEvolving
+	public RMQSink(RMQConnectionConfig rmqConnectionConfig, SerializationSchema<IN> schema,
+			RMQSinkPublishOptions<IN> publishOptions, ReturnListener returnListener) {
+		this(rmqConnectionConfig, null, schema, publishOptions, returnListener);
 	}
 
 	/**
@@ -64,7 +109,9 @@ public RMQSink(RMQConnectionConfig rmqConnectionConfig, String queueName, Serial
 	 * defining custom queue parameters)
 	 */
 	protected void setupQueue() throws IOException {
-		channel.queueDeclare(queueName, false, false, false, null);
+		if (queueName != null) {
+			channel.queueDeclare(queueName, false, false, false, null);
+		}
 	}
 
 	/**
@@ -89,6 +136,9 @@ public void open(Configuration config) throws Exception {
 				throw new RuntimeException("None of RabbitMQ channels are available");
 			}
 			setupQueue();
+			if (returnListener != null) {
+				channel.addReturnListener(returnListener);
+			}
 		} catch (IOException e) {
 			throw new RuntimeException("Error while creating the channel", e);
 		}
@@ -105,7 +155,29 @@ public void invoke(IN value) {
 		try {
 			byte[] msg = schema.serialize(value);
 
-			channel.basicPublish("", queueName, null, msg);
+			if (publishOptions == null) {
+				channel.basicPublish("", queueName, null, msg);
+			} else {
+				boolean mandatory = publishOptions.computeMandatory(value);
+				boolean immediate = publishOptions.computeImmediate(value);
+
+				if (returnListener == null && (mandatory || immediate)) {
+					throw new IllegalStateException("Setting mandatory and/or immediate flags to true requires a ReturnListener.");
+				} else {
+					String rk = publishOptions.computeRoutingKey(value);
+					if (rk == null) {
+						throw new NullPointerException("computeRoutingKey returned an anormal 'null' value.");
+					}
+					String exchange = publishOptions.computeExchange(value);
+					if (exchange == null) {
+						throw new NullPointerException("computeExchange returned an anormal 'null' value.");
+					}
+
+					channel.basicPublish(exchange, rk, mandatory, immediate,
+									publishOptions.computeProperties(value), msg);
+				}
+			}
+
 		} catch (IOException e) {
 			if (logFailuresOnly) {
 				LOG.error("Cannot send RMQ message {} at {}", queueName, rmqConnectionConfig.getHost(), e);

flink-connectors/flink-connector-rabbitmq/src/main/java/org/apache/flink/streaming/connectors/rabbitmq/RMQSinkPublishOptions.java

@@ -0,0 +1,72 @@
+/*
+ * 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.rabbitmq;
+
+import org.apache.flink.annotation.PublicEvolving;
+
+import com.rabbitmq.client.AMQP.BasicProperties;
+
+/**
+ * The message computation provides methods to compute the message routing key and/or the properties.
+ *
+ * @param <IN> The type of the data used by the sink.
+ */
+@PublicEvolving
+public interface RMQSinkPublishOptions<IN> extends java.io.Serializable {
+
+	/**
+	 * Compute the message's routing key from the data.
+	 * @param a The data used by the sink
+	 * @return The routing key of the message
+	 * null will raise a NullPointerException
+	 */
+	String computeRoutingKey(IN a);
+
+	/**
+	 * Compute the message's properties from the data.
+	 * @param a The data used by the sink
+	 * @return The message's properties (can be null)
+	 */
+	BasicProperties computeProperties(IN a);
+
+	/**
+	 * Compute the exchange from the data.
+	 * @param a The data used by the sink
+	 * @return The exchange to publish the message to
+	 * null will raise a NullPointerException
+	 */
+	String computeExchange(IN a);
+
+	/**
+	 * Compute the mandatory flag used in basic.publish method
+	 * See AMQP API help for values.
+	 * A ReturnListener is mandatory if this flag can be true (if not it is ignored and forced to false)
+	 * @param a The data used by the sink
+	 * @return The mandatory flag
+	 */
+	boolean computeMandatory(IN a);
+
+	/**
+	 * Compute the immediate flag
+	 * See AMQP API help for values.
+	 * A ReturnListener is mandatory if this flag can be true (if not it is ignored and forced to false)
+	 * @param a The data used by the sink
+	 * @return The mandatory flag
+	 */
+	boolean computeImmediate(IN a);
+}