[Feature][Connector-V2] rabbitmq source and sink connector

docs/en/connector-v2/sink/Rabbitmq.md

@@ -0,0 +1,95 @@
+# Rabbitmq
+
+> Rabbitmq sink connector
+
+## Description
+
+Used to write data to Rabbitmq.
+
+## Key features
+
+- [ ] [exactly-once](../../concept/connector-v2-features.md)
+- [ ] [schema projection](../../concept/connector-v2-features.md)
+
+##  Options
+
+| name                        | type    | required | default value |
+|-----------------------------|---------|----------|---------------|
+| host                        | string  | yes      | -             |
+| port                        | int     | yes      | -             |
+| virtual_host                | string  | yes      | -             |
+| username                    | string  | yes      | -             |
+| password                    | string  | yes      | -             |
+| queue_name                  | string  | yes      | -             |
+| url                         | string  | no       | -             |
+| network_recovery_interval   | int     | no       | -             |
+| topology_recovery_enabled   | boolean | no       | -             |
+| automatic_recovery_enabled  | boolean | no       | -             |
+| connection_timeout          | int     | no       | -             |
+
+### host [string]
+
+the default host to use for connections
+
+### port [int]
+
+the default port to use for connections
+
+### virtual_host [string]
+
+virtual host – the virtual host to use when connecting to the broker
+
+### username [string]
+
+the AMQP user name to use when connecting to the broker
+
+### password [string]
+
+the password to use when connecting to the broker
+
+### url [string]
+
+convenience method for setting the fields in an AMQP URI: host, port, username, password and virtual host
+
+### queue_name [string]
+
+the queue to write the message to
+
+### schema [Config]
+
+#### fields [Config]
+
+the schema fields of upstream data.
+
+### network_recovery_interval [string]
+
+how long will automatic recovery wait before attempting to reconnect, in ms
+
+### topology_recovery [string]
+
+if true, enables topology recovery
+
+### automatic_recovery [string]
+
+if true, enables connection recovery
+
+### connection_timeout [int]
+
+connection TCP establishment timeout in milliseconds; zero for infinite
+
+## Example
+
+simple:
+
+```hocon
+sink {
+      RabbitMQ {
+          host = "rabbitmq-e2e"
+          port = 5672
+          virtual_host = "/"
+          username = "guest"
+          password = "guest"
+          queue_name = "test1"
+      }
+}
+```

docs/en/connector-v2/source/Rabbitmq.md

@@ -0,0 +1,148 @@
+# Rabbitmq
+
+> Rabbitmq source connector
+
+## Description
+
+Used to read data from Rabbitmq.
+
+## Key features
+
+- [ ] [batch](../../concept/connector-v2-features.md)
+- [x] [stream](../../concept/connector-v2-features.md)
+- [x] [exactly-once](../../concept/connector-v2-features.md)
+- [x] [schema projection](../../concept/connector-v2-features.md)
+- [ ] [parallelism](../../concept/connector-v2-features.md)
+- [ ] [support user-defined split](../../concept/connector-v2-features.md)
+
+:::tip
+The source must be non-parallel (parallelism set to 1) in order to achieve exactly-once. This limitation is mainly due to RabbitMQ’s approach to dispatching messages from a single queue to multiple consumers.
+
+##  Options
+
+| name                        | type    | required | default value |
+|-----------------------------|---------|----------|---------------|
+| host                        | string  | yes      | -             |
+| port                        | int     | yes      | -             |
+| virtual_host                | string  | yes      | -             |
+| username                    | string  | yes      | -             |
+| password                    | string  | yes      | -             |
+| queue_name                  | string  | yes      | -             |
+| schema                      | config  | yes      | -             |
+| url                         | string  | no       | -             |
+| routing_key                 | string  | no       | -             |
+| exchange                    | string  | no       | -             |
+| network_recovery_interval   | int     | no       | -             |
+| topology_recovery_enabled   | boolean | no       | -             |
+| automatic_recovery_enabled  | boolean | no       | -             |
+| connection_timeout          | int     | no       | -             |
+| requested_channel_max       | int     | no       | -             |
+| requested_frame_max         | int     | no       | -             |
+| requested_heartbeat         | int     | no       | -             |
+| prefetch_count              | int     | no       | -             |
+| delivery_timeout            | long    | no       | -             |
+
+
+### host [string]
+
+the default host to use for connections
+
+### port [int]
+
+the default port to use for connections
+
+### virtual_host [string]
+
+virtual host – the virtual host to use when connecting to the broker
+
+### username [string]
+
+the AMQP user name to use when connecting to the broker
+
+### password [string]
+
+the password to use when connecting to the broker
+
+### url [string]
+
+convenience method for setting the fields in an AMQP URI: host, port, username, password and virtual host
+
+### queue_name [string]
+
+the queue to publish the message to
+
+### routing_key [string]
+
+the routing key to publish the message to
+
+### exchange [string]
+
+the exchange to publish the message to
+
+### schema [Config]
+
+#### fields [Config]
+
+the schema fields of upstream data.
+
+### network_recovery_interval [string]
+
+how long will automatic recovery wait before attempting to reconnect, in ms
+
+### topology_recovery [string]
+
+if true, enables topology recovery
+
+### automatic_recovery [string]
+
+if true, enables connection recovery
+
+### connection_timeout [int]
+
+connection tcp establishment timeout in milliseconds; zero for infinite
+
+### requested_channel_max [int]
+
+initially requested maximum channel number; zero for unlimited
+**Note: Note the value must be between 0 and 65535 (unsigned short in AMQP 0-9-1).
+
+### requested_frame_max [int]
+
+the requested maximum frame size
+
+### requested_heartbeat [int]
+
+Set the requested heartbeat timeout
+**Note: Note the value must be between 0 and 65535 (unsigned short in AMQP 0-9-1).
+
+### prefetch_count [int]
+
+prefetchCount the max number of messages to receive without acknowledgement
+
+### delivery_timeout [long]
+
+deliveryTimeout maximum wait time, in milliseconds, for the next message delivery
+
+## Example
+
+simple:
+
+```hocon
+source {
+    RabbitMQ {
+        host = "rabbitmq-e2e"
+        port = 5672
+        virtual_host = "/"
+        username = "guest"
+        password = "guest"
+        queue_name = "test"
+        schema = {
+            fields {
+            id = bigint
+            c_map = "map<string, smallint>"
+            c_array = "array<tinyint>"
+            }
+        }
+    }
+}
+```

