<a id="rabbit-m-q-sink-connector"></a>

# RabbitMQ Sink Connector for Confluent Platform

The Kafka Connect RabbitMQ Sink connector integrates with RabbitMQ servers,
using the Advanced Message Queuing Protocol (AMQP) protocol. The RabbitMQ Sink
connector reads data from one or more Apache Kafka® topics and sends the data to a
RabbitMQ exchange.

#### IMPORTANT
<!-- WARNING: THIS IS A SHARED FILE AND THE SOURCE IS LOCATED IN DOCS-COMMON. DO NOT ADD TO ANY OTHER REPO. -->

Effective July 6, 2025, only self-managed connector versions that meet or exceed the minimum version listed on the
[Supported Connector Versions](https://docs.confluent.io/platform/7.8/connect/supported-connector-version.html#supported-connector-versions-till-cp-7-8)
page receive support from Confluent. Older, unsupported connector versions have been removed from Confluent Marketplace and
are no longer available for download.

## Features

The RabbitMQ Sink connector includes the following features:

- [At least once delivery](#rabbitmq-sink-at-least-once-delivery)
- [Dead Letter Queue](#rabbitmq-sink-dead-letter-queue)
- [Multiple tasks](#rabbitmq-sink-multiple-tasks)
- [Header forwarding](#rabbitmq-sink-header-forwarding)
- [Delivery to RabbitMQ exchange](#rabbitmq-sink-delivery-exchange)
- [Topic-to-queue map routing](#rabbitmq-sink-topic-queue-map)
- [Publishing bytes as payload](#rabbitmq-sink-publishing-bytes-as-payload)
- [SSL/TLS security](#rabbitmq-sink-ssl-security)
- [Record batching](#rabbitmq-sink-record-batching)

<a id="rabbitmq-sink-at-least-once-delivery"></a>

### At least once delivery

This connector guarantees that records from the Kafka topic are delivered at
least once.

<a id="rabbitmq-sink-dead-letter-queue"></a>

### Dead Letter Queue

This connector supports the Dead Letter Queue (DLQ) functionality. For
information about accessing and using the DLQ, see [Confluent Platform
Dead Letter Queue](/platform/current/connect/concepts.html#dead-letter-queue).

<a id="rabbitmq-sink-multiple-tasks"></a>

### Multiple tasks

The RabbitMQ Sink connector supports running one or more tasks. You can specify
the number of tasks in the `tasks.max` configuration parameter. This can lead
to performance gains when multiple files need to be parsed.

<a id="rabbitmq-sink-header-forwarding"></a>

### Header forwarding

The connector supports forwarding Kafka headers and metadata to the RabbitMQ
message as headers. The Kafka message key can also be forwarded as the
`correlationID` on the RabbitMQ message.

<a id="rabbitmq-sink-delivery-exchange"></a>

### Delivery to RabbitMQ exchange

The connector supports delivering to one configured RabbitMQ exchange. When
you specify more than one Kafka topic to read from, records from unmapped
topics are all delivered to this one RabbitMQ exchange with the configured
`rabbitmq.routing.key`. For per-topic routing to different destinations, see
[Topic-to-queue map routing](#rabbitmq-sink-topic-queue-map).

<a id="rabbitmq-sink-topic-queue-map"></a>

### Topic-to-queue map routing

Use `rabbitmq.topic.queue.map` to publish records from a specific Kafka topic
directly to a specific RabbitMQ queue, without running a separate connector
instance per topic. The map is a JSON object of the form
`{"topic-a":"queue-1","topic-b":"queue-2"}`. Records from a mapped topic are
published through the RabbitMQ default exchange. Unmapped topics fall
back to `rabbitmq.exchange` with `rabbitmq.routing.key`.

#### IMPORTANT
The connector verifies mapped queue existence only during configuration
submission. If a mapped queue is deleted on the broker while the
connector is running, the connector publishes subsequent records for
that topic to the default RabbitMQ exchange with a routing key that
no longer matches any queue, causing the broker to silently drop the
records.

<a id="rabbitmq-sink-publishing-bytes-as-payload"></a>

### Publishing bytes as payload

The RabbitMQ message supports publishing bytes as payload. The connector
supports storing raw bytes in RabbitMQ using the `value.converter` as
`org.apache.kafka.connect.converters.ByteArrayConverter`. Use the
`ByteArrayConverter` to store the binary serialized form (for example, JSON,
Avro, Strings, etc.) of the Kafka record values in RabbitMQ as byte arrays.
Applications accessing these values can then read this information from RabbitMQ
and deserialize the bytes into a usable form. If your data in Kafka is not in the
format you want to persist in RabbitMQ, consider using a [Single
Message Transformation](/platform/current/connect/transforms/index.html) to change records before
they are sent to RabbitMQ.

<a id="rabbitmq-sink-ssl-security"></a>

### SSL/TLS security

The connector also supports SSL/TLS security to connect to the RabbitMQ server.

<a id="rabbitmq-sink-record-batching"></a>

### Record batching

The connector batches the records from Kafka while publishing to RabbitMQ. This
is controlled by `rabbitmq.publish.max.batch.size` configuration. Please note
this does not supersede the `consumer.max.poll.records` configuration, and in
effect will always be smaller than that. Please also note that the
configurations prefixed by `rabbitmq.publish` can, in tandem, influence
throughput.

## License

You can use this connector for a 30-day trial period without a license key.

After 30 days, you must purchase a connector subscription which includes [Confluent enterprise license](/platform/current/installation/license.html#enterprise-subscription-license) keys to subscribers, along with [enterprise-level support](https://www.confluent.io/subscription/) for Confluent Platform and your connectors. If you are a subscriber, you can contact [Confluent Support](https://support.confluent.io/) for more information.

For license properties, see [Confluent Platform license](config.md#rabbitmq-sink-connector-license-config). For
information about the license topic, see
[License topic configuration](config.md#rabbit-m-q-sink-license-topic-configuration).

## Linux on IBM Z (s390x) support

Starting with Confluent Platform 8.2, this connector supports Linux on IBM Z (s390x).
The connector supports the same capability available on x86_64 unless
otherwise noted. For more information, see [Linux on IBM Z (s390x) support](https://docs.confluent.io/platform/current/installation/versions-interoperability.html#linux-on-ibm-z-s390x-support).

## Configuration Properties

For a complete list of configuration properties for this connector, see
[Configuration Reference for RabbitMQ Sink Connector for Confluent Platform](config.md#rabbit-m-q-sink-connector-config).

For an example of how to get Kafka Connect connected to [Confluent Cloud](/cloud/current/index.html), see
[Connect Self-Managed Kafka Connect to Confluent Cloud](/cloud/current/cp-component/connect-cloud-config.html#distributed-cluster).

## Install the RabbitMQ Sink Connector

You can install this connector by using the [confluent connect
plugin
install](https://docs.confluent.io/confluent-cli/current/command-reference/connect/plugin/confluent_connect_plugin_install.html)
command, or by manually downloading the ZIP file.

### Prerequisites

#### NOTE
- You must install the connector on every machine where Connect will run.
- This connector is based on the AMQP 0-9-1 protocol, so it may work with
  other servers that implement this protocol.

- Kafka Broker: Confluent Platform 3.3.0 or later, or Kafka 0.11.0 or later.
- Connect: Confluent Platform 4.1.0 or later, or Kafka 1.1.0 or later (requires header
  support in Connect).
- Java 1.8.
- RabbitMQ Server version 3.x.
- An installation of the latest (`latest`) connector version.

  To install the `latest` connector version, navigate to your Confluent Platform
  installation directory and run the following command:
  ```bash
  confluent connect plugin install confluentinc/kafka-connect-rabbitmq-sink:latest
  ```

  You can install a specific version by replacing `latest` with a version
  number as shown in the following example:
  ```bash
  confluent connect plugin install confluentinc/kafka-connect-rabbitmq-sink:1.8.0
  ```

### Install the connector manually

[Download and extract the ZIP file](https://www.confluent.io/hub/confluentinc/kafka-connect-rabbitmq-sink) for
your connector and then follow the manual connector installation
[instructions](/kafka-connectors/self-managed/userguide.html#connect-installing-plugins).

## Usage Notes

<a id="rabbit-m-q-sink-features"></a>

## Examples

### Property-based example

This configuration is typically used with [standalone
workers](/platform/current/connect/concepts.html#standalone-workers).

```bash
 name=RabbitMQSinkConnector
 connector.class=io.confluent.connect.rabbitmq.sink.RabbitMQSinkConnector
 tasks.max=1
 confluent.topic.bootstrap.servers=< Required Configuration >
 topics=< Required Configuration >
 rabbitmq.host=< Required Configuration >
 rabbitmq.port=< Required Configuration >
 rabbitmq.username=< Required Configuration >
 rabbitmq.password=< Required Configuration >
 rabbitmq.exchange=< Required Configuration >
 rabbitmq.routing.key=< Required Configuration >
 rabbitmq.delivery.mode=< Required Configuration >
 key.converter=< Required Configuration >
 value.converter==< Required Configuration >
```

### REST-based example

This configuration is typically used with [distributed
workers](/platform/current/connect/concepts.html#distributed-workers). Write the following JSON to
`connector.json`, configure all of the required values, and use the command
below to post the configuration to one or more distributed connect workers.
Check here for more information about the Kafka Connect [Kafka
Connect REST Interface](/platform/current/connect/references/restapi.html)

#### Connect Distributed REST example

```json
 {
   "name" : "RabbitMQSinkConnector",
   "config" : {
     "connector.class" : "io.confluent.connect.rabbitmq.sink.RabbitMQSinkConnector",
     "tasks.max" : "1",
     "confluent.topic.bootstrap.servers" : "< Required Configuration >",
     "topics" : "< Required Configuration >",
     "rabbitmq.host" : "< Required Configuration >",
     "rabbitmq.port" : "< Required Configuration >",
     "rabbitmq.username" : "< Required Configuration >",
     "rabbitmq.password" : "< Required Configuration >",
     "rabbitmq.exchange" : "< Required Configuration >",
     "rabbitmq.routing.key" : "< Required Configuration >",
     "rabbitmq.delivery.mode" : "< Required Configuration >",
     "key.converter" : "< Required Configuration >",
     "value.converter" : "< Required Configuration >"
   }
 }
```

Use curl to post the configuration to the Connect worker. Change
`http://localhost:8083/`` the endpoint of one of your Kafka Connect
worker(s).

#### Create a new connector

```bash
curl -s -X POST -H 'Content-Type: application/json' --data @connector.json http://localhost:8083/connectors
```

#### Update an existing Connector

```bash
curl -s -X PUT -H 'Content-Type: application/json' --data @connector.json http://localhost:8083/connectors/RabbitMQSinkConnector1/config
```

<a id="rabbit-m-q-sink-quickstart"></a>

## Quick Start

The RabbitMQ Sink connector streams records from Kafka topics to a RabbitMQ
exchange with high throughput. This quick start shows example data production
and consumption setups in detail.

1. Start the [RabbitMQ Server](https://www.rabbitmq.com/download.html) broker, specifying the docker image on basis of required RabbitMQ version.
   ```bash
   docker run -it --rm --name rabbitmq \
       -p 5672:5672 \
       -p 15672:15672 \
       rabbitmq:3.8.4-management
   ```
2. Create a RabbitMQ exchange. To produce messages from Kafka to RabbitMQ, you also create a queue and binding.
   1. Once the RabbitMQ docker container has started, navigate to [http://localhost:15672](http://localhost:15672) in your browser and login with `guest`/`guest`.
   2. In the `Exchanges` tab click on `Add a new exchange`. Name it `exchange1` and leave other options as the default settings.
   3. In the `Queues` tab click on `Add a new queue`. Name it `queue1` and leave other options as the default settings.
   4. In the `Exchanges` tab click on the exchange created `exchange1`. In the `Bindings` section add a binding in the field `To queue` to `queue1` with routing key `rkey1`.
3. Install the connector through the [Confluent Hub Client](/kafka-connectors/self-managed/confluent-hub/client.html).
   ```bash
   # run from your Confluent Platform installation directory
   confluent connect plugin install confluentinc/kafka-connect-rabbitmq-sink:latest
   ```
4. Start Confluent Platform.
   ```bash
   confluent local start
   ```
5. [Produce](https://docs.confluent.io/current/cli/command-reference/confluent-produce.html) test data to a pre-created `rabbitmq-messages` topic in Kafka.
   ```bash
   seq 10 | confluent local produce rabbitmq-messages
   ```
6. Create a `rabbitmq-sink.json` file with the following contents:
   ```json
   {
     "name": "RabbitMQSinkConnector",
     "config": {
       "connector.class": "io.confluent.connect.rabbitmq.sink.RabbitMQSinkConnector",
       "tasks.max": "1",
       "topics": "rabbitmq-messages",
       "key.converter": "org.apache.kafka.connect.storage.StringConverter",
       "value.converter": "org.apache.kafka.connect.converters.ByteArrayConverter",
       "confluent.topic.bootstrap.servers": "localhost:9092",
       "confluent.topic.replication.factor": "1",
       "rabbitmq.host": "localhost",
       "rabbitmq.port": "5672",
       "rabbitmq.username": "guest",
       "rabbitmq.password": "guest",
       "rabbitmq.exchange": "exchange1",
       "rabbitmq.routing.key": "rkey1",
       "rabbitmq.delivery.mode": "PERSISTENT"
     }
   }
   ```
7. Load the RabbitMQ Sink connector.
   ```bash
   confluent local load RabbitMQSinkConnector --config rabbitmq-sink.json
   ```

   #### IMPORTANT
   Don’t use the [Confluent CLI](https://docs.confluent.io/confluent-cli/current/index.html) in production environments.
8. Confirm that the connector is in a `RUNNING` state.
   ```bash
   confluent local status RabbitMQSinkConnector
   ```
9. Navigate to the [RabbitMQ UI](http://localhost:15672) to confirm the messages were delivered to the `queue1` queue.