Skip to main content
Version: Next

Rabbitmq

Rabbitmq source connector

Description​

Used to read data from Rabbitmq.

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_hoststringyes-
usernamestringyes-
passwordstringyes-
queue_namestringyes-
schemaconfigyes-
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_timeoutlongno-
common-optionsno-

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 for details

Example​

simple:

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​

Change Log
ChangeCommitVersion
[Improve] rabbit mq options (#8740)https://github.com/apache/seatunnel/commit/4eec9be012dev
[Improve] restruct connector common options (#8634)https://github.com/apache/seatunnel/commit/f3499a6eebdev
[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