You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@flink.apache.org by pduveau <gi...@git.apache.org> on 2018/02/04 10:40:55 UTC
[GitHub] flink pull request #5410: [FLINK-8468] [RabbitMQ Connector] Take advantage o...
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 64072b1c8e6cc0af3203ffff61bd518ca4b0ed50
Author: Philippe Duveau <ph...@...>
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 <ph...@...>
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 <ph...@...>
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 <ph...@...>
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
----
---
[GitHub] flink pull request #5410: [FLINK-8468] [RabbitMQ Connector] Take advantage o...
Posted by GJL <gi...@git.apache.org>.
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 <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
+ */
+ 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...
Posted by GJL <gi...@git.apache.org>.
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<IN> 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...
Posted by GJL <gi...@git.apache.org>.
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<String> createRMQSink() throws Exception {
- RMQSink rmqSink = new RMQSink<String>(rmqConnectionConfig, QUEUE_NAME, serializationSchema);
+ RMQSink<String> rmqSink = new RMQSink<String>(rmqConnectionConfig, QUEUE_NAME, serializationSchema);
+ rmqSink.open(new Configuration());
+ return rmqSink;
+ }
+
+ private RMQSink<String> 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...
Posted by GJL <gi...@git.apache.org>.
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 <IN> The type of the data used by the sink.
+ */
+public interface RMQSinkPublishOptions<IN> {
+
+ /**
+ * 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...
Posted by GJL <gi...@git.apache.org>.
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<IN> schema;
- private boolean logFailuresOnly = false;
+ protected boolean logFailuresOnly = false;
+
+ private final RMQSinkPublishOptions<IN> 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<IN> schema) {
+ private RMQSink(RMQConnectionConfig rmqConnectionConfig, String queueName, SerializationSchema<IN> 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 issue #5410: [FLINK-8468] [RabbitMQ Connector] Take advantage of AMQP ...
Posted by GJL <gi...@git.apache.org>.
Github user GJL commented on the issue:
https://github.com/apache/flink/pull/5410
Taking a look once more.
---
[GitHub] flink pull request #5410: [FLINK-8468] [RabbitMQ Connector] Take advantage o...
Posted by GJL <gi...@git.apache.org>.
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<IN> schema;
- private boolean logFailuresOnly = false;
+ protected boolean logFailuresOnly = false;
+
+ private final RMQSinkPublishOptions<IN> 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<IN> schema) {
+ private RMQSink(RMQConnectionConfig rmqConnectionConfig, String queueName, SerializationSchema<IN> schema,
+ RMQSinkPublishOptions<IN> 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...
Posted by GJL <gi...@git.apache.org>.
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...
Posted by pduveau <gi...@git.apache.org>.
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 <IN> The type of the data used by the sink.
+ */
+public interface RMQSinkPublishOptions<IN> {
--- End diff --
Modified to your proposal.
---
[GitHub] flink pull request #5410: [FLINK-8468] [RabbitMQ Connector] Take advantage o...
Posted by GJL <gi...@git.apache.org>.
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<String> createRMQSink() throws Exception {
- RMQSink rmqSink = new RMQSink<String>(rmqConnectionConfig, QUEUE_NAME, serializationSchema);
+ RMQSink<String> rmqSink = new RMQSink<String>(rmqConnectionConfig, QUEUE_NAME, serializationSchema);
+ rmqSink.open(new Configuration());
+ return rmqSink;
+ }
+
+ private RMQSink<String> 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...
Posted by GJL <gi...@git.apache.org>.
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...
Posted by pduveau <gi...@git.apache.org>.
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 <IN> The type of the data used by the sink.
+ */
+public interface RMQSinkPublishOptions<IN> {
--- End diff --
I also add the immediate flag (even if ignored).
---
[GitHub] flink pull request #5410: [FLINK-8468] [RabbitMQ Connector] Take advantage o...
Posted by GJL <gi...@git.apache.org>.
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<String> createRMQSink() throws Exception {
- RMQSink rmqSink = new RMQSink<String>(rmqConnectionConfig, QUEUE_NAME, serializationSchema);
+ RMQSink<String> rmqSink = new RMQSink<String>(rmqConnectionConfig, QUEUE_NAME, serializationSchema);
+ rmqSink.open(new Configuration());
+ return rmqSink;
+ }
+
+ private RMQSink<String> createRMQSinkFeatured() throws Exception {
+ publishOptions = new DummyPublishOptions();
+ RMQSink<String> rmqSink = new RMQSink<String>(rmqConnectionConfig, serializationSchema, publishOptions);
+ rmqSink.open(new Configuration());
+ return rmqSink;
+ }
+
+ private RMQSink<String> createRMQSinkFeaturedReturnHandler() throws Exception {
--- End diff --
How about `createRMQSinkWithReturnHandler()`?
---
[GitHub] flink pull request #5410: [FLINK-8468] [RabbitMQ Connector] Take advantage o...
Posted by pduveau <gi...@git.apache.org>.
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...
Posted by GJL <gi...@git.apache.org>.
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 <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
+ */
+ 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...
Posted by pduveau <gi...@git.apache.org>.
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<IN> 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...
Posted by pduveau <gi...@git.apache.org>.
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...
Posted by GJL <gi...@git.apache.org>.
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...
Posted by pduveau <gi...@git.apache.org>.
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<String> createRMQSink() throws Exception {
- RMQSink rmqSink = new RMQSink<String>(rmqConnectionConfig, QUEUE_NAME, serializationSchema);
+ RMQSink<String> rmqSink = new RMQSink<String>(rmqConnectionConfig, QUEUE_NAME, serializationSchema);
+ rmqSink.open(new Configuration());
+ return rmqSink;
+ }
+
+ private RMQSink<String> createRMQSinkFeatured() throws Exception {
+ publishOptions = new DummyPublishOptions();
+ RMQSink<String> rmqSink = new RMQSink<String>(rmqConnectionConfig, serializationSchema, publishOptions);
--- End diff --
Done
---
[GitHub] flink pull request #5410: [FLINK-8468] [RabbitMQ Connector] Take advantage o...
Posted by GJL <gi...@git.apache.org>.
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...
Posted by pduveau <gi...@git.apache.org>.
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 <IN> The type of the data used by the sink.
+ */
+public interface RMQSinkPublishOptions<IN> {
--- End diff --
Ok done.
---
[GitHub] flink pull request #5410: [FLINK-8468] [RabbitMQ Connector] Take advantage o...
Posted by GJL <gi...@git.apache.org>.
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...
Posted by GJL <gi...@git.apache.org>.
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<IN> schema;
- private boolean logFailuresOnly = false;
+ protected boolean logFailuresOnly = false;
+
+ private final RMQSinkPublishOptions<IN> 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<IN> schema) {
+ private RMQSink(RMQConnectionConfig rmqConnectionConfig, String queueName, SerializationSchema<IN> schema,
+ RMQSinkPublishOptions<IN> messageCompute, ReturnListener returnListener) {
--- End diff --
ping
---
[GitHub] flink pull request #5410: [FLINK-8468] [RabbitMQ Connector] Take advantage o...
Posted by pduveau <gi...@git.apache.org>.
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<String> createRMQSink() throws Exception {
- RMQSink rmqSink = new RMQSink<String>(rmqConnectionConfig, QUEUE_NAME, serializationSchema);
+ RMQSink<String> rmqSink = new RMQSink<String>(rmqConnectionConfig, QUEUE_NAME, serializationSchema);
+ rmqSink.open(new Configuration());
+ return rmqSink;
+ }
+
+ private RMQSink<String> createRMQSinkFeatured() throws Exception {
+ publishOptions = new DummyPublishOptions();
+ RMQSink<String> rmqSink = new RMQSink<String>(rmqConnectionConfig, serializationSchema, publishOptions);
+ rmqSink.open(new Configuration());
+ return rmqSink;
+ }
+
+ private RMQSink<String> 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 issue #5410: [FLINK-8468] [RabbitMQ Connector] Take advantage of AMQP ...
Posted by GJL <gi...@git.apache.org>.
Github user GJL commented on the issue:
https://github.com/apache/flink/pull/5410
Thanks for updating @pduveau. There still some minor comments on `RMQSinkPublishOptions` and `README` that are open for discussion.
---
[GitHub] flink pull request #5410: [FLINK-8468] [RabbitMQ Connector] Take advantage o...
Posted by pduveau <gi...@git.apache.org>.
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<String> 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<String> 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<String> 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<String> 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<String> {
--- End diff --
Done
---
[GitHub] flink pull request #5410: [FLINK-8468] [RabbitMQ Connector] Take advantage o...
Posted by pduveau <gi...@git.apache.org>.
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<String> createRMQSink() throws Exception {
- RMQSink rmqSink = new RMQSink<String>(rmqConnectionConfig, QUEUE_NAME, serializationSchema);
+ RMQSink<String> rmqSink = new RMQSink<String>(rmqConnectionConfig, QUEUE_NAME, serializationSchema);
--- End diff --
Done
---
[GitHub] flink pull request #5410: [FLINK-8468] [RabbitMQ Connector] Take advantage o...
Posted by asfgit <gi...@git.apache.org>.
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...
Posted by pduveau <gi...@git.apache.org>.
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<String> createRMQSink() throws Exception {
- RMQSink rmqSink = new RMQSink<String>(rmqConnectionConfig, QUEUE_NAME, serializationSchema);
+ RMQSink<String> rmqSink = new RMQSink<String>(rmqConnectionConfig, QUEUE_NAME, serializationSchema);
+ rmqSink.open(new Configuration());
+ return rmqSink;
+ }
+
+ private RMQSink<String> 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...
Posted by GJL <gi...@git.apache.org>.
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<IN> schema;
- private boolean logFailuresOnly = false;
+ protected boolean logFailuresOnly = false;
+
+ protected RMQSinkPublishOptions<IN> 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<IN> schema,
RMQSinkPublishOptions<IN> messageCompute) {
this.rmqConnectionConfig = rmqConnectionConfig;
this.queueName = queueName;
this.schema = schema;
this.messageCompute = messageCompute;
}
```
Then you can write
```
public RMQSink(RMQConnectionConfig rmqConnectionConfig,
SerializationSchema<IN> schema,
RMQSinkPublishOptions<IN> messageCompute) {
this(rmqConnectionConfig, null, schema, messageCompute);
}
```
---
[GitHub] flink pull request #5410: [FLINK-8468] [RabbitMQ Connector] Take advantage o...
Posted by GJL <gi...@git.apache.org>.
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<String> createRMQSink() throws Exception {
- RMQSink rmqSink = new RMQSink<String>(rmqConnectionConfig, QUEUE_NAME, serializationSchema);
+ RMQSink<String> rmqSink = new RMQSink<String>(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...
Posted by GJL <gi...@git.apache.org>.
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...
Posted by GJL <gi...@git.apache.org>.
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<String> createRMQSink() throws Exception {
- RMQSink rmqSink = new RMQSink<String>(rmqConnectionConfig, QUEUE_NAME, serializationSchema);
+ RMQSink<String> rmqSink = new RMQSink<String>(rmqConnectionConfig, QUEUE_NAME, serializationSchema);
+ rmqSink.open(new Configuration());
+ return rmqSink;
+ }
+
+ private RMQSink<String> createRMQSinkFeatured() throws Exception {
+ publishOptions = new DummyPublishOptions();
+ RMQSink<String> rmqSink = new RMQSink<String>(rmqConnectionConfig, serializationSchema, publishOptions);
+ rmqSink.open(new Configuration());
+ return rmqSink;
+ }
+
+ private RMQSink<String> 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...
Posted by pduveau <gi...@git.apache.org>.
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...
Posted by GJL <gi...@git.apache.org>.
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 <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
+ */
+ 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...
Posted by pduveau <gi...@git.apache.org>.
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<IN> schema;
- private boolean logFailuresOnly = false;
+ protected boolean logFailuresOnly = false;
+
+ protected RMQSinkPublishOptions<IN> messageCompute;
--- End diff --
Modified to your proposal.
---
[GitHub] flink pull request #5410: [FLINK-8468] [RabbitMQ Connector] Take advantage o...
Posted by GJL <gi...@git.apache.org>.
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 <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
+ */
+ 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...
Posted by GJL <gi...@git.apache.org>.
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...
Posted by pduveau <gi...@git.apache.org>.
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 = "10000";
private static final String MESSAGE_STR = "msg";
private static final byte[] MESSAGE = new byte[1];
+ private static Map<String, Object> headers = new HashMap<String, Object>();
+ private static AMQP.BasicProperties props;
--- End diff --
Modified to your proposal.
---
[GitHub] flink pull request #5410: [FLINK-8468] [RabbitMQ Connector] Take advantage o...
Posted by GJL <gi...@git.apache.org>.
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<IN> schema;
- private boolean logFailuresOnly = false;
+ protected boolean logFailuresOnly = false;
+
+ protected RMQSinkPublishOptions<IN> 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<IN> 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<IN> 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...
Posted by GJL <gi...@git.apache.org>.
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<IN> schema;
- private boolean logFailuresOnly = false;
+ protected boolean logFailuresOnly = false;
+
+ private final RMQSinkPublishOptions<IN> 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<IN> schema) {
+ private RMQSink(RMQConnectionConfig rmqConnectionConfig, String queueName, SerializationSchema<IN> schema,
--- End diff --
ping
---
[GitHub] flink issue #5410: [FLINK-8468] [RabbitMQ Connector] Take advantage of AMQP ...
Posted by pduveau <gi...@git.apache.org>.
Github user pduveau commented on the issue:
https://github.com/apache/flink/pull/5410
Hi, Is there anyone to check this PR ? We have interest in getting an official version including this.
---
[GitHub] flink pull request #5410: [FLINK-8468] [RabbitMQ Connector] Take advantage o...
Posted by GJL <gi...@git.apache.org>.
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<String> createRMQSink() throws Exception {
- RMQSink rmqSink = new RMQSink<String>(rmqConnectionConfig, QUEUE_NAME, serializationSchema);
+ RMQSink<String> rmqSink = new RMQSink<String>(rmqConnectionConfig, QUEUE_NAME, serializationSchema);
+ rmqSink.open(new Configuration());
+ return rmqSink;
+ }
+
+ private RMQSink<String> createRMQSinkFeatured() throws Exception {
+ publishOptions = new DummyPublishOptions();
+ RMQSink<String> rmqSink = new RMQSink<String>(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...
Posted by pduveau <gi...@git.apache.org>.
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...
Posted by GJL <gi...@git.apache.org>.
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 <IN> The type of the data used by the sink.
+ */
+public interface RMQSinkPublishOptions<IN> {
--- 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...
Posted by GJL <gi...@git.apache.org>.
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 = "10000";
private static final String MESSAGE_STR = "msg";
private static final byte[] MESSAGE = new byte[1];
+ private static Map<String, Object> headers = new HashMap<String, Object>();
+ 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...
Posted by GJL <gi...@git.apache.org>.
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...
Posted by GJL <gi...@git.apache.org>.
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<String> createRMQSink() throws Exception {
- RMQSink rmqSink = new RMQSink<String>(rmqConnectionConfig, QUEUE_NAME, serializationSchema);
+ RMQSink<String> rmqSink = new RMQSink<String>(rmqConnectionConfig, QUEUE_NAME, serializationSchema);
+ rmqSink.open(new Configuration());
+ return rmqSink;
+ }
+
+ private RMQSink<String> createRMQSinkFeatured() throws Exception {
+ publishOptions = new DummyPublishOptions();
+ RMQSink<String> rmqSink = new RMQSink<String>(rmqConnectionConfig, serializationSchema, publishOptions);
+ rmqSink.open(new Configuration());
+ return rmqSink;
+ }
+
+ private RMQSink<String> createRMQSinkFeaturedReturnHandler() throws Exception {
+ publishOptions = new DummyPublishOptions();
+ returnListener = new DummyReturnHandler();
+ RMQSink<String> rmqSink = new RMQSink<String>(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...
Posted by GJL <gi...@git.apache.org>.
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...
Posted by GJL <gi...@git.apache.org>.
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 <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
+ */
+ 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...
Posted by pduveau <gi...@git.apache.org>.
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...
Posted by GJL <gi...@git.apache.org>.
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 <IN> The type of the data used by the sink.
+ */
+public interface RMQSinkPublishOptions<IN> {
--- 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...
Posted by pduveau <gi...@git.apache.org>.
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...
Posted by pduveau <gi...@git.apache.org>.
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<IN> schema;
- private boolean logFailuresOnly = false;
+ protected boolean logFailuresOnly = false;
+
+ protected RMQSinkPublishOptions<IN> 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<IN> 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<IN> schema,
--- End diff --
Ok Done
---
[GitHub] flink pull request #5410: [FLINK-8468] [RabbitMQ Connector] Take advantage o...
Posted by pduveau <gi...@git.apache.org>.
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 <IN> The type of the data used by the sink.
+ */
+public interface RMQSinkPublishOptions<IN> {
--- 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...
Posted by GJL <gi...@git.apache.org>.
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<String> 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<String> 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<String> 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<String> 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<String> {
--- End diff --
nit: `serialVersionUID` is missing.
---
[GitHub] flink pull request #5410: [FLINK-8468] [RabbitMQ Connector] Take advantage o...
Posted by GJL <gi...@git.apache.org>.
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...
Posted by GJL <gi...@git.apache.org>.
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 <IN> The type of the data used by the sink.
+ */
+public interface RMQSinkPublishOptions<IN> {
--- End diff --
I think the interface should be annotated with `@PublicEvolving`.
---
[GitHub] flink pull request #5410: [FLINK-8468] [RabbitMQ Connector] Take advantage o...
Posted by GJL <gi...@git.apache.org>.
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...
Posted by GJL <gi...@git.apache.org>.
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 <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
+ */
+ 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[])}`
---