Skip to main content
Version: 3.2.x

RabbitMQ source connector


You can download all the Pulsar connectors on download page.

The RabbitMQ source connector receives messages from RabbitMQ clusters and writes messages to Pulsar topics.


The configuration of the RabbitMQ source connector has the following properties.


connectionNameStringtrue(empty string)The connection name.
hostStringtrue(empty string)The RabbitMQ host.
portinttrue5672The RabbitMQ port.
virtualHostStringtrue/The virtual host used to connect to RabbitMQ.
usernameStringfalseguestThe username used to authenticate to RabbitMQ.
passwordStringfalseguestThe password used to authenticate to RabbitMQ.
queueNameStringtrue(empty string)The RabbitMQ queue name that messages should be read from or written to.
requestedChannelMaxintfalse0The initially requested maximum channel number.
0 means unlimited.
requestedFrameMaxintfalse0The initially requested maximum frame size in octets.
0 means unlimited.
connectionTimeoutintfalse60000The timeout of TCP connection establishment in milliseconds.
0 means infinite.
handshakeTimeoutintfalse10000The timeout of AMQP0-9-1 protocol handshake in milliseconds.
requestedHeartbeatintfalse60The requested heartbeat timeout in seconds.
prefetchCountintfalse0The maximum number of messages that the server delivers.
0 means unlimited.
prefetchGlobalbooleanfalsefalseWhether the setting should be applied to the entire channel rather than each consumer.
passivebooleanfalsefalseWhether the rabbitmq consumer should create its own queue or bind to an existing one.


Before using the RabbitMQ source connector, you need to create a configuration file through one of the following methods.

  • JSON

    "configs": {
    "host": "localhost",
    "port": "5672",
    "virtualHost": "/",
    "username": "guest",
    "password": "guest",
    "queueName": "test-queue",
    "connectionName": "test-connection",
    "requestedChannelMax": "0",
    "requestedFrameMax": "0",
    "connectionTimeout": "60000",
    "handshakeTimeout": "10000",
    "requestedHeartbeat": "60",
    "prefetchCount": "0",
    "prefetchGlobal": "false",
    "passive": "false"
  • YAML

    host: "localhost"
    port: 5672
    virtualHost: "/"
    username: "guest"
    password: "guest"
    queueName: "test-queue"
    connectionName: "test-connection"
    requestedChannelMax: 0
    requestedFrameMax: 0
    connectionTimeout: 60000
    handshakeTimeout: 10000
    requestedHeartbeat: 60
    prefetchCount: 0
    prefetchGlobal: "false"
    passive: "false"