Starlight for RabbitMQ acts as a proxy between your RabbitMQ application and Apache Pulsar. It implements the AMQP 0.9.1 protocol used by RabbitMQ clients and translates AMQP frames and concepts to Pulsar ones. The proxy can be run as a standalone jar, a Pulsar Pluggable Protocol Handler or a Pulsar Proxy extension.

This is currently not implemented but on the roadmap:

• Headers exchanges
• Exclusive consumers
• Non durable exchanges and queues
• Transient messages (all messages are persisted)

RabbitMQ and Pulsar work in a pretty different way. Starlight for RabbitMQ was designed to make the most benefit from Pulsar's scalability. This results in some differences of behavior:

• Canceling an AMQP consumer will requeue the messages that were received through it since it also closes the associated Pulsar consumers.

To build from code, complete the following steps:

1. Clone the project from GitHub.

git clone https://github.com/datastax/starlight-for-rabbitmq.git
cd starlight-for-rabbitmq

2. Build the project.

mvn clean install -DskipTests

You can find the executable jar file in the following directory.

./starlight-rabbitmq/target/starlight-rabbitmq-${version}-jar-with-dependencies.jar

You can find the nar file in the following directory.

./starlight-rabbitmq/target/starlight-rabbitmq-${version}.nar

1. Set the URLs of the Pulsar brokers and the ZooKeeper configuration store in a configuration file. Eg:

   brokerServiceURL=pulsar://localhost:6650
   brokerWebServiceURL=http://localhost:8080
   configurationStoreServers=localhost:2181

2. Run as a Java application and provide the configuration file path in the -c/--config option:

   java -jar ./starlight-rabbitmq/target/starlight-rabbitmq-${version}-jar-with-dependencies.jar -c conf/starlight-rabbitmq.conf

Starlight for RabbitMQ can be embedded directly into the Pulsar brokers by loading it as a protocol handler.

1. Set the configuration of the Starlight for RabbitMQ protocol handler in the broker configuration file (generally broker.conf or standalone.conf). Example where the NAR file was copied into the ./protocols directory:

   messagingProtocols=rabbitmq
   protocolHandlerDirectory=./protocols

2. Set the AMQP service listeners. Note that the hostname value in listeners is the same as Pulsar broker's advertisedAddress. The following is an example.

   amqpListeners=amqp://127.0.0.1:5672
   advertisedAddress=127.0.0.1

3. Start the Pulsar broker

Starlight for RabbitMQ can be embedded into the Pulsar Proxy by loading it as a proxy extension.

1. Set the configuration of the Starlight for RabbitMQ proxy extension in the proxy configuration file (generally proxy.conf or standalone.conf). Example where the NAR file was copied into the ./proxyextensions directory:

   proxyExtensions=rabbitmq
   proxyExtensionsDirectory=./proxyextensions

2. Set the AMQP service listeners. Note that the hostname value in listeners is the same as Pulsar proxy's advertisedAddress. The following is an example.

   amqpListeners=amqp://127.0.0.1:5672
   advertisedAddress=127.0.0.1

3. Start the Pulsar Proxy

You can use a RabbitMQ/AMQP-0.9.1 client or a tool such as RabbitMQ PerfTest to check that everything works correctly.

For instance the following Python script creates a queue, publishes a message that will be routed to this queue, reads the message from the queue and deletes the queue

#!/usr/bin/env python
import pika

connection = pika.BlockingConnection(pika.ConnectionParameters(port=5672))
channel = connection.channel()

try:
    channel.queue_declare("test-queue")
    print("created test-queue queue")
    channel.basic_publish(exchange="", routing_key="test-queue", body="test".encode('utf-8'))
    print("published message test")
    _, _, res = channel.basic_get(queue="test-queue", auto_ack=True)
    assert res is not None, "should have received a message"
    print("received message: " + res.decode())
    channel.queue_delete("test-queue")
    print("deleted test-queue queue")
finally:
    connection.close()

