Skip to main content
Version: Next

RabbitMQ

RabbitMQ source connector

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-
urlstringno-
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

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

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.

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.

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.
  • If you configure username, you must also configure password, and vice versa.
  • 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"
  }
}

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