Rabbitmq

Rabbitmq source connector

Description​

Used to read data from Rabbitmq.

Key features​

• batch

• stream

• exactly-once

• column projection

• parallelism

• support user-defined split

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​

name	type	required	default value
host	string	yes	-
port	int	yes	-
virtual_host	string	yes	-
username	string	yes	-
password	string	yes	-
queue_name	string	yes	-
schema	config	yes	-
url	string	no	-
routing_key	string	no	-
exchange	string	no	-
network_recovery_interval	int	no	-
topology_recovery_enabled	boolean	no	-
automatic_recovery_enabled	boolean	no	-
connection_timeout	int	no	-
requested_channel_max	int	no	-
requested_frame_max	int	no	-
requested_heartbeat	int	no	-
prefetch_count	int	no	-
delivery_timeout	long	no	-
common-options		no	-

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>"
            }
        }
    }
}