[pulsar-io-rabbitmq] Added integration test for RabbitMQ. (#6033)

Fixes https://github.com/apache/pulsar/issues/5920

### Motivation

The current RabbitMQ connector lacks an integration test.

### Modifications

Adds an integration test for the RabbitMQ connector(both source and sink):
- A container definition using `rabbitmq:3.8-management` as suggested.
- A RabbitMQ source tester, which provides configurations for the RabbitMQ source connector and publishes messages to the RabbitMQ.
- A RabbitMQ sink tester, which provides configurations for the RabbitMQ sink connector and consumes messages from the RabbitMQ for verification.
- A test case that invokes the testers with the current test framework.


### Verifying this change

This change added tests and can be verified as follows:

*(example:)*
  - *Added integration tests for the RabbitMQ connector*

tests/integration/pom.xml

@@ -131,6 +131,12 @@
   	  <artifactId>elasticsearch-rest-high-level-client</artifactId>
   	</dependency>
 
+    <dependency>
+      <groupId>com.rabbitmq</groupId>
+      <artifactId>amqp-client</artifactId>
+      <version>${rabbitmq-client.version}</version>
+    </dependency>
+
     <dependency>
       <groupId>com.facebook.presto</groupId>
       <artifactId>presto-jdbc</artifactId>

tests/integration/src/test/java/org/apache/pulsar/tests/integration/containers/RabbitMQContainer.java

+/**
+ * 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.pulsar.tests.integration.containers;
+
+import org.testcontainers.containers.wait.strategy.HostPortWaitStrategy;
+
+public class RabbitMQContainer extends ChaosContainer<RabbitMQContainer> {
+    public static final String NAME = "RabbitMQ";
+    public static final Integer[] PORTS = { 5672 };
+    private static final String IMAGE_NAME = "rabbitmq:3.8-management";
+
+    private String networkAlias;
+
+    public RabbitMQContainer(String clusterName, String networkAlias) {
+        super(clusterName, IMAGE_NAME);
+        this.networkAlias = networkAlias;
+    }
+
+    @Override
+    protected void configure() {
+        super.configure();
+        this.withNetworkAliases(networkAlias)
+                .withExposedPorts(PORTS)
+                .withCreateContainerCmdModifier(createContainerCmd -> {
+                    createContainerCmd.withHostName(NAME);
+                    createContainerCmd.withName(clusterName + "-" + NAME);
+                })
+                .waitingFor(new HostPortWaitStrategy());
+    }
+}

tests/integration/src/test/java/org/apache/pulsar/tests/integration/functions/PulsarFunctionsTest.java

@@ -124,6 +124,12 @@ public abstract class PulsarFunctionsTest extends PulsarFunctionsTestBase {
         testSink(new ElasticSearchSinkTester(), true);
     }
 
+    @Test
+    public void testRabbitMQSink() throws Exception {
+        final String containerName = "rabbitmq-" + randomName(8);
+        testSink(new RabbitMQSinkTester(containerName), true, new RabbitMQSourceTester(containerName));
+    }
+
     @Test
     public void testDebeziumMySqlSource() throws Exception {
         testDebeziumMySqlConnect();

tests/integration/src/test/java/org/apache/pulsar/tests/integration/io/RabbitMQSinkTester.java

+/**
+ * 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.pulsar.tests.integration.io;
+
+import com.rabbitmq.client.AMQP;
+import com.rabbitmq.client.Channel;
+import com.rabbitmq.client.Connection;
+import com.rabbitmq.client.ConnectionFactory;
+import com.rabbitmq.client.DefaultConsumer;
+import com.rabbitmq.client.Envelope;
+import lombok.Data;
+import lombok.extern.slf4j.Slf4j;
+import org.apache.pulsar.tests.integration.containers.RabbitMQContainer;
+import org.apache.pulsar.tests.integration.topologies.PulsarCluster;
+
+import java.io.IOException;
+import java.util.Map;
+import java.util.concurrent.BlockingQueue;
+import java.util.concurrent.LinkedBlockingQueue;
+import java.util.concurrent.TimeoutException;
+
+import static org.testng.Assert.assertEquals;
+import static org.testng.Assert.fail;
+
+@Slf4j
+public class RabbitMQSinkTester extends SinkTester<RabbitMQContainer> {
+    private final String exchangeName = "test-sink-exchange";
+    private final String queueName = "test-sink-queue";
+    private final String keyName = "test-key";
+
+    public RabbitMQSinkTester(String networkAlias) {
+        super(networkAlias, SinkType.RABBITMQ);
+
+        sinkConfig.put("connectionName", "test-sink-connection");
+        sinkConfig.put("host", networkAlias);
+        sinkConfig.put("port", RabbitMQContainer.PORTS[0]);
+        sinkConfig.put("queueName", queueName);
+        sinkConfig.put("exchangeName", exchangeName);
+        sinkConfig.put("routingKey", keyName);
+    }
+
+    @Override
+    protected RabbitMQContainer createSinkService(PulsarCluster cluster) {
+        return new RabbitMQContainer(cluster.getClusterName(), networkAlias);
+    }
+
+    @Override
+    public void prepareSink() throws Exception {
+    }
+
+    static ConnectionFactory createConnectionFactory(RabbitMQContainer container) {
+        ConnectionFactory connectionFactory = new ConnectionFactory();
+        connectionFactory.setHost(container.getContainerIpAddress());
+        connectionFactory.setPort(container.getMappedPort(RabbitMQContainer.PORTS[0]));
+
+        return connectionFactory;
+    }
+
+    @Override
+    public void validateSinkResult(Map<String, String> kvs) {
+        ConnectionFactory connectionFactory = createConnectionFactory(serviceContainer);
+        try (Connection connection = connectionFactory.newConnection("rabbitmq-sink-tester");
+             Channel channel = connection.createChannel()) {
+            BlockingQueue<Record> records = new LinkedBlockingQueue<>();
+            channel.queueDeclare(queueName, true, false, false, null);
+            channel.basicConsume(queueName, new DefaultConsumer(channel) {
+                @Override
+                public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
+                    records.add(new Record(envelope.getRoutingKey(), body));
+                    channel.basicAck(envelope.getDeliveryTag(), false);
+                }
+            });
+            // keys are discards when  into rabbitMQ
+            for (String value : kvs.values()) {
+                try {
+                    Record record = records.take();
+                    assertEquals(record.key, keyName);
+                    assertEquals(new String(record.body), value);
+                } catch (InterruptedException e) {
+                    break;
+                }
+            }
+        } catch (TimeoutException | IOException e) {
+            fail("RabbitMQ Sink test failed", e);
+        }
+    }
+
+    @Data
+    private static class Record {
+        private final String key;
+        private final byte[] body;
+    }
+}

tests/integration/src/test/java/org/apache/pulsar/tests/integration/io/RabbitMQSourceTester.java

+/**
+ * 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.pulsar.tests.integration.io;
+
+import com.rabbitmq.client.BuiltinExchangeType;
+import com.rabbitmq.client.Channel;
+import com.rabbitmq.client.Connection;
+import com.rabbitmq.client.ConnectionFactory;
+import org.apache.pulsar.tests.integration.containers.RabbitMQContainer;
+
+import java.util.LinkedHashMap;
+import java.util.Map;
+
+import static org.apache.pulsar.tests.integration.io.RabbitMQSinkTester.createConnectionFactory;
+
+public class RabbitMQSourceTester extends SourceTester<RabbitMQContainer> {
+    private RabbitMQContainer serviceContainer;
+    private final String exchangeName = "test-src-exchange";
+    private final String queueName = "test-src-queue";
+
+    public RabbitMQSourceTester(String networkAlias) {
+        super("rabbitmq");
+
+        sourceConfig.put("connectionName", "test-source-connection");
+        sourceConfig.put("host", networkAlias);
+        sourceConfig.put("port", RabbitMQContainer.PORTS[0]);
+        sourceConfig.put("queueName", queueName);
+    }
+
+    @Override
+    public void setServiceContainer(RabbitMQContainer serviceContainer) {
+        this.serviceContainer = serviceContainer;
+    }
+
+    @Override
+    public void prepareSource() throws Exception {
+    }
+
+    @Override
+    public void prepareInsertEvent() throws Exception {
+        // pass
+    }
+
+    @Override
+    public void prepareDeleteEvent() throws Exception {
+        // pass
+    }
+
+    @Override
+    public void prepareUpdateEvent() throws Exception {
+        // pass
+    }
+
+    @Override
+    public Map<String, String> produceSourceMessages(int numMessages) throws Exception {
+        ConnectionFactory connectionFactory = createConnectionFactory(serviceContainer);
+        try (Connection connection = connectionFactory.newConnection("rabbitmq-source-tester");
+             Channel channel = connection.createChannel()) {
+            // the queue declaration has to be aligned with that in RabbitMQSource
+            channel.queueDeclare(queueName, false, false, false, null);
+            // use topic mode exchange in order to publish messages with any keys
+            channel.exchangeDeclare(exchangeName, BuiltinExchangeType.TOPIC);
+            channel.queueBind(queueName, exchangeName, "#");
+
+            Map<String, String> values = new LinkedHashMap<>();
+            for (int i = 0; i < numMessages; i++) {
+                String key = "rb-key-" + i;
+                String value = "rb-value-" + i;
+                values.put(key, value);
+                channel.basicPublish(exchangeName, key, null, value.getBytes());
+            }
+            return values;
+        }
+    }
+}

tests/integration/src/test/java/org/apache/pulsar/tests/integration/io/SinkTester.java

@@ -37,7 +37,8 @@ public abstract class SinkTester<ServiceContainerT extends GenericContainer> {
         KAFKA,
         JDBC,
         HDFS,
-        ELASTIC_SEARCH
+        ELASTIC_SEARCH,
+        RABBITMQ
     }
 
     protected final String networkAlias;