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.
---
