sink
RabbitMQ Sink
The RabbitMQ sink connector pulls messages from Pulsar topics and persist the messages to RabbitMQ queues.
Authored by
ASF
Support type
StreamNative
License
Apache License 2.0

The RabbitMQ sink connector pulls messages from Pulsar topics and persist the messages to RabbitMQ queues.

Configuration

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

Property

NameTypeRequiredDefaultDescription
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. <br><br>0 means unlimited.
requestedFrameMaxintfalse0The initially requested maximum frame size in octets. <br><br>0 means unlimited.
connectionTimeoutintfalse60000The timeout of TCP connection establishment in milliseconds. <br><br>0 means infinite.
handshakeTimeoutintfalse10000The timeout of AMQP0-9-1 protocol handshake in milliseconds.
requestedHeartbeatintfalse60The exchange to publish messages.
exchangeNameStringtrue" " (empty string)The maximum number of messages that the server delivers.<br><br> 0 means unlimited.
prefetchGlobalStringtrue" " (empty string)The routing key used to publish messages.

Example

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

  • JSON

    {
        "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",
        "exchangeName": "test-exchange",
        "routingKey": "test-key"
    }
    
  • YAML

    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
        exchangeName: "test-exchange"
        routingKey: "test-key"