Skip to main content
Version: Next

RabbitMQ

RabbitMQ source connector

Support Those Engines​

Spark
Flink
SeaTunnel Zeta

Description​

Used to read data from RabbitMQ queues.

Key features​

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​

nametyperequireddefault value
hoststringyes-
portintyes-
virtual_hoststringno-
usernamestringno-
passwordstringno-
queue_namestringno-
schemaconfigno-
tables_configsarrayno-
formatstringnojson
protobuf_schemastringno-
protobuf_message_namestringno-
urlstringno-
uristringno-
sslbooleannofalse
routing_keystringno-
exchangestringno-
network_recovery_intervalintno-
topology_recovery_enabledbooleanno-
AUTOMATIC_RECOVERY_ENABLEDbooleanno-
connection_timeoutintno-
requested_channel_maxintno-
requested_frame_maxintno-
requested_heartbeatintno-
prefetch_countintno-
delivery_timeoutintno-
use_correlation_idbooleanno-
common-optionsno-
durablebooleannotrue
exclusivebooleannofalse
auto_deletebooleannofalse
passivebooleannofalse

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

username and password should be configured together.

url [string]​

convenience method for setting the fields in an AMQP URI: host, port, username, password and virtual host

uri [string]​

Legacy alias for url. Configure only one of url and uri.

ssl [boolean]​

Enables SSL/TLS for host-and-port configuration. Use url with an amqps:// URI when the URI itself supplies the connection settings.

When url uses an amqps:// URI, the broker certificate is verified against the JVM trust store with hostname verification enabled. Connections that previously relied on the implicit trust-all behavior with self-signed or private-CA certificates must import the broker certificate into the trust store, or they will fail to connect.

queue_name [string]​

the queue to consume messages from. Note: Required if tables_configs is not configured.

routing_key [string]​

Optional RabbitMQ routing key inherited from the shared RabbitMQ configuration. It is not required for normal queue consumption.

exchange [string]​

Optional RabbitMQ exchange inherited from the shared RabbitMQ configuration. It is not required for normal queue consumption.

schema [Config]​

fields [Config]​

the schema fields of upstream data. For more details, please refer to Schema Feature. Note: Required if tables_configs is not configured.

tables_configs [array]​

Used to read from multiple queues simultaneously. Each object in the array must contain queue_name and schema.

format [string]​

The message payload format. Supported values are json and protobuf. The default value is json.

protobuf_schema [string]​

Effective when format is protobuf. Defines the Protobuf schema used to deserialize the RabbitMQ message payload.

protobuf_message_name [string]​

Effective when format is protobuf. Specifies the Protobuf message name to deserialize.

network_recovery_interval [int]​

how long will automatic recovery wait before attempting to reconnect, in ms

topology_recovery_enabled [boolean]​

if true, enables topology recovery

AUTOMATIC_RECOVERY_ENABLED [boolean]​

If true, enables connection recovery.

The option key is currently uppercase in the connector configuration. Use AUTOMATIC_RECOVERY_ENABLED, not automatic_recovery_enabled.

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: 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: 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 [int]​

deliveryTimeout maximum wait time, in milliseconds, for the next message delivery

use_correlation_id [boolean]​

Whether the consumed messages provide a unique correlation id that can be used to deduplicate messages when acknowledgments fail.

common options​

Source plugin common parameters, please refer to Source Common Options for details

durable​

  • true: The queue will survive on server restart.
  • false: The queue will be deleted on server restart.

exclusive​

  • true: The queue is used only by the current connection and will be deleted when the connection closes.
  • false: The queue can be used by multiple connections.

auto_delete​

  • true: The queue will be deleted automatically when the last consumer unsubscribes.
  • false: The queue will not be automatically deleted.

passive​

  • false: Declare the queue with the configured durable, exclusive, and auto-delete settings.
  • true: Verify that the queue already exists without creating or modifying it. Use this for consumer accounts without queue-declaration permission.

Migration Guide & Configuration Rules​

If you are upgrading from a previous version that only supported single-table reads, your existing configuration will work without any changes.

Configuration Priority:

  • You cannot configure both tables_configs and the root-level queue_name at the same time. They are mutually exclusive. Doing so will result in a configuration validation error.
  • Use tables_configs for multi-table mode.
  • Use root-level queue_name and schema for single-queue mode.
  • In multi-table mode, put each queue's schema inside its own tables_configs item.
  • When format is protobuf, configure both protobuf_schema and protobuf_message_name at the same level as the queue configuration.
  • If you configure username, you must also configure password, and vice versa.
  • Configure only one of url and uri. uri is retained for existing configurations; use url in new configurations.
  • Set ssl = true when connecting to an AMQPS endpoint with host and port settings.
  • host and port are always required. virtual_host is optional unless your RabbitMQ deployment requires a non-default virtual host.

Example​

Single-table Read Example​

env {
    parallelism = 1
    job.mode = "STREAMING"
}

source {
    RabbitMQ {
        host = "rabbitmq-e2e"
        port = 5672
        virtual_host = "/"
        username = "guest"
        password = "guest"
        queue_name = "test"
        durable = true
        exclusive = false
        auto_delete = false
        schema = {
            fields {
                id = bigint
                c_map = "map<string, smallint>"
                c_array = "array<tinyint>"
                c_string = string
                c_boolean = boolean
            }
        }
    }
}

sink {
    Console {}
}

Multi-table Read Example​

You can use the tables_configs option to consume messages from multiple RabbitMQ queues simultaneously within a single job. The connector will automatically assign the correct table identifier to each row based on the queue it originated from, allowing you to route them to different sinks using plugin_input.