AMQP 0.9.1 (the protocol used by RabbitMQ) employs the concepts of Exchanges, Queues and Bindings to provide basic routing capabilities inside the message broker. These concepts are mapped to Pulsar topics and features. One important architectural decision is that Starlight for RabbitMQ doesn’t interact directly with the managed ledger as in other RabbitMQ integrations for Pulsar. Interacting with the ledger has the advantage of being performant, but the disadvantage is that the broker which interacts with the ledger must have ownership of the topic. Since in AMQP 0.9.1 there is a many-to-many relationship between Exchanges and Queues for a given Virtual host, all Exchanges and Queues and related topics would have to be owned by the same broker. There are techniques to do this using Topic bundles, but the result is that a full AMQP Virtual host `can be handled by only one broker at a time. This is an issue for scalability. So instead, Starlight for RabbitMQ acts as a proxy and uses the Pulsar binary protocol to communicate with the brokers. This way it can leverage Pulsar features such as load balancing of the topics on the brokers, batching of messages, partitioning of topics, and load balancing of the data on the consumers.

On the publishing side, an AMQP exchange is mapped to a topic. Depending on the type of exchange, the publishing routing key may also be included in the topic name.

On the consumer side, Pulsar shared subscriptions are used to represent the AMQP Bindings from an Exchange to a Queue. When creating an AMQP Queue consumer, the proxy creates Pulsar consumers for all the Bindings of the Queue.

When you unbind the Queue, the Pulsar subscription isn’t deleted right away since the consumer may be lagging. Messages are still received from the subscription and filtered if their position is past the end of the binding. When all messages from the binding have been acknowledged, then the corresponding subscription can finally be removed by the SubscriptionCleaner task.

Starlight for RabbitMQ uses Apache Zookeeper to store the AMQP entities metadata consistently. The existing ZooKeeper configuration store can be reused for this, and Starlight for RabbitMQ will employ the /starlight-rabbitmq prefix to write its entries into ZooKeeper.

Starlight for RabbitMQ supports connections using TLS/mTLS to ensure privacy and security of the communication. It also supports the PLAIN and EXTERNAL mechanisms used by RabbitMQ. Internally, it uses the same AuthenticationService as Apache Pulsar and maps the AMQP mechanisms to existing Pulsar authentication modes. At the moment there is no support for authorization so an authenticated user has full access to all Virtual hosts. Starlight for RabbitMQ can connect to brokers that have TLS and/or authentication, and/or authorization enabled. To perform its operations, Starlight for RabbitMQ currently needs to use an “admin role”. Future versions will relay the principal authenticated to the proxy and use a “proxy role” so operations on the broker will have permissions from the originating application.

The PLAIN mechanism is mapped to the AuthenticationProviderToken mode of authentication. The username is ignored and the password is used as the JSON Web Token (JWT).

The EXTERNAL mechanism is mapped to the AuthenticationProviderTls mode of authentication. This is the equivalent of the rabbitmq-auth-mechanism-ssl plugin with ssl_cert_login_from parameter set to common_name.

Multiple Starlight for RabbitMQ proxies can be launched at the same time for scalability and high availability needs. The proxies are stateless and can be started and stopped at will. They share their configuration in Zookeeper so you can create/delete/bind/unbind exchanges and queues on any proxy, and the configuration will be synchronized on the other proxies. Publishing messages can be done on any proxy. On the receiving side, messages will be dispatched evenly to all connected AMQP consumers since the Pulsar subscriptions are shared ones.

Starlight for RabbitMQ offers support for multi-tenancy by mapping an AMQP Virtual host to a Pulsar tenant and namespace. The mapping depends on the following configuration parameters

The mapping is done as follows:

• AMQP vhost / is mapped to Pulsar namespace <amqpDefaultTenant>/<amqpDefaultNamespace>
• AMQP vhost /<vhost> or <vhost> is mapped to Pulsar namespace <amqpDefaultTenant>/<vhost> if amqpMapShortVhostToTenant is false (default) or to <vhost>/<amqpDefaultNamespace> if amqpMapShortVhostToTenant is true
• AMQP vhost /<tenant>/<namespace> or <tenant>/<namespace> is mapped to Pulsar namespace <tenant>/<namespace>

This means that AMQP vhosts must only contain characters that are accepted in Pulsar tenant and namespace names (ie. a-zA-Z0-9_-=:.)

mvn release:prepare -DautoVersionSubmodules -Prelease -Darguments='-DskipTests'

The GitHub release is handled by a GitHub action whenever a tag is being pushed

System tests is a test suite which runs a RabbitMQ client against a real Pulsar Cluster and tests the protocol handler is working correctly.

mvn -f rabbitmq-tests integration-test failsafe:verify -Dgroups=com.datastax.oss.starlight.rabbitmqtests.SystemTest \
   -Dtests.systemtests.enabled=true \
   -Dtests.systemtests.pulsar.host=your-pulsar-broker-or-proxy-hostname \
   -Dtests.systemtests.ampqlistener.port=5672