[Pulsar IO] Allow routing key per message to RabbitMQ sink connector

pulsar-io/rabbitmq/src/main/java/org/apache/pulsar/io/rabbitmq/RabbitMQAbstractConfig.java

@@ -82,12 +82,6 @@ public class RabbitMQAbstractConfig implements Serializable {
         help = "The password used to authenticate to RabbitMQ")
     private String password = "guest";
 
-    @FieldDoc(
-        required = true,
-        defaultValue = "",
-        help = "The RabbitMQ queue name from which messages should be read from or written to")
-    private String queueName;
-
     @FieldDoc(
         required = false,
         defaultValue = "0",
@@ -123,7 +117,6 @@ public void validate() {
         Preconditions.checkNotNull(port, "port property not set.");
         Preconditions.checkNotNull(virtualHost, "virtualHost property not set.");
         Preconditions.checkNotNull(connectionName, "connectionName property not set.");
-        Preconditions.checkNotNull(queueName, "queueName property not set.");
     }
 
     public ConnectionFactory createConnectionFactory() {

pulsar-io/rabbitmq/src/main/java/org/apache/pulsar/io/rabbitmq/RabbitMQSink.java

@@ -49,7 +49,6 @@ public class RabbitMQSink implements Sink<byte[]> {
     private Channel rabbitMQChannel;
     private RabbitMQSinkConfig rabbitMQSinkConfig;
     private String exchangeName;
-    private String routingKey;
 
     @Override
     public void open(Map<String, Object> config, SinkContext sinkContext) throws Exception {
@@ -64,21 +63,17 @@ public void open(Map<String, Object> config, SinkContext sinkContext) throws Exc
         );
 
         exchangeName = rabbitMQSinkConfig.getExchangeName();
-        routingKey = rabbitMQSinkConfig.getRoutingKey();
+        String exchangeType = rabbitMQSinkConfig.getExchangeType();
 
         rabbitMQChannel = rabbitMQConnection.createChannel();
-
-        // several clients share a queue
-        rabbitMQChannel.exchangeDeclare(exchangeName, BuiltinExchangeType.DIRECT, true);
-        rabbitMQChannel.queueDeclare(rabbitMQSinkConfig.getQueueName(), true, false, false, null);
-        rabbitMQChannel.queueBind(rabbitMQSinkConfig.getQueueName(), exchangeName, routingKey);
+        rabbitMQChannel.exchangeDeclare(exchangeName, exchangeType, true);
     }
 
     @Override
     public void write(Record<byte[]> record) {
         byte[] value = record.getValue();
         try {
-            rabbitMQChannel.basicPublish(exchangeName, routingKey, null, value);
+            rabbitMQChannel.basicPublish(exchangeName, record.getProperties().get("routingKey"), null, value);
             record.ack();
         } catch (IOException e) {
             record.fail();

pulsar-io/rabbitmq/src/main/java/org/apache/pulsar/io/rabbitmq/RabbitMQSinkConfig.java

@@ -51,10 +51,10 @@ public class RabbitMQSinkConfig extends RabbitMQAbstractConfig implements Serial
     private String exchangeName;
 
     @FieldDoc(
-        required = true,
-        defaultValue = "",
-        help = "The routing key used for publishing the messages")
-    private String routingKey;
+        required = false,
+        defaultValue = "topic",
+        help = "The exchange type to publish the messages on")
+    private String exchangeType = "topic";
 
     public static RabbitMQSinkConfig load(String yamlFile) throws IOException {
         ObjectMapper mapper = new ObjectMapper(new YAMLFactory());
@@ -70,6 +70,5 @@ public static RabbitMQSinkConfig load(Map<String, Object> map) throws IOExceptio
     public void validate() {
         super.validate();
         Preconditions.checkNotNull(exchangeName, "exchangeName property not set.");
-        Preconditions.checkNotNull(routingKey, "routingKey property not set.");
     }
 }

pulsar-io/rabbitmq/src/main/java/org/apache/pulsar/io/rabbitmq/RabbitMQSourceConfig.java

@@ -44,6 +44,12 @@ public class RabbitMQSourceConfig extends RabbitMQAbstractConfig implements Seri
 
     private static final long serialVersionUID = 1L;
 
+    @FieldDoc(
+        required = true,
+        defaultValue = "",
+        help = "The RabbitMQ queue name from which messages should be read from or written to")
+    private String queueName;
+
     @FieldDoc(
         required = false,
         defaultValue = "0",
@@ -69,6 +75,7 @@ public static RabbitMQSourceConfig load(Map<String, Object> map) throws IOExcept
     @Override
     public void validate() {
         super.validate();
+        Preconditions.checkNotNull(queueName, "queueName property not set.");
         Preconditions.checkArgument(prefetchCount >= 0, "prefetchCount must be non-negative.");
     }
 }

pulsar-io/rabbitmq/src/test/java/org/apache/pulsar/io/rabbitmq/sink/RabbitMQSinkConfigTest.java

@@ -44,15 +44,14 @@ public final void loadFromYamlFileTest() throws IOException {
         assertEquals("/", config.getVirtualHost());
         assertEquals("guest", config.getUsername());
         assertEquals("guest", config.getPassword());
-        assertEquals("test-queue", config.getQueueName());
         assertEquals("test-connection", config.getConnectionName());
         assertEquals(Integer.parseInt("0"), config.getRequestedChannelMax());
         assertEquals(Integer.parseInt("0"), config.getRequestedFrameMax());
         assertEquals(Integer.parseInt("60000"), config.getConnectionTimeout());
         assertEquals(Integer.parseInt("10000"), config.getHandshakeTimeout());
         assertEquals(Integer.parseInt("60"), config.getRequestedHeartbeat());
         assertEquals("test-exchange", config.getExchangeName());
-        assertEquals("test-key", config.getRoutingKey());
+        assertEquals("test-exchange-type", config.getExchangeType());
     }
 
     @Test
@@ -63,15 +62,14 @@ public final void loadFromMapTest() throws IOException {
         map.put("virtualHost", "/");
         map.put("username", "guest");
         map.put("password", "guest");
-        map.put("queueName", "test-queue");
         map.put("connectionName", "test-connection");
         map.put("requestedChannelMax", "0");
         map.put("requestedFrameMax", "0");
         map.put("connectionTimeout", "60000");
         map.put("handshakeTimeout", "10000");
         map.put("requestedHeartbeat", "60");
         map.put("exchangeName", "test-exchange");
-        map.put("routingKey", "test-key");
+        map.put("exchangeType", "test-exchange-type");
 
         RabbitMQSinkConfig config = RabbitMQSinkConfig.load(map);
         assertNotNull(config);
@@ -80,15 +78,14 @@ public final void loadFromMapTest() throws IOException {
         assertEquals("/", config.getVirtualHost());
         assertEquals("guest", config.getUsername());
         assertEquals("guest", config.getPassword());
-        assertEquals("test-queue", config.getQueueName());
         assertEquals("test-connection", config.getConnectionName());
         assertEquals(Integer.parseInt("0"), config.getRequestedChannelMax());
         assertEquals(Integer.parseInt("0"), config.getRequestedFrameMax());
         assertEquals(Integer.parseInt("60000"), config.getConnectionTimeout());
         assertEquals(Integer.parseInt("10000"), config.getHandshakeTimeout());
         assertEquals(Integer.parseInt("60"), config.getRequestedHeartbeat());
         assertEquals("test-exchange", config.getExchangeName());
-        assertEquals("test-key", config.getRoutingKey());
+        assertEquals("test-exchange-type", config.getExchangeType());
     }
 
     @Test
@@ -99,15 +96,14 @@ public final void validValidateTest() throws IOException {
         map.put("virtualHost", "/");
         map.put("username", "guest");
         map.put("password", "guest");
-        map.put("queueName", "test-queue");
         map.put("connectionName", "test-connection");
         map.put("requestedChannelMax", "0");
         map.put("requestedFrameMax", "0");
         map.put("connectionTimeout", "60000");
         map.put("handshakeTimeout", "10000");
         map.put("requestedHeartbeat", "60");
         map.put("exchangeName", "test-exchange");
-        map.put("routingKey", "test-key");
+        map.put("exchangeType", "test-exchange-type");
 
         RabbitMQSinkConfig config = RabbitMQSinkConfig.load(map);
         config.validate();
@@ -122,14 +118,13 @@ public final void missingExchangeValidateTest() throws IOException {
         map.put("virtualHost", "/");
         map.put("username", "guest");
         map.put("password", "guest");
-        map.put("queueName", "test-queue");
         map.put("connectionName", "test-connection");
         map.put("requestedChannelMax", "0");
         map.put("requestedFrameMax", "0");
         map.put("connectionTimeout", "60000");
         map.put("handshakeTimeout", "10000");
         map.put("requestedHeartbeat", "60");
-        map.put("routingKey", "test-key");
+        map.put("exchangeType", "test-exchange-type");
 
         RabbitMQSinkConfig config = RabbitMQSinkConfig.load(map);
         config.validate();

pulsar-io/rabbitmq/src/test/java/org/apache/pulsar/io/rabbitmq/sink/RabbitMQSinkTest.java

@@ -53,29 +53,28 @@ public void TestOpenAndWriteSink() throws Exception {
         configs.put("virtualHost", "default");
         configs.put("username", "guest");
         configs.put("password", "guest");
-        configs.put("queueName", "test-queue");
         configs.put("connectionName", "test-connection");
         configs.put("requestedChannelMax", "0");
         configs.put("requestedFrameMax", "0");
         configs.put("connectionTimeout", "60000");
         configs.put("handshakeTimeout", "10000");
         configs.put("requestedHeartbeat", "60");
         configs.put("exchangeName", "test-exchange");
-        configs.put("routingKey", "test-key");
+        configs.put("exchangeType", "fanout");
 
         RabbitMQSink sink = new RabbitMQSink();
 
         // open should success
         sink.open(configs, null);
 
         // write should success
-        Record<byte[]> record = build("test-topic", "fakeKey", "fakeValue");
+        Record<byte[]> record = build("test-topic", "fakeKey", "fakeValue", "fakeRoutingKey");
         sink.write(record);
 
         sink.close();
     }
 
-    private Record<byte[]> build(String topic, String key, String value) {
+    private Record<byte[]> build(String topic, String key, String value, String routingKey) {
         // prepare a SinkRecord
         SinkRecord<byte[]> record = new SinkRecord<>(new Record<byte[]>() {
             @Override
@@ -96,6 +95,15 @@ public Optional<String> getDestinationTopic() {
                     return Optional.empty();
                 }
             }
+
+            @Override
+            public Map<String, String> getProperties() {
+                return new HashMap<String, String>() {
+                    {
+                        put("routingKey", routingKey);
+                    }
+                };
+            }
         }, value.getBytes(StandardCharsets.UTF_8));
         return record;
     }

pulsar-io/rabbitmq/src/test/resources/sinkConfig.yaml

@@ -23,14 +23,12 @@
 "virtualHost": "/",
 "username": "guest",
 "password": "guest",
-"queueName": "test-queue",
 "connectionName": "test-connection",
 "requestedChannelMax": "0",
 "requestedFrameMax": "0",
 "connectionTimeout": "60000",
 "handshakeTimeout": "10000",
 "requestedHeartbeat": "60",
 "exchangeName": "test-exchange",
-"routingKey": "test-key"
-
+"exchangeType": "test-exchange-type"
 }