efcasado opened a new pull request, #90: URL: https://github.com/apache/pulsar-connectors/pull/90
### Motivation Years ago, we worked on a project that required us to connect to to an external AMQP source. All our services were already using Pulsar as a source, and we were not particularly happy with the introduction of this asymmetry. We thought we could leverage the official RabbitMQ source connector, but very soon we run into issues that required us to fork it and roll out our own version. This fixes were sitting in our own fork, until today. We believe these small improvements may be of interest to the wider Pulsar community. In a few words, the requirements we had (and still have) were as follows > You cannot create your own queues. Instead you have to request a server-named queue (empty queue name in the request). Passive, Exclusive, Non-durable. The problem is that the current implementation of the RabbitMQ source connector hardcodes some of these values (see [here](https://github.com/apache/pulsar-connectors/blob/518742a3fe7507d0dcde7495063dc603a885c93c/rabbitmq/src/main/java/org/apache/pulsar/io/rabbitmq/RabbitMQSource.java#L70)). More concretely, a RabbitMQ queue's [exclusive](https://www.rabbitmq.com/docs/queues#exclusive-queues), [durable](https://www.rabbitmq.com/docs/queues#durability) and `auto-delete` were hardcoded to `false`, `false` and `false`, respectively, and couldn't be configured. Because of the requirements we shared above, we needed to be able to set `exclusive` and `auto-delete` to `true`. We also needed to be able to bind a queue to a particular [exchange](https://www.rabbitmq.com/docs/exchanges) using a configurable [routing key](https://www.rabbitmq.com/docs/exchanges#topic). None of this was supported by the RabbitMQ source connector. In our use case, we also needed for the consuming services to be able to access some of the metadata from the AMQP endpoint as if they were connecting directly to the AMQP endpoint. Last but not least, the current implementation of the RabbitMQ source connector does not support connections to secure AMQP endpoints. Below is an example of how the RabbitMQ source connector could be configured once these changes are merged ``` configs: host: "<AMQP-HOST>" port: 5671 ssl: true virtualHost: "<AMQP-VIRTUAL-HOST>" username: "<AMQP-USERNAME>" password: "" queueName: "" durable: false exclusive: true autoDelete: true exchangeName: "<EXCHANGE-NAME>" routingKey: "<ROUTING-KEY>" connectionName: "<AMQP-CONNECTION-NAME>" ``` ### Modifications `RabbitMQSource.java` has been extended to allow users to override the previously hardcoded values of important RabbitMQ settings such as `exchange`, `routingKey`, `durable`, `autoDelete` and `exclusive`. The emitted Pulsar message now includes some AMQP metadata, i.e. [queue name](https://www.rabbitmq.com/docs/queues#names) and [consumer tag](https://www.rabbitmq.com/docs/consumers#consumer-tags), encoded as message properties. ### Verifying this change - [x] Make sure that the change passes the CI checks. This change added tests and can be verified as follows: - Extended the existing `RabbitMQSourceConfigTest` and `RabbitMQSourceTest` tests with additional tests to confirm the new configuration parameters are correctly applied and the AMQP fields are correctly propagated to the resulting Pulsar message. ### Documentation - [ ] `doc` <!-- Your PR contains doc changes. --> - [ ] `doc-required` <!-- Your PR changes impact docs and you will update later --> - [x] `doc-not-needed` <!-- Your PR changes do not impact docs --> - [ ] `doc-complete` <!-- Docs have been already added --> ### Matching PR in forked repository PR in forked repository: https://github.com/efcasado/pulsar-connectors/pull/3 -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
