[GitHub] flink pull request #5410: [FLINK-8468] [RabbitMQ Connector] Take advantage o...

[asfgit]

Github user asfgit closed the pull request at:

https://github.com/apache/flink/pull/5410


---

[GitHub] flink pull request #5410: [FLINK-8468] [RabbitMQ Connector] Take advantage o...

[pduveau]

Github user pduveau commented on a diff in the pull request:

https://github.com/apache/flink/pull/5410#discussion_r194239264
  
--- Diff: 
flink-connectors/flink-connector-rabbitmq/src/main/java/org/apache/flink/streaming/connectors/rabbitmq/RMQSink.java
 ---
@@ -105,7 +155,24 @@ 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);
+   String exchange = 
publishOptions.computeExchange(value);
+
+   channel.basicPublish((exchange != null) 
? exchange : "",
--- End diff --

Ok, I changed it.


---

[GitHub] flink pull request #5410: [FLINK-8468] [RabbitMQ Connector] Take advantage o...

[GJL]

Github user GJL commented on a diff in the pull request:

https://github.com/apache/flink/pull/5410#discussion_r194109528
  
--- Diff: 
flink-connectors/flink-connector-rabbitmq/src/main/java/org/apache/flink/streaming/connectors/rabbitmq/RMQSink.java
 ---
@@ -40,22 +42,63 @@
 
private static final Logger LOG = 
LoggerFactory.getLogger(RMQSink.class);
 
-   protected final String queueName;
+   private final String queueName;
private final RMQConnectionConfig rmqConnectionConfig;
protected transient Connection connection;
protected transient Channel channel;
protected SerializationSchema schema;
-   private boolean logFailuresOnly = false;
+   protected boolean logFailuresOnly = false;
+
+   private final RMQSinkPublishOptions messageCompute;
+   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 messageCompute A {@link RMQSinkPublishOptions} for providing 
message's routing key and/or properties
  */
-   public RMQSink(RMQConnectionConfig rmqConnectionConfig, String 
queueName, SerializationSchema schema) {
+   private RMQSink(RMQConnectionConfig rmqConnectionConfig, String 
queueName, SerializationSchema schema,
--- End diff --

ping


---

[GitHub] flink pull request #5410: [FLINK-8468] [RabbitMQ Connector] Take advantage o...

[GJL]

Github user GJL commented on a diff in the pull request:

https://github.com/apache/flink/pull/5410#discussion_r194109518
  
--- Diff: 
flink-connectors/flink-connector-rabbitmq/src/main/java/org/apache/flink/streaming/connectors/rabbitmq/RMQSink.java
 ---
@@ -105,7 +153,18 @@ public void invoke(IN value) {
try {
byte[] msg = schema.serialize(value);
 
-   channel.basicPublish("", queueName, null, msg);
+   if (messageCompute == null) {
+   channel.basicPublish("", queueName, null, msg);
+   } else {
+   String rk = 
messageCompute.computeRoutingKey(value);
+   String exchange = 
messageCompute.computeExchange(value);
+   channel.basicPublish((exchange != null) ? 
exchange : "",
--- End diff --

ping


---

[GitHub] flink pull request #5410: [FLINK-8468] [RabbitMQ Connector] Take advantage o...

[GJL]

Github user GJL commented on a diff in the pull request:

https://github.com/apache/flink/pull/5410#discussion_r194109540
  
--- Diff: 
flink-connectors/flink-connector-rabbitmq/src/main/java/org/apache/flink/streaming/connectors/rabbitmq/RMQSink.java
 ---
@@ -40,22 +42,63 @@
 
private static final Logger LOG = 
LoggerFactory.getLogger(RMQSink.class);
 
-   protected final String queueName;
+   private final String queueName;
private final RMQConnectionConfig rmqConnectionConfig;
protected transient Connection connection;
protected transient Channel channel;
protected SerializationSchema schema;
-   private boolean logFailuresOnly = false;
+   protected boolean logFailuresOnly = false;
+
+   private final RMQSinkPublishOptions messageCompute;
+   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 messageCompute A {@link RMQSinkPublishOptions} for providing 
message's routing key and/or properties
  */
-   public RMQSink(RMQConnectionConfig rmqConnectionConfig, String 
queueName, SerializationSchema schema) {
+   private RMQSink(RMQConnectionConfig rmqConnectionConfig, String 
queueName, SerializationSchema schema,
+   RMQSinkPublishOptions messageCompute, 
ReturnListener returnListener) {
--- End diff --

ping


---

[GitHub] flink pull request #5410: [FLINK-8468] [RabbitMQ Connector] Take advantage o...

[GJL]

Github user GJL commented on a diff in the pull request:

https://github.com/apache/flink/pull/5410#discussion_r193965626
  
--- Diff: 
flink-connectors/flink-connector-rabbitmq/src/main/java/org/apache/flink/streaming/connectors/rabbitmq/RMQSink.java
 ---
@@ -105,7 +155,24 @@ 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);
+   String exchange = 
publishOptions.computeExchange(value);
+
+   channel.basicPublish((exchange != null) 
? exchange : "",
--- End diff --

What do you think about not allowing the return of `computeRoutingKey` and 
`computeExchange` to be `null`. `basicPublish` does not allow `null`s and will 
throw an exception. This way our interfaces will be consistent with the ones 
from RMQ.


---

[GitHub] flink pull request #5410: [FLINK-8468] [RabbitMQ Connector] Take advantage o...

[GJL]

Github user GJL commented on a diff in the pull request:

https://github.com/apache/flink/pull/5410#discussion_r193964437
  
--- Diff: 
flink-connectors/flink-connector-rabbitmq/src/main/java/org/apache/flink/streaming/connectors/rabbitmq/RMQSinkPublishOptions.java
 ---
@@ -0,0 +1,70 @@
+/*
+ * 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  The type of the data used by the sink.
+ */
+@PublicEvolving
+public interface RMQSinkPublishOptions 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
+*/
+   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
+*/
+   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);
--- End diff --

ping


---

[GitHub] flink pull request #5410: [FLINK-8468] [RabbitMQ Connector] Take advantage o...

[GJL]

Github user GJL commented on a diff in the pull request:

https://github.com/apache/flink/pull/5410#discussion_r193964455
  
--- Diff: 
flink-connectors/flink-connector-rabbitmq/src/main/java/org/apache/flink/streaming/connectors/rabbitmq/RMQSinkPublishOptions.java
 ---
@@ -0,0 +1,70 @@
+/*
+ * 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  The type of the data used by the sink.
+ */
+@PublicEvolving
+public interface RMQSinkPublishOptions 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
+*/
+   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
+*/
+   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);
--- End diff --

ping


---

[GitHub] flink pull request #5410: [FLINK-8468] [RabbitMQ Connector] Take advantage o...

[GJL]

Github user GJL commented on a diff in the pull request:

https://github.com/apache/flink/pull/5410#discussion_r193964429
  
--- Diff: 
flink-connectors/flink-connector-rabbitmq/src/main/java/org/apache/flink/streaming/connectors/rabbitmq/RMQSinkPublishOptions.java
 ---
@@ -0,0 +1,70 @@
+/*
+ * 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  The type of the data used by the sink.
+ */
+@PublicEvolving
+public interface RMQSinkPublishOptions 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
+*/
+   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
+*/
+   String computeExchange(IN a);
+
+   /**
+* Compute the mandatory flag used in basic.publish method
--- End diff --

ping


---

[GitHub] flink pull request #5410: [FLINK-8468] [RabbitMQ Connector] Take advantage o...

[GJL]

Github user GJL commented on a diff in the pull request:

https://github.com/apache/flink/pull/5410#discussion_r193964402
  
--- Diff: 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
--- End diff --

ping


---

[GitHub] flink pull request #5410: [FLINK-8468] [RabbitMQ Connector] Take advantage o...

[GJL]

Github user GJL commented on a diff in the pull request:

https://github.com/apache/flink/pull/5410#discussion_r193963772
  
--- Diff: 
flink-connectors/flink-connector-rabbitmq/src/test/java/org/apache/flink/streaming/connectors/rabbitmq/RMQSinkTest.java
 ---
@@ -83,7 +103,22 @@ public void throwExceptionIfChannelIsNull() throws 
Exception {
}
 
private RMQSink createRMQSink() throws Exception {
-   RMQSink rmqSink = new RMQSink(rmqConnectionConfig, 
QUEUE_NAME, serializationSchema);
+   RMQSink rmqSink = new 
RMQSink(rmqConnectionConfig, QUEUE_NAME, serializationSchema);
+   rmqSink.open(new Configuration());
+   return rmqSink;
+   }
+
+   private RMQSink createRMQSinkFeatured() throws Exception {
--- End diff --

I see that you have already changed it to `createRMQSinkWithOptions`. I 
think this works.


---

[GitHub] flink pull request #5410: [FLINK-8468] [RabbitMQ Connector] Take advantage o...

[GJL]

Github user GJL commented on a diff in the pull request:

https://github.com/apache/flink/pull/5410#discussion_r193963587
  
--- Diff: 
flink-connectors/flink-connector-rabbitmq/src/main/java/org/apache/flink/streaming/connectors/rabbitmq/RMQSink.java
 ---
@@ -105,7 +153,18 @@ public void invoke(IN value) {
try {
byte[] msg = schema.serialize(value);
 
-   channel.basicPublish("", queueName, null, msg);
+   if (messageCompute == null) {
+   channel.basicPublish("", queueName, null, msg);
+   } else {
+   String rk = 
messageCompute.computeRoutingKey(value);
+   String exchange = 
messageCompute.computeExchange(value);
+   channel.basicPublish((exchange != null) ? 
exchange : "",
+   (rk != null) ? rk : "",
+   (returnListener != null) && 
messageCompute.computeMandatory(value),
--- End diff --

I think failing fast is a reasonable behavior. Can you also adapt the 
javadocs?


---

[GitHub] flink pull request #5410: [FLINK-8468] [RabbitMQ Connector] Take advantage o...

[pduveau]

Github user pduveau commented on a diff in the pull request:

https://github.com/apache/flink/pull/5410#discussion_r193377245
  
--- Diff: 
flink-connectors/flink-connector-rabbitmq/src/main/java/org/apache/flink/streaming/connectors/rabbitmq/RMQSink.java
 ---
@@ -105,7 +153,18 @@ public void invoke(IN value) {
try {
byte[] msg = schema.serialize(value);
 
-   channel.basicPublish("", queueName, null, msg);
+   if (messageCompute == null) {
+   channel.basicPublish("", queueName, null, msg);
+   } else {
+   String rk = 
messageCompute.computeRoutingKey(value);
+   String exchange = 
messageCompute.computeExchange(value);
+   channel.basicPublish((exchange != null) ? 
exchange : "",
+   (rk != null) ? rk : "",
+   (returnListener != null) && 
messageCompute.computeMandatory(value),
--- End diff --

For the two behavior are OK, if you think that throwing an exception is a 
better pattern I will change it. And in fact, throwing exception is easier to 
explain in documentation.   


---

[GitHub] flink pull request #5410: [FLINK-8468] [RabbitMQ Connector] Take advantage o...

[pduveau]

Github user pduveau commented on a diff in the pull request:

https://github.com/apache/flink/pull/5410#discussion_r193361587
  
--- Diff: 
flink-connectors/flink-connector-rabbitmq/src/test/java/org/apache/flink/streaming/connectors/rabbitmq/RMQSinkTest.java
 ---
@@ -83,7 +103,22 @@ public void throwExceptionIfChannelIsNull() throws 
Exception {
}
 
private RMQSink createRMQSink() throws Exception {
-   RMQSink rmqSink = new RMQSink(rmqConnectionConfig, 
QUEUE_NAME, serializationSchema);
+   RMQSink rmqSink = new 
RMQSink(rmqConnectionConfig, QUEUE_NAME, serializationSchema);
+   rmqSink.open(new Configuration());
+   return rmqSink;
+   }
+
+   private RMQSink createRMQSinkFeatured() throws Exception {
--- End diff --

I am using Featured but perhaps Enhanced should be a better choice. What do 
you think ?


---

[GitHub] flink pull request #5410: [FLINK-8468] [RabbitMQ Connector] Take advantage o...

[pduveau]

Github user pduveau commented on a diff in the pull request:

https://github.com/apache/flink/pull/5410#discussion_r193360923
  
--- Diff: 
flink-connectors/flink-connector-rabbitmq/src/test/java/org/apache/flink/streaming/connectors/rabbitmq/RMQSinkTest.java
 ---
@@ -83,7 +103,22 @@ public void throwExceptionIfChannelIsNull() throws 
Exception {
}
 
private RMQSink createRMQSink() throws Exception {
-   RMQSink rmqSink = new RMQSink(rmqConnectionConfig, 
QUEUE_NAME, serializationSchema);
+   RMQSink rmqSink = new 
RMQSink(rmqConnectionConfig, QUEUE_NAME, serializationSchema);
+   rmqSink.open(new Configuration());
+   return rmqSink;
+   }
+
+   private RMQSink createRMQSinkFeatured() throws Exception {
+   publishOptions = new DummyPublishOptions();
+   RMQSink rmqSink = new 
RMQSink(rmqConnectionConfig, serializationSchema, publishOptions);
+   rmqSink.open(new Configuration());
+   return rmqSink;
+   }
+
+   private RMQSink createRMQSinkFeaturedReturnHandler() throws 
Exception {
--- End diff --

You can see that if you do  not provide a returnedMessage Handler. You must 
not set immediate and/or mandatory to true when you publish even if you set 
them to true in class DummyPublishOptions they are overwritten in basicPublish 
call as you can see in the two test invokeFeaturedPublishBytesToQueue 
(overwritten) and  invokeFeaturedReturnHandlerPublishBytesToQueue (not 
overwritten).


---

[GitHub] flink pull request #5410: [FLINK-8468] [RabbitMQ Connector] Take advantage o...

[pduveau]

Github user pduveau commented on a diff in the pull request:

https://github.com/apache/flink/pull/5410#discussion_r193359300
  
--- Diff: 
flink-connectors/flink-connector-rabbitmq/src/test/java/org/apache/flink/streaming/connectors/rabbitmq/RMQSinkTest.java
 ---
@@ -83,7 +103,22 @@ public void throwExceptionIfChannelIsNull() throws 
Exception {
}
 
private RMQSink createRMQSink() throws Exception {
-   RMQSink rmqSink = new RMQSink(rmqConnectionConfig, 
QUEUE_NAME, serializationSchema);
+   RMQSink rmqSink = new 
RMQSink(rmqConnectionConfig, QUEUE_NAME, serializationSchema);
--- End diff --

Done


---

[GitHub] flink pull request #5410: [FLINK-8468] [RabbitMQ Connector] Take advantage o...

[pduveau]

Github user pduveau commented on a diff in the pull request:

https://github.com/apache/flink/pull/5410#discussion_r193359253
  
--- Diff: 
flink-connectors/flink-connector-rabbitmq/src/test/java/org/apache/flink/streaming/connectors/rabbitmq/RMQSinkTest.java
 ---
@@ -83,7 +103,22 @@ public void throwExceptionIfChannelIsNull() throws 
Exception {
}
 
private RMQSink createRMQSink() throws Exception {
-   RMQSink rmqSink = new RMQSink(rmqConnectionConfig, 
QUEUE_NAME, serializationSchema);
+   RMQSink rmqSink = new 
RMQSink(rmqConnectionConfig, QUEUE_NAME, serializationSchema);
+   rmqSink.open(new Configuration());
+   return rmqSink;
+   }
+
+   private RMQSink createRMQSinkFeatured() throws Exception {
+   publishOptions = new DummyPublishOptions();
+   RMQSink rmqSink = new 
RMQSink(rmqConnectionConfig, serializationSchema, publishOptions);
--- End diff --

Done


---

[GitHub] flink pull request #5410: [FLINK-8468] [RabbitMQ Connector] Take advantage o...

[pduveau]

Github user pduveau commented on a diff in the pull request:

https://github.com/apache/flink/pull/5410#discussion_r193358805
  
--- Diff: 
flink-connectors/flink-connector-rabbitmq/src/test/java/org/apache/flink/streaming/connectors/rabbitmq/RMQSinkTest.java
 ---
@@ -66,12 +79,19 @@ public void before() throws Exception {
}
 
@Test
-   public void openCallDeclaresQueue() throws Exception {
+   public void openCallDeclaresQueueInStandardMode() throws Exception {
createRMQSink();
 
verify(channel).queueDeclare(QUEUE_NAME, false, false, false, 
null);
}
 
+   @Test
+   public void openCallDontDeclaresQueueInFeaturedMode() throws Exception {
+   doThrow(Exception.class).when(channel).queueDeclare(null, 
false, false, false, null);
--- End diff --

Done


---

[GitHub] flink pull request #5410: [FLINK-8468] [RabbitMQ Connector] Take advantage o...

[pduveau]

Github user pduveau commented on a diff in the pull request:

https://github.com/apache/flink/pull/5410#discussion_r193357238
  
--- Diff: 
flink-connectors/flink-connector-rabbitmq/src/main/java/org/apache/flink/streaming/connectors/rabbitmq/RMQSink.java
 ---
@@ -40,22 +42,63 @@
 
private static final Logger LOG = 
LoggerFactory.getLogger(RMQSink.class);
 
-   protected final String queueName;
+   private final String queueName;
private final RMQConnectionConfig rmqConnectionConfig;
protected transient Connection connection;
protected transient Channel channel;
protected SerializationSchema schema;
-   private boolean logFailuresOnly = false;
+   protected boolean logFailuresOnly = false;
--- End diff --

In fact, now we have a method to change the value.


---

[GitHub] flink pull request #5410: [FLINK-8468] [RabbitMQ Connector] Take advantage o...

[pduveau]

Github user pduveau commented on a diff in the pull request:

https://github.com/apache/flink/pull/5410#discussion_r193356675
  
--- Diff: 
flink-connectors/flink-connector-rabbitmq/src/test/java/org/apache/flink/streaming/connectors/rabbitmq/RMQSinkTest.java
 ---
@@ -124,7 +159,82 @@ public void closeAllResources() throws Exception {
verify(connection).close();
}
 
+   @Test
+   public void invokeFeaturedPublishBytesToQueue() throws Exception {
+   RMQSink rmqSink = createRMQSinkFeatured();
+
+   rmqSink.invoke(MESSAGE_STR, SinkContextUtil.forTimestamp(0));
+   verify(serializationSchema).serialize(MESSAGE_STR);
+   verify(channel).basicPublish(EXCHANGE, ROUTING_KEY, false, 
false,
+   publishOptions.computeProperties(""), MESSAGE);
+   }
+
+   @Test
+   public void invokeFeaturedReturnHandlerPublishBytesToQueue() throws 
Exception {
+   RMQSink rmqSink = createRMQSinkFeaturedReturnHandler();
+
+   rmqSink.invoke(MESSAGE_STR, SinkContextUtil.forTimestamp(0));
+   verify(serializationSchema).serialize(MESSAGE_STR);
+   verify(channel).basicPublish(EXCHANGE, ROUTING_KEY, true, true,
+   publishOptions.computeProperties(""), MESSAGE);
+   }
+
+   @Test(expected = RuntimeException.class)
+   public void exceptionDuringFeaturedPublishingIsNotIgnored() throws 
Exception {
+   RMQSink rmqSink = createRMQSinkFeatured();
+
+   doThrow(IOException.class).when(channel).basicPublish(EXCHANGE, 
ROUTING_KEY, false, false,
+   publishOptions.computeProperties(""), MESSAGE);
+   rmqSink.invoke("msg", SinkContextUtil.forTimestamp(0));
+   }
+
+   @Test
+   public void 
exceptionDuringFeaturedPublishingIsIgnoredIfLogFailuresOnly() throws Exception {
+   RMQSink rmqSink = createRMQSinkFeatured();
+   rmqSink.setLogFailuresOnly(true);
+
+   doThrow(IOException.class).when(channel).basicPublish(EXCHANGE, 
ROUTING_KEY, false, false,
+   publishOptions.computeProperties(""), MESSAGE);
+   rmqSink.invoke("msg", SinkContextUtil.forTimestamp(0));
+   }
+
+   private class DummyPublishOptions implements 
RMQSinkPublishOptions {
--- End diff --

Done


---

[GitHub] flink pull request #5410: [FLINK-8468] [RabbitMQ Connector] Take advantage o...

[pduveau]

Github user pduveau commented on a diff in the pull request:

https://github.com/apache/flink/pull/5410#discussion_r193356713
  
--- Diff: 
flink-connectors/flink-connector-rabbitmq/src/main/java/org/apache/flink/streaming/connectors/rabbitmq/RMQSink.java
 ---
@@ -40,22 +42,63 @@
 
private static final Logger LOG = 
LoggerFactory.getLogger(RMQSink.class);
 
-   protected final String queueName;
+   private final String queueName;
--- End diff --

OK


---

[GitHub] flink pull request #5410: [FLINK-8468] [RabbitMQ Connector] Take advantage o...

[pduveau]

Github user pduveau commented on a diff in the pull request:

https://github.com/apache/flink/pull/5410#discussion_r193356620
  
--- Diff: 
flink-connectors/flink-connector-rabbitmq/src/main/java/org/apache/flink/streaming/connectors/rabbitmq/common/RMQConnectionConfig.java
 ---
@@ -109,6 +109,7 @@ private RMQConnectionConfig(String host, Integer port, 
String virtualHost, Strin
* @param requestedChannelMax requested maximum channel number
* @param requestedFrameMax requested maximum frame size
* @param requestedHeartbeat requested heartbeat interval
+   * @param createQueue enable or diable queue create on setup
--- End diff --

Sorry for this.


---

[GitHub] flink pull request #5410: [FLINK-8468] [RabbitMQ Connector] Take advantage o...

[GJL]

Github user GJL commented on a diff in the pull request:

https://github.com/apache/flink/pull/5410#discussion_r193126744
  
--- Diff: 
flink-connectors/flink-connector-rabbitmq/src/test/java/org/apache/flink/streaming/connectors/rabbitmq/RMQSinkTest.java
 ---
@@ -83,7 +103,22 @@ public void throwExceptionIfChannelIsNull() throws 
Exception {
}
 
private RMQSink createRMQSink() throws Exception {
-   RMQSink rmqSink = new RMQSink(rmqConnectionConfig, 
QUEUE_NAME, serializationSchema);
+   RMQSink rmqSink = new 
RMQSink(rmqConnectionConfig, QUEUE_NAME, serializationSchema);
+   rmqSink.open(new Configuration());
+   return rmqSink;
+   }
+
+   private RMQSink createRMQSinkFeatured() throws Exception {
--- End diff --

How about `createRMQSinkWithPublishOptions()`? I am not a AMQP specialist, 
if `RMQSinkFeatured` has a special meaning, then leave it.


---

[GitHub] flink pull request #5410: [FLINK-8468] [RabbitMQ Connector] Take advantage o...

[GJL]

Github user GJL commented on a diff in the pull request:

https://github.com/apache/flink/pull/5410#discussion_r193101683
  
--- Diff: 
flink-connectors/flink-connector-rabbitmq/src/test/java/org/apache/flink/streaming/connectors/rabbitmq/RMQSinkTest.java
 ---
@@ -83,7 +103,22 @@ public void throwExceptionIfChannelIsNull() throws 
Exception {
}
 
private RMQSink createRMQSink() throws Exception {
-   RMQSink rmqSink = new RMQSink(rmqConnectionConfig, 
QUEUE_NAME, serializationSchema);
+   RMQSink rmqSink = new 
RMQSink(rmqConnectionConfig, QUEUE_NAME, serializationSchema);
+   rmqSink.open(new Configuration());
+   return rmqSink;
+   }
+
+   private RMQSink createRMQSinkFeatured() throws Exception {
+   publishOptions = new DummyPublishOptions();
+   RMQSink rmqSink = new 
RMQSink(rmqConnectionConfig, serializationSchema, publishOptions);
+   rmqSink.open(new Configuration());
+   return rmqSink;
+   }
+
+   private RMQSink createRMQSinkFeaturedReturnHandler() throws 
Exception {
+   publishOptions = new DummyPublishOptions();
--- End diff --

`publishOptions` and `returnListener` can be local variables.


---

[GitHub] flink pull request #5410: [FLINK-8468] [RabbitMQ Connector] Take advantage o...

[GJL]

Github user GJL commented on a diff in the pull request:

https://github.com/apache/flink/pull/5410#discussion_r193127128
  
--- Diff: 
flink-connectors/flink-connector-rabbitmq/src/main/java/org/apache/flink/streaming/connectors/rabbitmq/RMQSink.java
 ---
@@ -105,7 +153,18 @@ public void invoke(IN value) {
try {
byte[] msg = schema.serialize(value);
 
-   channel.basicPublish("", queueName, null, msg);
+   if (messageCompute == null) {
+   channel.basicPublish("", queueName, null, msg);
+   } else {
+   String rk = 
messageCompute.computeRoutingKey(value);
+   String exchange = 
messageCompute.computeExchange(value);
+   channel.basicPublish((exchange != null) ? 
exchange : "",
+   (rk != null) ? rk : "",
+   (returnListener != null) && 
messageCompute.computeMandatory(value),
--- End diff --

I would prefer that an`IllegalStateException` is thrown if the user code 
returns `true` without a `returnListener`. 


---

[GitHub] flink pull request #5410: [FLINK-8468] [RabbitMQ Connector] Take advantage o...

[GJL]

Github user GJL commented on a diff in the pull request:

https://github.com/apache/flink/pull/5410#discussion_r193126347
  
--- Diff: 
flink-connectors/flink-connector-rabbitmq/src/test/java/org/apache/flink/streaming/connectors/rabbitmq/RMQSinkTest.java
 ---
@@ -83,7 +103,22 @@ public void throwExceptionIfChannelIsNull() throws 
Exception {
}
 
private RMQSink createRMQSink() throws Exception {
-   RMQSink rmqSink = new RMQSink(rmqConnectionConfig, 
QUEUE_NAME, serializationSchema);
+   RMQSink rmqSink = new 
RMQSink(rmqConnectionConfig, QUEUE_NAME, serializationSchema);
+   rmqSink.open(new Configuration());
+   return rmqSink;
+   }
+
+   private RMQSink createRMQSinkFeatured() throws Exception {
+   publishOptions = new DummyPublishOptions();
+   RMQSink rmqSink = new 
RMQSink(rmqConnectionConfig, serializationSchema, publishOptions);
+   rmqSink.open(new Configuration());
+   return rmqSink;
+   }
+
+   private RMQSink createRMQSinkFeaturedReturnHandler() throws 
Exception {
--- End diff --

How about `createRMQSinkWithReturnHandler()`?


---

[GitHub] flink pull request #5410: [FLINK-8468] [RabbitMQ Connector] Take advantage o...

[GJL]

Github user GJL commented on a diff in the pull request:

https://github.com/apache/flink/pull/5410#discussion_r193126169
  
--- Diff: 
flink-connectors/flink-connector-rabbitmq/src/test/java/org/apache/flink/streaming/connectors/rabbitmq/RMQSinkTest.java
 ---
@@ -83,7 +103,22 @@ public void throwExceptionIfChannelIsNull() throws 
Exception {
}
 
private RMQSink createRMQSink() throws Exception {
-   RMQSink rmqSink = new RMQSink(rmqConnectionConfig, 
QUEUE_NAME, serializationSchema);
+   RMQSink rmqSink = new 
RMQSink(rmqConnectionConfig, QUEUE_NAME, serializationSchema);
+   rmqSink.open(new Configuration());
+   return rmqSink;
+   }
+
+   private RMQSink createRMQSinkFeatured() throws Exception {
+   publishOptions = new DummyPublishOptions();
+   RMQSink rmqSink = new 
RMQSink(rmqConnectionConfig, serializationSchema, publishOptions);
+   rmqSink.open(new Configuration());
+   return rmqSink;
+   }
+
+   private RMQSink createRMQSinkFeaturedReturnHandler() throws 
Exception {
+   publishOptions = new DummyPublishOptions();
+   returnListener = new DummyReturnHandler();
+   RMQSink rmqSink = new 
RMQSink(rmqConnectionConfig, serializationSchema, publishOptions, 
returnListener);
--- End diff --

nit: Java 7 diamond operator should be preferred, i.e., `new RMQSink<>(...)`


---

[GitHub] flink pull request #5410: [FLINK-8468] [RabbitMQ Connector] Take advantage o...

[GJL]

Github user GJL commented on a diff in the pull request:

https://github.com/apache/flink/pull/5410#discussion_r193125374
  
--- Diff: 
flink-connectors/flink-connector-rabbitmq/src/main/java/org/apache/flink/streaming/connectors/rabbitmq/RMQSinkPublishOptions.java
 ---
@@ -0,0 +1,70 @@
+/*
+ * 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  The type of the data used by the sink.
+ */
+@PublicEvolving
+public interface RMQSinkPublishOptions 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
+*/
+   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
+*/
+   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);
--- End diff --

How about we add a default implementation that returns RabbitMQs default 
value:
```
default boolean computeMandatory(IN a) {
return false;
}


---

[GitHub] flink pull request #5410: [FLINK-8468] [RabbitMQ Connector] Take advantage o...

[GJL]

Github user GJL commented on a diff in the pull request:

https://github.com/apache/flink/pull/5410#discussion_r193122068
  
--- Diff: 
flink-connectors/flink-connector-rabbitmq/src/main/java/org/apache/flink/streaming/connectors/rabbitmq/RMQSink.java
 ---
@@ -40,22 +42,63 @@
 
private static final Logger LOG = 
LoggerFactory.getLogger(RMQSink.class);
 
-   protected final String queueName;
+   private final String queueName;
private final RMQConnectionConfig rmqConnectionConfig;
protected transient Connection connection;
protected transient Channel channel;
protected SerializationSchema schema;
-   private boolean logFailuresOnly = false;
+   protected boolean logFailuresOnly = false;
+
+   private final RMQSinkPublishOptions messageCompute;
+   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 messageCompute A {@link RMQSinkPublishOptions} for providing 
message's routing key and/or properties
  */
-   public RMQSink(RMQConnectionConfig rmqConnectionConfig, String 
queueName, SerializationSchema schema) {
+   private RMQSink(RMQConnectionConfig rmqConnectionConfig, String 
queueName, SerializationSchema schema,
--- End diff --

We should add some validation to assert  that `queueName` is `null` iff. 
`messageCompute` is set. We should also check that `RMQConnectionConfig` and 
`SerializationSchema` must not be null.
You may want to use `org.apache.flink.util.Preconditions` for that.


---

[GitHub] flink pull request #5410: [FLINK-8468] [RabbitMQ Connector] Take advantage o...

[GJL]

Github user GJL commented on a diff in the pull request:

https://github.com/apache/flink/pull/5410#discussion_r193119522
  
--- Diff: 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
--- End diff --

I think the Javadocs of the constructors are in better shape  than the 
contents of the `README.md`. I think we can drop this paragraph for now. What 
do you think?


---

[GitHub] flink pull request #5410: [FLINK-8468] [RabbitMQ Connector] Take advantage o...

[GJL]

Github user GJL commented on a diff in the pull request:

https://github.com/apache/flink/pull/5410#discussion_r193070753
  
--- Diff: 
flink-connectors/flink-connector-rabbitmq/src/test/java/org/apache/flink/streaming/connectors/rabbitmq/RMQSinkTest.java
 ---
@@ -66,12 +79,19 @@ public void before() throws Exception {
}
 
@Test
-   public void openCallDeclaresQueue() throws Exception {
+   public void openCallDeclaresQueueInStandardMode() throws Exception {
createRMQSink();
 
verify(channel).queueDeclare(QUEUE_NAME, false, false, false, 
null);
}
 
+   @Test
+   public void openCallDontDeclaresQueueInFeaturedMode() throws Exception {
+   doThrow(Exception.class).when(channel).queueDeclare(null, 
false, false, false, null);
--- End diff --

There is no assertion in this test. I would re-write it as:
```
@Test
public void openCallDontDeclaresQueueInFeaturedMode() throws Exception {
createRMQSinkFeatured();

verify(channel, never()).queueDeclare(null, false, false, 
false, null);
}
```


---

[GitHub] flink pull request #5410: [FLINK-8468] [RabbitMQ Connector] Take advantage o...

[GJL]

Github user GJL commented on a diff in the pull request:

https://github.com/apache/flink/pull/5410#discussion_r193120662
  
--- Diff: 
flink-connectors/flink-connector-rabbitmq/src/main/java/org/apache/flink/streaming/connectors/rabbitmq/RMQSink.java
 ---
@@ -105,7 +153,18 @@ public void invoke(IN value) {
try {
byte[] msg = schema.serialize(value);
 
-   channel.basicPublish("", queueName, null, msg);
+   if (messageCompute == null) {
+   channel.basicPublish("", queueName, null, msg);
+   } else {
+   String rk = 
messageCompute.computeRoutingKey(value);
+   String exchange = 
messageCompute.computeExchange(value);
+   channel.basicPublish((exchange != null) ? 
exchange : "",
--- End diff --

I think it is better to assume that `computeRoutingKey` and 
`computeExchange` must return a non-null value. It would be consistent with the 
`basicPublish` method which also rejects nulls.


---

[GitHub] flink pull request #5410: [FLINK-8468] [RabbitMQ Connector] Take advantage o...

[GJL]

Github user GJL commented on a diff in the pull request:

https://github.com/apache/flink/pull/5410#discussion_r193126219
  
--- Diff: 
flink-connectors/flink-connector-rabbitmq/src/test/java/org/apache/flink/streaming/connectors/rabbitmq/RMQSinkTest.java
 ---
@@ -83,7 +103,22 @@ public void throwExceptionIfChannelIsNull() throws 
Exception {
}
 
private RMQSink createRMQSink() throws Exception {
-   RMQSink rmqSink = new RMQSink(rmqConnectionConfig, 
QUEUE_NAME, serializationSchema);
+   RMQSink rmqSink = new 
RMQSink(rmqConnectionConfig, QUEUE_NAME, serializationSchema);
--- End diff --

nit: Java 7 diamond operator should be preferred, i.e., `new RMQSink<>(...)`


---

[GitHub] flink pull request #5410: [FLINK-8468] [RabbitMQ Connector] Take advantage o...

[GJL]

Github user GJL commented on a diff in the pull request:

https://github.com/apache/flink/pull/5410#discussion_r193125823
  
--- Diff: 
flink-connectors/flink-connector-rabbitmq/src/main/java/org/apache/flink/streaming/connectors/rabbitmq/RMQSinkPublishOptions.java
 ---
@@ -0,0 +1,70 @@
+/*
+ * 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  The type of the data used by the sink.
+ */
+@PublicEvolving
+public interface RMQSinkPublishOptions 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
+*/
+   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
+*/
+   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);
--- End diff --

Sorry I missed that RMQ does not implement this flag. We can remove it or 
add a Java 8 default implementation that returns `false`.


---

[GitHub] flink pull request #5410: [FLINK-8468] [RabbitMQ Connector] Take advantage o...

[GJL]

Github user GJL commented on a diff in the pull request:

https://github.com/apache/flink/pull/5410#discussion_r193126189
  
--- Diff: 
flink-connectors/flink-connector-rabbitmq/src/test/java/org/apache/flink/streaming/connectors/rabbitmq/RMQSinkTest.java
 ---
@@ -83,7 +103,22 @@ public void throwExceptionIfChannelIsNull() throws 
Exception {
}
 
private RMQSink createRMQSink() throws Exception {
-   RMQSink rmqSink = new RMQSink(rmqConnectionConfig, 
QUEUE_NAME, serializationSchema);
+   RMQSink rmqSink = new 
RMQSink(rmqConnectionConfig, QUEUE_NAME, serializationSchema);
+   rmqSink.open(new Configuration());
+   return rmqSink;
+   }
+
+   private RMQSink createRMQSinkFeatured() throws Exception {
+   publishOptions = new DummyPublishOptions();
+   RMQSink rmqSink = new 
RMQSink(rmqConnectionConfig, serializationSchema, publishOptions);
--- End diff --

nit: Java 7 diamond operator should be preferred, i.e., `new RMQSink<>(...)`


---

[GitHub] flink pull request #5410: [FLINK-8468] [RabbitMQ Connector] Take advantage o...

[GJL]

Github user GJL commented on a diff in the pull request:

https://github.com/apache/flink/pull/5410#discussion_r193122883
  
--- Diff: 
flink-connectors/flink-connector-rabbitmq/src/main/java/org/apache/flink/streaming/connectors/rabbitmq/RMQSink.java
 ---
@@ -40,22 +42,63 @@
 
private static final Logger LOG = 
LoggerFactory.getLogger(RMQSink.class);
 
-   protected final String queueName;
+   private final String queueName;
private final RMQConnectionConfig rmqConnectionConfig;
protected transient Connection connection;
protected transient Channel channel;
protected SerializationSchema schema;
-   private boolean logFailuresOnly = false;
+   protected boolean logFailuresOnly = false;
+
+   private final RMQSinkPublishOptions messageCompute;
+   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 messageCompute A {@link RMQSinkPublishOptions} for providing 
message's routing key and/or properties
  */
-   public RMQSink(RMQConnectionConfig rmqConnectionConfig, String 
queueName, SerializationSchema schema) {
+   private RMQSink(RMQConnectionConfig rmqConnectionConfig, String 
queueName, SerializationSchema schema,
+   RMQSinkPublishOptions messageCompute, 
ReturnListener returnListener) {
--- End diff --

`ReturnListener` is not `Serializable`. To work around this, we can add an 
interface as so:

```
public interface SerializableReturnListener extends ReturnListener, 
Serializable {
}
```


---

[GitHub] flink pull request #5410: [FLINK-8468] [RabbitMQ Connector] Take advantage o...

[GJL]

Github user GJL commented on a diff in the pull request:

https://github.com/apache/flink/pull/5410#discussion_r193123998
  
--- Diff: 
flink-connectors/flink-connector-rabbitmq/src/main/java/org/apache/flink/streaming/connectors/rabbitmq/RMQSinkPublishOptions.java
 ---
@@ -0,0 +1,70 @@
+/*
+ * 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  The type of the data used by the sink.
+ */
+@PublicEvolving
+public interface RMQSinkPublishOptions 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
+*/
+   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
+*/
+   String computeExchange(IN a);
+
+   /**
+* Compute the mandatory flag used in basic.publish method
--- End diff --

I think it's better to use Javadoc links: 
`{@link Channel#basicPublish(String, String, boolean, boolean, 
BasicProperties, byte[])}`



---

[GitHub] flink pull request #5410: [FLINK-8468] [RabbitMQ Connector] Take advantage o...

[GJL]

Github user GJL commented on a diff in the pull request:

https://github.com/apache/flink/pull/5410#discussion_r193010305
  
--- Diff: 
flink-connectors/flink-connector-rabbitmq/src/main/java/org/apache/flink/streaming/connectors/rabbitmq/RMQSink.java
 ---
@@ -40,22 +42,63 @@
 
private static final Logger LOG = 
LoggerFactory.getLogger(RMQSink.class);
 
-   protected final String queueName;
+   private final String queueName;
--- End diff --

Let's not change the visibility to `private` for now as it could break 
backwards compatibility.


---

[GitHub] flink pull request #5410: [FLINK-8468] [RabbitMQ Connector] Take advantage o...

[GJL]

Github user GJL commented on a diff in the pull request:

https://github.com/apache/flink/pull/5410#discussion_r193004950
  
--- Diff: 
flink-connectors/flink-connector-rabbitmq/src/main/java/org/apache/flink/streaming/connectors/rabbitmq/common/RMQConnectionConfig.java
 ---
@@ -109,6 +109,7 @@ private RMQConnectionConfig(String host, Integer port, 
String virtualHost, Strin
* @param requestedChannelMax requested maximum channel number
* @param requestedFrameMax requested maximum frame size
* @param requestedHeartbeat requested heartbeat interval
+   * @param createQueue enable or diable queue create on setup
--- End diff --

This parameter does not exist anymore.


---

[GitHub] flink pull request #5410: [FLINK-8468] [RabbitMQ Connector] Take advantage o...

[GJL]

Github user GJL commented on a diff in the pull request:

https://github.com/apache/flink/pull/5410#discussion_r193010378
  
--- Diff: 
flink-connectors/flink-connector-rabbitmq/src/main/java/org/apache/flink/streaming/connectors/rabbitmq/RMQSink.java
 ---
@@ -40,22 +42,63 @@
 
private static final Logger LOG = 
LoggerFactory.getLogger(RMQSink.class);
 
-   protected final String queueName;
+   private final String queueName;
private final RMQConnectionConfig rmqConnectionConfig;
protected transient Connection connection;
protected transient Channel channel;
protected SerializationSchema schema;
-   private boolean logFailuresOnly = false;
+   protected boolean logFailuresOnly = false;
--- End diff --

Is it needed to change the visibility to `protected`?


---

[GitHub] flink pull request #5410: [FLINK-8468] [RabbitMQ Connector] Take advantage o...

[GJL]

Github user GJL commented on a diff in the pull request:

https://github.com/apache/flink/pull/5410#discussion_r193006836
  
--- Diff: 
flink-connectors/flink-connector-rabbitmq/src/test/java/org/apache/flink/streaming/connectors/rabbitmq/RMQSinkTest.java
 ---
@@ -124,7 +159,82 @@ public void closeAllResources() throws Exception {
verify(connection).close();
}
 
+   @Test
+   public void invokeFeaturedPublishBytesToQueue() throws Exception {
+   RMQSink rmqSink = createRMQSinkFeatured();
+
+   rmqSink.invoke(MESSAGE_STR, SinkContextUtil.forTimestamp(0));
+   verify(serializationSchema).serialize(MESSAGE_STR);
+   verify(channel).basicPublish(EXCHANGE, ROUTING_KEY, false, 
false,
+   publishOptions.computeProperties(""), MESSAGE);
+   }
+
+   @Test
+   public void invokeFeaturedReturnHandlerPublishBytesToQueue() throws 
Exception {
+   RMQSink rmqSink = createRMQSinkFeaturedReturnHandler();
+
+   rmqSink.invoke(MESSAGE_STR, SinkContextUtil.forTimestamp(0));
+   verify(serializationSchema).serialize(MESSAGE_STR);
+   verify(channel).basicPublish(EXCHANGE, ROUTING_KEY, true, true,
+   publishOptions.computeProperties(""), MESSAGE);
+   }
+
+   @Test(expected = RuntimeException.class)
+   public void exceptionDuringFeaturedPublishingIsNotIgnored() throws 
Exception {
+   RMQSink rmqSink = createRMQSinkFeatured();
+
+   doThrow(IOException.class).when(channel).basicPublish(EXCHANGE, 
ROUTING_KEY, false, false,
+   publishOptions.computeProperties(""), MESSAGE);
+   rmqSink.invoke("msg", SinkContextUtil.forTimestamp(0));
+   }
+
+   @Test
+   public void 
exceptionDuringFeaturedPublishingIsIgnoredIfLogFailuresOnly() throws Exception {
+   RMQSink rmqSink = createRMQSinkFeatured();
+   rmqSink.setLogFailuresOnly(true);
+
+   doThrow(IOException.class).when(channel).basicPublish(EXCHANGE, 
ROUTING_KEY, false, false,
+   publishOptions.computeProperties(""), MESSAGE);
+   rmqSink.invoke("msg", SinkContextUtil.forTimestamp(0));
+   }
+
+   private class DummyPublishOptions implements 
RMQSinkPublishOptions {
--- End diff --

nit: `serialVersionUID` is missing.


---

[GitHub] flink pull request #5410: [FLINK-8468] [RabbitMQ Connector] Take advantage o...

[pduveau]

Github user pduveau commented on a diff in the pull request:

https://github.com/apache/flink/pull/5410#discussion_r191191853
  
--- Diff: 
flink-connectors/flink-connector-rabbitmq/src/main/java/org/apache/flink/streaming/connectors/rabbitmq/RMQSink.java
 ---
@@ -40,22 +40,36 @@
 
private static final Logger LOG = 
LoggerFactory.getLogger(RMQSink.class);
 
-   protected final String queueName;
+   protected String queueName;
private final RMQConnectionConfig rmqConnectionConfig;
protected transient Connection connection;
protected transient Channel channel;
protected SerializationSchema schema;
-   private boolean logFailuresOnly = false;
+   protected boolean logFailuresOnly = false;
+
+   protected RMQSinkPublishOptions messageCompute;
 
/**
 * @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
  */
public RMQSink(RMQConnectionConfig rmqConnectionConfig, String 
queueName, SerializationSchema schema) {
-   this.rmqConnectionConfig = rmqConnectionConfig;
+   this(rmqConnectionConfig, schema, null);
this.queueName = queueName;
+   }
+
+   /**
+* @param rmqConnectionConfig The RabbitMQ connection configuration 
{@link RMQConnectionConfig}.
+* @param schema A {@link SerializationSchema} for turning the Java 
objects received into bytes
+* @param messageCompute A {@link RMQSinkPublishOptions} for providing 
message's routing key and/or properties
+ */
+   public RMQSink(RMQConnectionConfig rmqConnectionConfig, 
SerializationSchema schema,
--- End diff --

Ok Done


---

[GitHub] flink pull request #5410: [FLINK-8468] [RabbitMQ Connector] Take advantage o...

[pduveau]

Github user pduveau commented on a diff in the pull request:

https://github.com/apache/flink/pull/5410#discussion_r191187177
  
--- Diff: 
flink-connectors/flink-connector-rabbitmq/src/main/java/org/apache/flink/streaming/connectors/rabbitmq/RMQSinkPublishOptions.java
 ---
@@ -0,0 +1,50 @@
+/*
+ * 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 com.rabbitmq.client.AMQP.BasicProperties;
+
+/**
+ * The message computation provides methods to compute the message routing 
key and/or the properties.
+ *
+ * @param  The type of the data used by the sink.
+ */
+public interface RMQSinkPublishOptions {
--- End diff --

I also add the immediate flag (even if ignored).


---

[GitHub] flink pull request #5410: [FLINK-8468] [RabbitMQ Connector] Take advantage o...

[pduveau]

Github user pduveau commented on a diff in the pull request:

https://github.com/apache/flink/pull/5410#discussion_r191185023
  
--- Diff: 
flink-connectors/flink-connector-rabbitmq/src/main/java/org/apache/flink/streaming/connectors/rabbitmq/RMQSinkPublishOptions.java
 ---
@@ -0,0 +1,50 @@
+/*
+ * 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 com.rabbitmq.client.AMQP.BasicProperties;
+
+/**
+ * The message computation provides methods to compute the message routing 
key and/or the properties.
+ *
+ * @param  The type of the data used by the sink.
+ */
+public interface RMQSinkPublishOptions {
--- End diff --

Ok done.


---

[GitHub] flink pull request #5410: [FLINK-8468] [RabbitMQ Connector] Take advantage o...

[pduveau]

Github user pduveau commented on a diff in the pull request:

https://github.com/apache/flink/pull/5410#discussion_r191184906
  
--- Diff: 
flink-connectors/flink-connector-rabbitmq/src/main/java/org/apache/flink/streaming/connectors/rabbitmq/RMQSinkPublishOptions.java
 ---
@@ -0,0 +1,50 @@
+/*
+ * 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 com.rabbitmq.client.AMQP.BasicProperties;
+
+/**
+ * The message computation provides methods to compute the message routing 
key and/or the properties.
+ *
+ * @param  The type of the data used by the sink.
+ */
+public interface RMQSinkPublishOptions {
--- End diff --

Very good notice, I did not think about this. I'll add it. 


---

[GitHub] flink pull request #5410: [FLINK-8468] [RabbitMQ Connector] Take advantage o...

[pduveau]

Github user pduveau commented on a diff in the pull request:

https://github.com/apache/flink/pull/5410#discussion_r191184148
  
--- Diff: 
flink-connectors/flink-connector-rabbitmq/src/main/java/org/apache/flink/streaming/connectors/rabbitmq/RMQSource.java
 ---
@@ -151,7 +151,9 @@ public void open(Configuration config) throws Exception 
{
if (channel == null) {
throw new RuntimeException("None of RabbitMQ 
channels are available");
}
-   setupQueue();
+   if (rmqConnectionConfig.hasToCreateQueueOnSetup()) {
--- End diff --

Acceptable in devel environment, in production the application should not 
be able to operate not-owned objects and creating an extended class to avoid 
the queueDeclare on start seems to be an heavy solution. However, we have a 
solution and it's complies to our needed. So we modify to your proposal.


---

[GitHub] flink pull request #5410: [FLINK-8468] [RabbitMQ Connector] Take advantage o...

[pduveau]

Github user pduveau commented on a diff in the pull request:

https://github.com/apache/flink/pull/5410#discussion_r191180121
  
--- Diff: 
flink-connectors/flink-connector-rabbitmq/src/main/java/org/apache/flink/streaming/connectors/rabbitmq/RMQSinkPublishOptions.java
 ---
@@ -0,0 +1,50 @@
+/*
+ * 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 com.rabbitmq.client.AMQP.BasicProperties;
+
+/**
+ * The message computation provides methods to compute the message routing 
key and/or the properties.
+ *
+ * @param  The type of the data used by the sink.
+ */
+public interface RMQSinkPublishOptions {
--- End diff --

Modified to your proposal.


---

[GitHub] flink pull request #5410: [FLINK-8468] [RabbitMQ Connector] Take advantage o...

[pduveau]

Github user pduveau commented on a diff in the pull request:

https://github.com/apache/flink/pull/5410#discussion_r191179994
  
--- Diff: 
flink-connectors/flink-connector-rabbitmq/src/main/java/org/apache/flink/streaming/connectors/rabbitmq/RMQSink.java
 ---
@@ -40,22 +40,36 @@
 
private static final Logger LOG = 
LoggerFactory.getLogger(RMQSink.class);
 
-   protected final String queueName;
+   protected String queueName;
private final RMQConnectionConfig rmqConnectionConfig;
protected transient Connection connection;
protected transient Channel channel;
protected SerializationSchema schema;
-   private boolean logFailuresOnly = false;
+   protected boolean logFailuresOnly = false;
+
+   protected RMQSinkPublishOptions messageCompute;
--- End diff --

Modified to your proposal.


---

[GitHub] flink pull request #5410: [FLINK-8468] [RabbitMQ Connector] Take advantage o...

[pduveau]

Github user pduveau commented on a diff in the pull request:

https://github.com/apache/flink/pull/5410#discussion_r191179983
  
--- Diff: 
flink-connectors/flink-connector-rabbitmq/src/test/java/org/apache/flink/streaming/connectors/rabbitmq/RMQSinkTest.java
 ---
@@ -43,14 +49,27 @@
 public class RMQSinkTest {
 
private static final String QUEUE_NAME = "queue";
+   private static final String EXCHANGE = "exchange";
+   private static final String ROUTING_KEY = "application.component.error";
+   private static final String EXPIRATION = "1";
private static final String MESSAGE_STR = "msg";
private static final byte[] MESSAGE = new byte[1];
+   private static Map headers = new HashMap();
+   private static AMQP.BasicProperties props;
--- End diff --

Modified to your proposal.


---

[GitHub] flink pull request #5410: [FLINK-8468] [RabbitMQ Connector] Take advantage o...

[GJL]

Github user GJL commented on a diff in the pull request:

https://github.com/apache/flink/pull/5410#discussion_r190658100
  
--- Diff: 
flink-connectors/flink-connector-rabbitmq/src/main/java/org/apache/flink/streaming/connectors/rabbitmq/RMQSinkPublishOptions.java
 ---
@@ -0,0 +1,50 @@
+/*
+ * 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 com.rabbitmq.client.AMQP.BasicProperties;
+
+/**
+ * The message computation provides methods to compute the message routing 
key and/or the properties.
+ *
+ * @param  The type of the data used by the sink.
+ */
+public interface RMQSinkPublishOptions {
+
+   /**
+* Compute the message's routing key from the data.
+* @param a The data used by the sink
+* @return The routing key of the message
+*/
+   public String computeRoutingKey(IN a);
--- End diff --

The `public` keyword is redundant here because method declarations in an 
interface are always `public`. 


---

[GitHub] flink pull request #5410: [FLINK-8468] [RabbitMQ Connector] Take advantage o...

[GJL]

Github user GJL commented on a diff in the pull request:

https://github.com/apache/flink/pull/5410#discussion_r190888216
  
--- Diff: 
flink-connectors/flink-connector-rabbitmq/src/main/java/org/apache/flink/streaming/connectors/rabbitmq/RMQSinkPublishOptions.java
 ---
@@ -0,0 +1,50 @@
+/*
+ * 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 com.rabbitmq.client.AMQP.BasicProperties;
+
+/**
+ * The message computation provides methods to compute the message routing 
key and/or the properties.
+ *
+ * @param  The type of the data used by the sink.
+ */
+public interface RMQSinkPublishOptions {
--- End diff --

I think the interface should be annotated with `@PublicEvolving`.


---

[GitHub] flink pull request #5410: [FLINK-8468] [RabbitMQ Connector] Take advantage o...

[GJL]

Github user GJL commented on a diff in the pull request:

https://github.com/apache/flink/pull/5410#discussion_r190650550
  
--- Diff: 
flink-connectors/flink-connector-rabbitmq/src/main/java/org/apache/flink/streaming/connectors/rabbitmq/RMQSinkPublishOptions.java
 ---
@@ -0,0 +1,50 @@
+/*
+ * 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 com.rabbitmq.client.AMQP.BasicProperties;
+
+/**
+ * The message computation provides methods to compute the message routing 
key and/or the properties.
+ *
+ * @param  The type of the data used by the sink.
+ */
+public interface RMQSinkPublishOptions {
--- End diff --

This must extend `java.io.Serializable` because `RMQSink` has a 
non-transient field of this type.


---

[GitHub] flink pull request #5410: [FLINK-8468] [RabbitMQ Connector] Take advantage o...

[GJL]

Github user GJL commented on a diff in the pull request:

https://github.com/apache/flink/pull/5410#discussion_r190654087
  
--- Diff: 
flink-connectors/flink-connector-rabbitmq/src/main/java/org/apache/flink/streaming/connectors/rabbitmq/RMQSource.java
 ---
@@ -151,7 +151,9 @@ public void open(Configuration config) throws Exception 
{
if (channel == null) {
throw new RuntimeException("None of RabbitMQ 
channels are available");
}
-   setupQueue();
+   if (rmqConnectionConfig.hasToCreateQueueOnSetup()) {
--- End diff --

There was a PR that did something similar and got rejected. See: 
https://github.com/apache/flink/pull/4979 and 
https://issues.apache.org/jira/browse/FLINK-8018


---

[GitHub] flink pull request #5410: [FLINK-8468] [RabbitMQ Connector] Take advantage o...

[GJL]

Github user GJL commented on a diff in the pull request:

https://github.com/apache/flink/pull/5410#discussion_r190888669
  
--- Diff: 
flink-connectors/flink-connector-rabbitmq/src/main/java/org/apache/flink/streaming/connectors/rabbitmq/RMQSink.java
 ---
@@ -40,22 +40,36 @@
 
private static final Logger LOG = 
LoggerFactory.getLogger(RMQSink.class);
 
-   protected final String queueName;
+   protected String queueName;
private final RMQConnectionConfig rmqConnectionConfig;
protected transient Connection connection;
protected transient Channel channel;
protected SerializationSchema schema;
-   private boolean logFailuresOnly = false;
+   protected boolean logFailuresOnly = false;
+
+   protected RMQSinkPublishOptions messageCompute;
 
/**
 * @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
  */
public RMQSink(RMQConnectionConfig rmqConnectionConfig, String 
queueName, SerializationSchema schema) {
-   this.rmqConnectionConfig = rmqConnectionConfig;
+   this(rmqConnectionConfig, schema, null);
this.queueName = queueName;
+   }
+
+   /**
+* @param rmqConnectionConfig The RabbitMQ connection configuration 
{@link RMQConnectionConfig}.
+* @param schema A {@link SerializationSchema} for turning the Java 
objects received into bytes
+* @param messageCompute A {@link RMQSinkPublishOptions} for providing 
message's routing key and/or properties
+ */
+   public RMQSink(RMQConnectionConfig rmqConnectionConfig, 
SerializationSchema schema,
--- End diff --

I think the constructor should be annotated with `@PublicEvolving`, and 
`messageCompute` should come after `rmqConnectionConfig` to be consistent with 
the other constructor.


---

[GitHub] flink pull request #5410: [FLINK-8468] [RabbitMQ Connector] Take advantage o...

[GJL]

Github user GJL commented on a diff in the pull request:

https://github.com/apache/flink/pull/5410#discussion_r190887946
  
--- Diff: 
flink-connectors/flink-connector-rabbitmq/src/main/java/org/apache/flink/streaming/connectors/rabbitmq/RMQSinkPublishOptions.java
 ---
@@ -0,0 +1,50 @@
+/*
+ * 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 com.rabbitmq.client.AMQP.BasicProperties;
+
+/**
+ * The message computation provides methods to compute the message routing 
key and/or the properties.
+ *
+ * @param  The type of the data used by the sink.
+ */
+public interface RMQSinkPublishOptions {
--- End diff --

Shouldn't be the `mandatory` flag in 
`
basicPublish(String exchange, String routingKey, boolean mandatory, 
BasicProperties props, byte[] body) throws IOException` covered as well?


---

[GitHub] flink pull request #5410: [FLINK-8468] [RabbitMQ Connector] Take advantage o...

[GJL]

Github user GJL commented on a diff in the pull request:

https://github.com/apache/flink/pull/5410#discussion_r190648628
  
--- Diff: 
flink-connectors/flink-connector-rabbitmq/src/main/java/org/apache/flink/streaming/connectors/rabbitmq/RMQSink.java
 ---
@@ -40,22 +40,36 @@
 
private static final Logger LOG = 
LoggerFactory.getLogger(RMQSink.class);
 
-   protected final String queueName;
+   protected String queueName;
private final RMQConnectionConfig rmqConnectionConfig;
protected transient Connection connection;
protected transient Channel channel;
protected SerializationSchema schema;
-   private boolean logFailuresOnly = false;
+   protected boolean logFailuresOnly = false;
+
+   protected RMQSinkPublishOptions messageCompute;
--- End diff --

This field and `queueName` should be `final`. I think you need a third 
private constructor to achieve this:

```
private RMQSink(RMQConnectionConfig rmqConnectionConfig,
String queueName,
SerializationSchema schema,
RMQSinkPublishOptions 
messageCompute) {
this.rmqConnectionConfig = rmqConnectionConfig;
this.queueName = queueName;
this.schema = schema;
this.messageCompute = messageCompute;
}
```

Then you can write
```
public RMQSink(RMQConnectionConfig rmqConnectionConfig,
   SerializationSchema schema,
   RMQSinkPublishOptions messageCompute) {
this(rmqConnectionConfig, null, schema, messageCompute);
}
```


---

[GitHub] flink pull request #5410: [FLINK-8468] [RabbitMQ Connector] Take advantage o...

[GJL]

Github user GJL commented on a diff in the pull request:

https://github.com/apache/flink/pull/5410#discussion_r190647503
  
--- Diff: 
flink-connectors/flink-connector-rabbitmq/src/test/java/org/apache/flink/streaming/connectors/rabbitmq/RMQSinkTest.java
 ---
@@ -43,14 +49,27 @@
 public class RMQSinkTest {
 
private static final String QUEUE_NAME = "queue";
+   private static final String EXCHANGE = "exchange";
+   private static final String ROUTING_KEY = "application.component.error";
+   private static final String EXPIRATION = "1";
private static final String MESSAGE_STR = "msg";
private static final byte[] MESSAGE = new byte[1];
+   private static Map headers = new HashMap();
+   private static AMQP.BasicProperties props;
--- End diff --

`headers.put("Test", new String("My Value"));` can be simplified to 
`headers.put("Test", "My Value");`
The static field `headers` is mutable state, which should be avoided.
Why not just:
```
private static AMQP.BasicProperties props = new 
AMQP.BasicProperties.Builder()
.headers(Collections.singletonMap("Test", "My Value"))
.expiration(EXPIRATION)
.build();
```
Then the static initializer is not even needed.


---

[GitHub] flink pull request #5410: [FLINK-8468] [RabbitMQ Connector] Take advantage o...

[pduveau]

GitHub user pduveau opened a pull request:

https://github.com/apache/flink/pull/5410

[FLINK-8468] [RabbitMQ Connector] Take advantage of AMQP features

## The purpose of the change

To extend the Sink Connector to take full advantage of AMQP features. The 
AMQP Message sent can :
- receive headers computed using the incoming data,
- be published to a computed exchange,
- be routed with a computed routing key.

This is done while using a new constructor. The existing constructor was 
not modified for compatibility but the queue creation on setup can be disabled 
by configuration. 

## Change log

A interface class is added to define methods to compute the three different 
parameter listed above. 
A new constructor is added to the Sink class that takes an implemented 
object of this interface as parameter.
Using this new constructor also disable the queue creation (queueName 
attribute is null) 

## Verifying this change

This change added tests and can be verified as follows:
  - a new test class build on existing one is provided
  - the test cover the new mechanisms for the three publishing parameter.

## Does this pull request potentially affect one of the following parts:

  - Dependencies (does it add or upgrade a dependency): no
  - The public API, i.e., is any changed class annotated with 
`@Public(Evolving)`: no
  - The serializers: no
  - The runtime per-record code paths (performance sensitive): no
  - Anything that affects deployment or recovery: JobManager (and its 
components), Checkpointing, Yarn/Mesos, ZooKeeper: no
  - The S3 file system connector: no

## Documentation

  - Does this pull request introduce a new feature? yes
  - If yes, how is the feature documented? JavaDocs

You can merge this pull request into a Git repository by running:

$ git pull https://github.com/pduveau/flink Extended_RabbitMQ_Connector

Alternatively you can review and apply these changes as the patch at:

https://github.com/apache/flink/pull/5410.patch

To close this pull request, make a commit to your master/trunk branch
with (at least) the following in the commit message:

This closes #5410


commit 64072b1c8e6cc0af320361bd518ca4b0ed50
Author: Philippe Duveau 
Date:   2018-02-04T00:15:25Z

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

commit 388278eb5ed8e8330d6c23555d81d87bc12dc8fd
Author: Philippe Duveau 
Date:   2018-02-04T10:10:04Z

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

commit eec3504c29e1f48d535a2fc65423fa5ccfe8985e
Author: Philippe Duveau 
Date:   2018-02-04T10:28:44Z

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

commit c900f507690ddaa7ec9734e5002fe1bc2494c5da
Author: Philippe Duveau 
Date:   2018-02-04T10:37:53Z

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




---