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

# RabbitMQ Source Connector for Confluent Platform

The Kafka Connect RabbitMQ Source connector integrates with RabbitMQ servers,
using the AMQP protocol. The RabbitMQ Source connector reads data from a
RabbitMQ queue or topic and persists the data in an Apache Kafka® topic.

The connector reads Base64 encoded data from RabbitMQ and then writes the data
to Kafka. The connector can’t infer the schema from this Base64 encoded string,
so it writes the same binary data with a bytes Avro schema to the Kafka topic.
Since the schema does not contain any record name, `RecordNameStrategy` and
`TopicRecordNameStrategy` can’t be used with this connector.

#### NOTE
- This connector is based on the AMQP protocol so it may work with other
  servers that implement this protocol.
- <!-- 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 Source connector includes the following features:

- [At least once delivery](#rabbitmq-source-at-least-once-delivery)
- [Multiple tasks](#rabbitmq-source-multiple-tasks)
- [Queue-to-topic map routing](#rabbitmq-source-queue-topic-map)
- [Per-queue failure isolation](#rabbitmq-source-failure-isolation)

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

### At least once delivery

This connector guarantees that records are delivered at least once to the Kafka
topic. If the connector restarts, there may be some duplicate
records in the Kafka topic.

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

### Multiple tasks

The RabbitMQ Source connector supports running one or more tasks. By default
(`rabbitmq.queue.consumer.mode=dedicated`), the `N` queues in
`rabbitmq.queue` are split into `ceil(N / rabbitmq.max.queue.per.task)`
round-robin chunks (`rabbitmq.max.queue.per.task` defaults to `10`) and
one chunk is distributed per task. Each queue is owned by exactly one task,
preserving per-queue message ordering. `tasks.max` must meet
`ceil(N / rabbitmq.max.queue.per.task) ≤ tasks.max ≤ N`.

Set `rabbitmq.queue.consumer.mode=shared` to opt out of task partitioning
when hot queues exist and you want their load spread across multiple tasks.
In `shared` mode every task subscribes to every queue as an AMQP
competing consumer, and per-queue message ordering is not preserved. This
mode is best suited to small queue counts. Because each task opens one AMQP
channel per queue on its connection, high queue counts can exceed the
RabbitMQ broker’s `channel_max` limit, causing subscription failures.

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

### Queue-to-topic map routing

Use `rabbitmq.queue.topic.map` to route messages from a specific queue to a
specific Kafka topic without running a separate connector instance for each
queue. The map is a JSON object of the form
`{"queue-1":"topic-a","queue-2":"topic-b"}`. Queues not listed (or when the
map is empty) fall back to the default `kafka.topic`.

<a id="rabbitmq-source-failure-isolation"></a>

### Per-queue failure isolation

Use `rabbitmq.queue.failure.tolerance` to control how the connector reacts
when a queue in `rabbitmq.queue` can no longer be subscribed to while the
connector is running (for example, an administrator deletes the queue on the
broker, permissions are revoked, or the queue is locked by another
connection) and the task attempts to resubscribe on the next channel
recreation. `fail-fast` (default): the connector transitions to `FAILED`
on the first such queue. `skip-and-continue`: the connector skips the
affected queues and continues consuming from the rest.

## Install the RabbitMQ Source 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.

#### NOTE
`kafka-connect-rabbitmq` contains only the source connector.

### Prerequisites

- You must install the connector on every machine where Connect will run.
- 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: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:1.1.1
  ```

### Install the connector manually

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

## 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-source-connector-license-config). For
information about the license topic, see
[License topic configuration](config.md#rabbit-m-q-source-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 Source Connector for Confluent Platform](config.md#rabbit-m-q-source-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).

## Usage notes

As a user of the RabbitMQ Source connector, you should be aware of the following:

- The queue or topic that the connector will read from must exist prior to
  starting the connector.
- All the headers associated with each message from RabbitMQ are prefixed with
  `rabbitmq.` and are copied over to each of the records produced to Kafka;
  therefore, you must configure the message format to support headers.
- The RabbitMQ Source connector is based on the AMQP protocol, so it may work
  with other servers that implement the protocol.

## Examples

### Property-based example

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

```bash
 name=RabbitMQSourceConnector1
 connector.class=io.confluent.connect.rabbitmq.RabbitMQSourceConnector
 tasks.max=1
 kafka.topic=< Required Configuration >
 rabbitmq.queue=< Required Configuration >
```

### REST-based example

This configuration is used typically along 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 the distributed connect worker(s). 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" : "RabbitMQSourceConnector1",
   "config" : {
     "connector.class" : "io.confluent.connect.rabbitmq.RabbitMQSourceConnector",
     "tasks.max" : "1",
     "kafka.topic" : "< Required Configuration >",
     "rabbitmq.queue" : "< Required Configuration >"
   }
 }
```

Use curl to post the configuration to one of the Kafka Connect Workers. 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/RabbitMQSourceConnector1/config
```

<a id="rabbitmq-quickstart"></a>

## Quick start

The RabbitMQ Source connector streams records from RabbitMQ queues into Kafka
topics with high throughput. This quick start shows example data production and
consumption setups in detail. The steps are tailored to a macOS terminal
environment.

### Install the connector

For the following tutorial, you must have Confluent Platform running locally. To install the
connector complete the following steps:

1. Navigate to your Confluent Platform installation directory and enter the following command:
   ```bash
   confluent connect plugin install confluentinc/kafka-connect-rabbitmq:latest
   ```

   Note that adding a new connector plugin requires restarting Kafka Connect.
2. Use the Confluent CLI to restart Connect.
   ```bash
   ./confluent local services connect stop && ./confluent local services connect start
   ```

   Wait for Connect to come up before continuing:
   ```bash
   This CLI is intended for development only, not for production
   https://docs.confluent.io/current/cli/index.html

   Using CONFLUENT_CURRENT: /var/folders/q_/33p31psj3f90ntlbpd82qr9w0000gn/T/confluent.6S9QWg14
   Stopping Connect
   Connect is [DOWN]
   This CLI is intended for development only, not for production
   https://docs.confluent.io/current/cli/index.html

   Using CONFLUENT_CURRENT: /var/folders/q_/33p31psj3f90ntlbpd82qr9w0000gn/T/confluent.6S9QWg14
   Starting Zookeeper
   Zookeeper is [UP]
   Starting Kafka
   Kafka is [UP]
   Starting Schema Registry
   Schema Registry is [UP]
   Starting Connect
   Connect is [UP]
   ```
3. Verify the RabbitMQ Source connector plugin has been installed correctly
   and recognized by the plugin loader:
   ```bash
   curl -sS localhost:8083/connector-plugins | jq .[].class | grep RabbitMQSourceConnector
   ```

   This should print the name of at least one class that has been found by the
   classloader.
   ```bash
   "io.confluent.connect.rabbitmq.RabbitMQSourceConnector"
   ```

### Install and test RabbitMQ broker locally

Use the following steps to install and test a RabbitMQ broker locally.

#### Install and deploy the broker

1. From a terminal session, [install the RabbitMQ broker](https://www.rabbitmq.com/download.html). For a macOS environment, you can
   use [homebrew](https://brew.sh/) :
   ```bash
   brew update
   brew install rabbitmq
   ```
2. Configure the terminal environment to include the RabbitMQ `sbin` folder in
   the PATH:
   ```bash
   export PATH=$PATH:/usr/local/opt/rabbitmq/sbin
   ```
3. In a new terminal session, start the RabbitMQ broker:
   ```bash
   rabbitmq-server
   ```

   The server should print the following startup message:
   ```bash
   ##  ##
   ##  ##      RabbitMQ 3.7.16. Copyright (C) 2007-2019 Pivotal Software, Inc.
   ##########  Licensed under the MPL.  See https://www.rabbitmq.com/
   ######  ##
   ##########  Logs: /usr/local/var/log/rabbitmq/rabbit@localhost.log
                  /usr/local/var/log/rabbitmq/rabbit@localhost_upgrade.log

               Starting broker...

   completed with 6 plugins.
   ```
4. Perform a health check using the `rabbitmqctl status` command:
   ```bash
   rabbitmqctl status
   ```

   A healthy server should have a number of `running_applications` as shown in
   the following example output:
   ```bash
   Status of node rabbit@localhost ...
   [{pid,82126},
   {running_applications,
   [{rabbitmq_management,"RabbitMQ Management Console","3.7.15"},
      {rabbitmq_web_dispatch,"RabbitMQ Web Dispatcher","3.7.15"},
      {cowboy,"Small, fast, modern HTTP server.","2.6.1"},
      {cowlib,"Support library for manipulating Web protocols.","2.7.0"},
      {rabbitmq_management_agent,"RabbitMQ Management Agent","3.7.15"},
      {rabbitmq_stomp,"RabbitMQ STOMP plugin","3.7.15"},
      {rabbitmq_amqp1_0,"AMQP 1.0 support for RabbitMQ","3.7.15"},
      {rabbitmq_mqtt,"RabbitMQ MQTT Adapter","3.7.15"},
      {rabbit,"RabbitMQ","3.7.15"},
      {amqp_client,"RabbitMQ AMQP Client","3.7.15"},
      {rabbit_common,
         "Modules shared by rabbitmq-server and rabbitmq-erlang-client",
         "3.7.15"},
      ...
      remainder clipped for brevity
   ```

#### Install Python for data generation

```bash
brew install python3
pip3 install pika --upgrade
```

#### RabbitMQ data generator in Python

Save this program into a file called `producer.py`:

```python
#!/usr/bin/env python3
import pika
import sys
import json

if len(sys.argv) != 3:
   print("Usage: " + sys.argv[0] + " <queueName> <count>")
   sys.exit(1)

queue  = sys.argv[1]
count = int(sys.argv[2])

print("count:\t%d\nqueue:\t%s" % (count, queue) )

msgBody = {
        "id" : 0 ,
        "body" :  "010101010101010101010101010101010101010101010101010101010101010101010"
        }

connection = pika.BlockingConnection(pika.ConnectionParameters('localhost'))
channel = connection.channel()
channel.queue_declare(queue = queue)

properties = pika.BasicProperties(content_type='application/json', delivery_mode=1, priority=1, content_encoding='utf-8')
for i in range(count):
    msgBody["id"] = i
    jsonStr = json.dumps(msgBody)
    properties.message_id = str(i)
    channel.basic_publish(exchange = '', routing_key = queue, body = jsonStr, properties = properties)
    print("Send\t%r" % msgBody)

connection.close()
print('Exiting')
```

#### IMPORTANT
Messages produced to the RabbitMQ queue must have basic properties defined.
In order to read messages lacking basic properties, use the
`ByteArrayConverter` [value
converter](/platform/current/connect/concepts.html#converters) in the connector properties
file. All other converters require basic properties.

#### RabbitMQ consumer program in Python

Save this program into a file called `consumer.py`:

```python
#!/usr/bin/env python3
import pika
import sys
import json

if len(sys.argv) != 2:
   print("Usage: " + sys.argv[0] + " <queueName>")
   sys.exit(1)

queue = sys.argv[1]

print("queue:\t%s" % (queue) )

connection = pika.BlockingConnection(pika.ConnectionParameters('localhost'))
channel = connection.channel()
channel.queue_declare(queue = queue)

def callback(ch, method, properties, body):
    msgBody = json.loads(body)
    print("Receive\t%r" % msgBody)

channel.basic_consume(queue = queue,
                      auto_ack=True,
                      on_message_callback=callback)

print('Waiting for messages. To exit press CTRL+C')
try:
    channel.start_consuming()
except KeyboardInterrupt:
    print('Exiting')
```

#### Generate and consume data in RabbitMQ

Run the following command to produce five records to a queue called `myqueue`:

```bash
./producer.py myqueue 5
```

The script should show five generated records in the console output.

```bash
count:      5
queue:      myqueue
Send        {'id': 0, 'body': '010101010101010101010101010101010101010101010101010101010101010101010'}
Send        {'id': 1, 'body': '010101010101010101010101010101010101010101010101010101010101010101010'}
Send        {'id': 2, 'body': '010101010101010101010101010101010101010101010101010101010101010101010'}
Send        {'id': 3, 'body': '010101010101010101010101010101010101010101010101010101010101010101010'}
Send        {'id': 4, 'body': '010101010101010101010101010101010101010101010101010101010101010101010'}
Exiting
```

Run the following command to consume all records in RabbitMQ queue `myqueue`:

```bash
./consumer.py myqueue
```

The script should show five consumed records in the console output.

```bash
queue:      myqueue
Waiting for messages. To exit press CTRL+C
Receive     {'id': 0, 'body': '010101010101010101010101010101010101010101010101010101010101010101010'}
Receive     {'id': 1, 'body': '010101010101010101010101010101010101010101010101010101010101010101010'}
Receive     {'id': 2, 'body': '010101010101010101010101010101010101010101010101010101010101010101010'}
Receive     {'id': 3, 'body': '010101010101010101010101010101010101010101010101010101010101010101010'}
Receive     {'id': 4, 'body': '010101010101010101010101010101010101010101010101010101010101010101010'}
```

### Start the Kafka RabbitMQ Source connector

To start the Kafka Connect RabbitMQ Source connector, complete the following
steps:

1. Create the  `register-rabbitmq-connect.json` file containing the following
   connector configuration:
   ```json
   {
         "name" : "RabbitMQSourceConnector1",
         "config" : {
         "connector.class" : "io.confluent.connect.rabbitmq.RabbitMQSourceConnector",
         "tasks.max" : "1",
         "key.converter":"org.apache.kafka.connect.storage.StringConverter",
         "value.converter": "org.apache.kafka.connect.converters.ByteArrayConverter",
         "kafka.topic" : "rabbitmq",
         "rabbitmq.queue" : "myqueue",
         "rabbitmq.host" : "localhost",
         "rabbitmq.username" : "guest",
         "rabbitmq.password" : "guest"
         }
      }
   ```

   #### NOTE
   The `username` and `password` values are defaults in RabbitMQ. These
   default values are only permitted for authentication when attempted on the
   localhost where RabbitMQ is installed and running. See this [article](https://www.rabbitmq.com/access-control.html) for instructions on
   allowing remote access to RabbitMQ’s default account.
2. Start the connector:
   ```bash
   curl -i -X POST -H "Accept:application/json" -H  "Content-Type:application/json" http://localhost:8083/connectors/ -d @register-rabbitmq-connect.json
   ```
3. Verify the result includes `201 Created` which indicates that the RabbitMQ
   connector was loaded.

### Perform the end-to-end test

To perform an end-to-end test, complete the following steps:

1. In a new terminal session, navigate to the Kafka bin folder and start a consumer:
   ```bash
   ./kafka-avro-console-consumer --bootstrap-server localhost:9092 --topic rabbitmq
   ```
2. Produce some records to the RabbitMQ queue:
   ```bash
   ./producer.py myqueue 5
   ```

   You should expect to see the following from the `kafka-avro-console-consumer`
   command, indicating that the records were successfully produced to Kafka:
   ```bash
   "´\u0001{\"id\": 0, \"body\": \"010101010101010101010101010101010101010101010101010101010101010101010\"}"
   "´\u0001{\"id\": 1, \"body\": \"010101010101010101010101010101010101010101010101010101010101010101010\"}"
   "´\u0001{\"id\": 2, \"body\": \"010101010101010101010101010101010101010101010101010101010101010101010\"}"
   "´\u0001{\"id\": 3, \"body\": \"010101010101010101010101010101010101010101010101010101010101010101010\"}"
   "´\u0001{\"id\": 4, \"body\": \"010101010101010101010101010101010101010101010101010101010101010101010\"}"
   ```

   Note that if RabbitMQ is restarted, data in memory disappears because the
   message queue used in this test is not durable.  Use the command
   `rabbitmqctl list_queues` to view active queues and message counts.

### Clean up resources

1. Delete the connector and stop Confluent services.
   > ```bash
   > curl -X DELETE localhost:8083/connectors/RabbitMQSourceConnector1
   > confluent local stop
   > ```
2. Stop your RabbitMQ server using `CTRL+C`.