plugin-mapping.properties

@@ -139,3 +139,6 @@ seatunnel.source.S3File = connector-file-s3
 seatunnel.sink.S3File = connector-file-s3
 seatunnel.source.Amazondynamodb = connector-amazondynamodb
 seatunnel.sink.Amazondynamodb = connector-amazondynamodb
+seatunnel.source.RabbitMQ = connector-rabbitmq
+seatunnel.sink.RabbitMQ = connector-rabbitmq
+

seatunnel-connectors-v2/connector-rabbitmq/pom.xml

@@ -0,0 +1,52 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<!--
+
+    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.
+
+-->
+<project xmlns="http://maven.apache.org/POM/4.0.0"
+         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+    <parent>
+        <artifactId>seatunnel-connectors-v2</artifactId>
+        <groupId>org.apache.seatunnel</groupId>
+        <version>${revision}</version>
+    </parent>
+    <modelVersion>4.0.0</modelVersion>
+
+    <artifactId>connector-rabbitmq</artifactId>
+
+    <properties>
+        <rabbitmq.version>5.9.0</rabbitmq.version>
+    </properties>
+    <dependencies>
+        <dependency>
+            <groupId>org.apache.seatunnel</groupId>
+            <artifactId>connector-common</artifactId>
+            <version>${project.version}</version>
+        </dependency>
+        <dependency>
+            <groupId>com.rabbitmq</groupId>
+            <artifactId>amqp-client</artifactId>
+            <version>${rabbitmq.version}</version>
+        </dependency>
+        <dependency>
+            <groupId>org.apache.seatunnel</groupId>
+            <artifactId>seatunnel-format-json</artifactId>
+            <version>${project.version}</version>
+        </dependency>
+    </dependencies>
+</project>

seatunnel-connectors-v2/connector-rabbitmq/src/main/java/org/apache/seatunnel/connectors/seatunnel/rabbitmq/client/QueueingConsumer.java

@@ -0,0 +1,89 @@
+/*
+ * 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.seatunnel.connectors.seatunnel.rabbitmq.client;
+
+import org.apache.seatunnel.common.Handover;
+
+import com.rabbitmq.client.AMQP;
+import com.rabbitmq.client.Channel;
+import com.rabbitmq.client.ConsumerCancelledException;
+import com.rabbitmq.client.DefaultConsumer;
+import com.rabbitmq.client.Delivery;
+import com.rabbitmq.client.Envelope;
+import com.rabbitmq.client.ShutdownSignalException;
+import com.rabbitmq.utility.Utility;
+import lombok.SneakyThrows;
+import lombok.extern.slf4j.Slf4j;
+
+import java.io.IOException;
+
+@Slf4j
+public class QueueingConsumer extends DefaultConsumer {
+    private final Handover<Delivery> handover;
+
+    // When this is non-null the queue is in shutdown mode and nextDelivery should
+    // throw a shutdown signal exception.
+    private volatile ShutdownSignalException shutdown;
+    private volatile ConsumerCancelledException cancelled;
+
+    private static final Delivery POISON = new Delivery(null, null, null);
+
+    public QueueingConsumer(Channel channel, Handover<Delivery> handover) {
+        this(channel, Integer.MAX_VALUE, handover);
+    }
+
+    public QueueingConsumer(Channel channel, int capacity, Handover<Delivery> handover) {
+        super(channel);
+        this.handover = handover;
+    }
+
+    private void checkShutdown() {
+        if (shutdown != null) {
+            throw Utility.fixStackTrace(shutdown);
+        }
+    }
+
+    @Override
+    public void handleShutdownSignal(String consumerTag, ShutdownSignalException sig) {
+        shutdown = sig;
+        try {
+            handover.produce(POISON);
+        } catch (InterruptedException e) {
+            throw new RuntimeException(e);
+        } catch (Handover.ClosedException e) {
+            throw new RuntimeException(e);
+        }
+    }
+
+    @SneakyThrows
+    @Override
+    public void handleCancel(String consumerTag) throws IOException {
+        cancelled = new ConsumerCancelledException();
+        handover.produce(POISON);
+    }
+
+    @SneakyThrows
+    @Override
+    public void handleDelivery(
+            String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body)
+            throws IOException {
+        checkShutdown();
+        log.info(new String(body));
+        handover.produce(new Delivery(envelope, properties, body));
+    }
+}