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

docs/en/connector-v2/Error-Quick-Reference-Manual.md

@@ -54,6 +54,20 @@ This document records some common error codes and corresponding solutions of Sea
 | SLACK-01  | Conversation can not be founded in channels | When users encounter this error code, it means that the channel is not existed in slack workspace, please check it |
 | SLACK-02  | Write to slack channel failed               | When users encounter this error code, it means that slack has some problems, please check it whether is work       |
 
+## Rabbitmq Connector Error Codes
+
+| code        | description                                                   | solution                                                                                                        |
+|-------------|---------------------------------------------------------------|-----------------------------------------------------------------------------------------------------------------|
+| RABBITMQ-01 | handle queue consumer shutdown signal failed                  | When users encounter this error code, it means that job has some problems, please check it whether is work well |
+| RABBITMQ-02 | create rabbitmq client failed                                 | When users encounter this error code, it means that rabbitmq has some problems, please check it whether is work |
+| RABBITMQ-03 | close connection failed                                       | When users encounter this error code, it means that rabbitmq has some problems, please check it whether is work |
+| RABBITMQ-04 | send messages failed                                          | When users encounter this error code, it means that rabbitmq has some problems, please check it whether is work |
+| RABBITMQ-05 | messages could not be acknowledged during checkpoint creation | When users encounter this error code, it means that job has some problems, please check it whether is work well |
+| RABBITMQ-06 | messages could not be acknowledged with basicReject           | When users encounter this error code, it means that job has some problems, please check it whether is work well |
+| RABBITMQ-07 | parse uri failed                                              | When users encounter this error code, it means that rabbitmq connect uri incorrect, please check it             |
+| RABBITMQ-08 | initialize ssl context failed                                 | When users encounter this error code, it means that rabbitmq has some problems, please check it whether is work |
+| RABBITMQ-09 | setup ssl factory failed                                      | When users encounter this error code, it means that rabbitmq has some problems, please check it whether is work |
+
 ## Socket Connector Error Codes
 
 | code      | description                                              | solution                                                                                                                       |

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

@@ -0,0 +1,106 @@
+# 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        | -             |
+| common-options              |         | 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 [int]
+
+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
+
+### common options
+
+Sink plugin common parameters, please refer to [Sink Common Options](common-options.md) for details
+
+## Example
+
+simple:
+
+```hocon
+sink {
+      RabbitMQ {
+          host = "rabbitmq-e2e"
+          port = 5672
+          virtual_host = "/"
+          username = "guest"
+          password = "guest"
+          queue_name = "test1"
+      }
+}
+```
+
+## Changelog
+
+### next version
+
+- Add Rabbitmq Sink Connector

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

@@ -0,0 +1,158 @@
+# 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       | -             |
+| common-options              |         | 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 [int]
+
+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
+
+### common options
+
+Source plugin common parameters, please refer to [Source Common Options](common-options.md) for details
+
+## 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>"
+            }
+        }
+    }
+}
+```
+
+## Changelog
+
+### next version
+
+- Add Rabbitmq source Connector

plugin-mapping.properties

@@ -154,3 +154,5 @@ seatunnel.sink.Slack = connector-slack
 seatunnel.source.OneSignal = connector-http-onesignal
 seatunnel.source.Jira = connector-http-jira
 seatunnel.source.Gitlab = connector-http-gitlab
+seatunnel.sink.RabbitMQ = connector-rabbitmq
+seatunnel.source.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>