AMQP 1.0 Source Connector
support sink/source for AMQP version 1.0.0

Available on
StreamNative Cloud console

Authored by
Support type
Apache License 2.0

AMQP 1.0 source connector

The AMQP 1.0 source connector receives messages from AMQP 1.0 and writes messages to Pulsar topics.

Quick start

1. Start AMQP 1.0 service

Start a service that supports the AMQP 1.0 protocol, such as Solace.

docker run -d -p 8080:8080 -p:8008:8008 -p:1883:1883 -p:8000:8000 -p:5672:5672 -p:9000:9000 -p:2222:2222 --shm-size=2g --env username_admin_globalaccesslevel=admin --env username_admin_password=admin --name=solace solace/solace-pubsub-standard

2. Create a connector

The following command shows how to use pulsarctl to create a builtin connector. If you want to create a non-builtin connector, you need to replace --source-type amqp1_0 with --archive /path/to/pulsar-io-amqp1_0.nar. You can find the button to download the nar package at the beginning of the document.

For StreamNative Cloud User

If you are a StreamNative Cloud user, you need set up your environment first.

pulsarctl sources create \
  --source-type amqp1_0 \
  --name amqp1_0-source \
  --tenant public \
  --namespace default \
  --destination-topic-name "Your topic name" \
  --parallelism 1 \
  --source-config \
    "connection": {
      "failover": {
        "useFailover": true
      "uris": [
          "protocol": "amqp",
          "host": "localhost",
          "port": 5672,
          "urlOptions": [
    "username": "guest",
    "password": "guest",
    "queue": "user-op-queue-pulsar"

The --source-config is the minimum necessary configuration for starting this connector, and it is a JSON string. You need to substitute the relevant parameters with your own. If you want to configure more parameters, see Configuration Properties for reference.


You can also choose to use a variety of other tools to create a connector:

2. Send messages to the AMQP 1.0 service


  • The following sample code uses the Apache qpid library.
    public static void main(String[] args) {
        ConnectionFactory connectionFactory = new JmsConnectionFactory("amqp://localhost:5672");
        Connection connection = connectionFactory.createConnection();
        JMSProducer producer = connectionFactory.createContext().createProducer();
        Destination destination = new JmsQueue("user-op-queue");
        for (int i = 0; i < 10; i++) {
            producer.send(destination, "Hello AMQP 1.0 - " + i);

3. Consume data from Pulsar


bin/pulsar-client \
--url "Your Pulsar serviceUrl" \
consume "The topic that you specified when you created the connector" -s "test-sub" -n 10 -p Earliest

Configuration Properties

Before using the AMQP 1.0 sink connector, you need to configure it.

You can create a configuration file (JSON or YAML) to set the following properties.

protocolStringrequired if connection is not used"amqp"[deprecated: use connection instead] The AMQP protocol.
hostStringrequired if connection is not used" " (empty string)[deprecated: use connection instead] The AMQP service host.
portintrequired if connection is not used5672[deprecated: use connection instead] The AMQP service port.
connectionConnectionrequired if protocol, host, port is not used" " (empty string)The connection details.
usernameStringfalse" " (empty string)The username used to authenticate to ActiveMQ.
passwordStringfalse" " (empty string)The password used to authenticate to ActiveMQ.
queueStringfalse" " (empty string)The queue name that messages should be read from or written to.
topicStringfalse" " (empty string)The topic name that messages should be read from or written to.
activeMessageTypeStringfalse0The ActiveMQ message simple class name.
onlyTextMessagebooleanfalsefalseIf it is set to true, the AMQP message type must be set to TextMessage. Pulsar consumers can consume the messages with schema ByteBuffer.

A Connection object can be specified as follows:

failoverFailoverfalse" " (empty string)The configuration for a failover connection.
urislist of ConnectionUritrue" " (empty string)A list of ConnectionUri objects. When useFailover is set to true 1 or more should be provided. Currently only 1 uri is supported when useFailover is set to false

A Failover object can be specified as follows:

useFailoverbooleantruefalseIf it is set to true, the connection will be created from the uris provided under uris, using qpid's failover connection factory.
jmsClientIdStringrequired if failoverConfigurationOptions is used" " (empty string)Identifying name for the jms Client
failoverConfigurationOptionsList of Stringrequired if jmsClientId is used" " (empty string)A list of options (e.g. <key=value>). The options wil be joined using an '&', prefixed with a the jmsClientId and added to the end of the failoverUri. see also:

A ConnectionUri object can be specified as follows:

protocolStringtrue" " (empty string)The AMQP protocol.
hostStringtrue" " (empty string)The AMQP service host.
portinttrue0The AMQP service port.
urlOptionsList of Stringfalse" " (empty string)A list of url-options (e.g. <key=value>). The url options wil be joined using an '&', prefixed with a '?' and added to the end of the uri