env {
    parallelism = 1
    job.mode = "STREAMING"
}

source {
  RabbitMQ {
    host = "rabbitmq-e2e"
    port = 5672
    virtual_host = "/"
    username = "guest"
    password = "guest"

    # Use tables_configs to read from multiple queues
    tables_configs = [
      {
        queue_name = "users_queue"
        schema = {
          table = "users_table" # Defines the table name for routing
          fields {
            user_id = bigint
            name = string
          }
        }
      },
      {
        queue_name = "orders_queue"
        schema = {
          table = "orders_table" # Defines the table name for routing
          fields {
            order_id = bigint
            amount = double
          }
        }
      }
    ]
  }
}

sink {
  # The first sink will only receive data from users_table
  Console {
    plugin_input = "users_table"
  }

  # The second sink will only receive data from orders_table
  Console {
    plugin_input = "orders_table"
  }
}

Protobuf Read Example​

source {
    RabbitMQ {
        host = "rabbitmq-e2e"
        port = 5672
        queue_name = "protobuf_queue"
        format = protobuf
        protobuf_message_name = Person
        protobuf_schema = """
            syntax = "proto3";
            message Person {
              int64 id = 1;
              string name = 2;
            }
        """
        schema = {
            fields {
                id = bigint
                name = string
            }
        }
    }
}

FAQ​

Why must parallelism be set to 1 to achieve exactly-once?​

RabbitMQ dispatches messages among multiple active consumers on the same queue in a round-robin manner. When multiple parallel readers consume from the same queue, message ordering and deterministic offset/acknowledgement coordination across distributed workers cannot be guaranteed. Therefore, setting parallelism to 1 is required for deterministic exactly-once delivery.

What message formats are supported by RabbitMQ source?​

RabbitMQ source supports JSON by default and Protobuf when format is set to protobuf. The connector deserializes each RabbitMQ message payload into one SeaTunnel row according to the configured schema.

How does the source handle unacknowledged messages when a failure occurs?​

When a SeaTunnel task fails or crashes, the RabbitMQ connection drops, and RabbitMQ automatically requeues any unacknowledged messages. Upon job restoration from a checkpoint, the reader resumes processing without message loss.

Changelog​

Change Log
ChangeCommitVersion
[Fix][connector-rabbitmq] Set default value for durable, exclusive and auto-delete (#9631)https://github.com/apache/seatunnel/commit/5f9492e62a2.3.12
[Feature][Checkpoint] Add check script for source/sink state class serialVersionUID missing (#9118)https://github.com/apache/seatunnel/commit/4f5adeb1c72.3.11
[Improve] rabbit mq options (#8740)https://github.com/apache/seatunnel/commit/4eec9be0122.3.10
[Improve] restruct connector common options (#8634)https://github.com/apache/seatunnel/commit/f3499a6eeb2.3.10
[Improve][dist]add shade check rule (#8136)https://github.com/apache/seatunnel/commit/51ef8000162.3.9
[Feature][Restapi] Allow metrics information to be associated to logical plan nodes (#7786)https://github.com/apache/seatunnel/commit/6b7c53d03c2.3.9
[Feature][Rabbitmq] Allow configuration of queue durability and deletion policy (#7365)https://github.com/apache/seatunnel/commit/aabfc8eb782.3.8
[Hotfix][connector-v2-rabbit] fix rabbit checkpoint exception in Flink mode (#7108)https://github.com/apache/seatunnel/commit/423a7b142b2.3.6
[Feature][Kafka] Support multi-table source read (#5992)https://github.com/apache/seatunnel/commit/60104602d12.3.6
[Improve] Remove use SeaTunnelSink::getConsumedType method and mark it as deprecated (#5755)https://github.com/apache/seatunnel/commit/8de74081002.3.4
Support config column/primaryKey/constraintKey in schema (#5564)https://github.com/apache/seatunnel/commit/eac76b4e502.3.4
[Bugfix][connector-v2][rabbitmq] Fix reduplicate ack msg bug and code style (#4842)https://github.com/apache/seatunnel/commit/985fb6642a2.3.2
[Hotfix][E2E] Fix RabbitmqIT (#4593)https://github.com/apache/seatunnel/commit/9bd5403d712.3.2
Merge branch 'dev' into merge/cdchttps://github.com/apache/seatunnel/commit/4324ee19122.3.1
[Improve][Project] Code format with spotless plugin.https://github.com/apache/seatunnel/commit/423b5830382.3.1
[improve][api] Refactoring schema parse (#4157)https://github.com/apache/seatunnel/commit/b2f573a13e2.3.1
[Improve][build] Give the maven module a human readable name (#4114)https://github.com/apache/seatunnel/commit/d7cd6010512.3.1
[Improve][Project] Code format with spotless plugin. (#4101)https://github.com/apache/seatunnel/commit/a2ab1665612.3.1
[Feature][Connector] add get source method to all source connector (#3846)https://github.com/apache/seatunnel/commit/417178fb842.3.1
[Improve][Connector-V2] Change Connector Custom Config Prefix To Map (#3719)https://github.com/apache/seatunnel/commit/ef1b8b1bb52.3.1
[Feature][API &amp; Connector &amp; Doc] add parallelism and column projection interface (#3829)https://github.com/apache/seatunnel/commit/b9164b8ba12.3.1
[Hotfix][OptionRule] Fix option rule about all connectors (#3592)https://github.com/apache/seatunnel/commit/226dc6a1192.3.0
[Feature][Connector-V2][RabbitMQ] Add RabbitMQ source & sink connector (#3312)https://github.com/apache/seatunnel/commit/4b12691a8d2.3.0