Copilot commented on code in PR #90:
URL: https://github.com/apache/pulsar-connectors/pull/90#discussion_r3560264146
##########
rabbitmq/src/main/java/org/apache/pulsar/io/rabbitmq/RabbitMQSource.java:
##########
@@ -97,6 +113,9 @@ public RabbitMQConsumer(RabbitMQSource source, Channel
channel) {
public void handleDelivery(String consumerTag, Envelope envelope,
AMQP.BasicProperties properties, byte[] body)
throws IOException {
long deliveryTag = envelope.getDeliveryTag();
+ Map<String, String> pulsarProperties = new HashMap<>();
+ pulsarProperties.put("consumerTag", consumerTag);
+ pulsarProperties.put("queueName", queueName);
Review Comment:
The injected Pulsar message property keys ("consumerTag" and "queueName")
are very generic and may collide with user-defined properties. Other source
connectors in this repo namespace connector-added metadata with a "__" prefix
(e.g. Kafka uses "__kafka_topic"), so RabbitMQ metadata should be similarly
namespaced to avoid accidental overrides.
##########
rabbitmq/src/main/java/org/apache/pulsar/io/rabbitmq/RabbitMQSourceConfig.java:
##########
@@ -63,6 +63,39 @@ public class RabbitMQSourceConfig extends
RabbitMQAbstractConfig implements Seri
help = "Set true if the queue should be declared passively - ie to
preserve durability/timeout settings")
private boolean passive = false;
+ @FieldDoc(
+ required = false,
+ defaultValue = "false",
+ help = "Set true if the queue should survive a broker restart")
+ private boolean durable = false;
+
+ @FieldDoc(
+ required = false,
+ defaultValue = "false",
+ help = "Set true if the queue can be used by only one connection
and get deleted when it closes")
+ private boolean exclusive = false;
+
+ @FieldDoc(
+ required = false,
+ defaultValue = "false",
+ help = "Set true if the queue should get deleted when the last
consumer unsubscribes")
+ private boolean autoDelete = false;
+
+ @FieldDoc(
+ required = false,
+ defaultValue = "",
+ help = "The name of an existing exchange to bind the queue to. The
exchange must already exist; "
+ + "it is not declared by the connector. Leave empty to consume
directly from the queue "
+ + "without binding to an exchange.")
+ private String exchangeName;
+
+ @FieldDoc(
+ required = false,
+ defaultValue = "#",
+ help = "The routing key used when binding the queue to the
exchange. "
+ + "Only used when exchangeName is set.")
+ private String routingKey = "#";
Review Comment:
New exchange binding support relies on routingKey being non-null when
exchangeName is set, and passive declarations require a non-empty queueName.
Adding explicit validation for these new configuration combinations prevents
runtime NPEs / broker errors and yields a clearer configuration error message.
##########
rabbitmq/src/test/java/org/apache/pulsar/io/rabbitmq/source/RabbitMQSourceTest.java:
##########
@@ -47,13 +53,67 @@ public void tearDown() {
@Test
public void testOpenAndWriteSink() throws Exception {
+ Map<String, Object> configs = baseConfigs("test-queue");
+
+ RabbitMQSource source = new RabbitMQSource();
+
+ // open should success
+ // rabbitmq service may need time to initialize
+ SourceContext sourceContext = mock(SourceContext.class);
+ Awaitility.await().ignoreExceptions().pollDelay(Duration.ofSeconds(1))
+ .untilAsserted(() -> source.open(configs, sourceContext));
+ source.close();
+ }
+
+ @Test(timeOut = 60000)
+ public void testBindToExchangeAndConsumeMessage() throws Exception {
+ String exchange = "test-exchange";
+ String queue = "test-bind-queue";
+ String routingKey = "test.key";
+
+ ConnectionFactory factory = new ConnectionFactory();
+ factory.setHost("localhost");
+ factory.setPort(rabbitMQBrokerManager.getPort());
+ factory.setUsername(rabbitMQBrokerManager.getUser());
+ factory.setPassword(rabbitMQBrokerManager.getPassword());
+ factory.setVirtualHost("default");
+
+ RabbitMQSource source = new RabbitMQSource();
+ SourceContext sourceContext = mock(SourceContext.class);
+
+ try (Connection connection = factory.newConnection();
+ Channel channel = connection.createChannel()) {
+ // the exchange must already exist before the source binds the
queue to it
+ channel.exchangeDeclare(exchange, "topic", true);
+
+ Map<String, Object> configs = baseConfigs(queue);
+ configs.put("exchangeName", exchange);
+ configs.put("routingKey", routingKey);
+ configs.put("durable", "true");
+
+ // rabbitmq service may need time to initialize
+
Awaitility.await().ignoreExceptions().pollDelay(Duration.ofSeconds(1))
+ .untilAsserted(() -> source.open(configs, sourceContext));
+
+ channel.basicPublish(exchange, routingKey, null,
"hello".getBytes(StandardCharsets.UTF_8));
+
+ Record<byte[]> record = source.read();
+ assertEquals("hello", new String(record.getValue(),
StandardCharsets.UTF_8));
+ assertEquals(queue, record.getProperties().get("queueName"));
+ record.ack();
Review Comment:
This assertion uses the old un-namespaced RabbitMQ metadata property key
("queueName"). If the source starts namespacing connector-added properties
(recommended to avoid collisions), this test should assert on the new
namespaced key.
##########
rabbitmq/src/test/java/org/apache/pulsar/io/rabbitmq/source/RabbitMQSourceTest.java:
##########
@@ -47,13 +53,67 @@ public void tearDown() {
@Test
public void testOpenAndWriteSink() throws Exception {
+ Map<String, Object> configs = baseConfigs("test-queue");
+
+ RabbitMQSource source = new RabbitMQSource();
+
+ // open should success
+ // rabbitmq service may need time to initialize
+ SourceContext sourceContext = mock(SourceContext.class);
+ Awaitility.await().ignoreExceptions().pollDelay(Duration.ofSeconds(1))
+ .untilAsserted(() -> source.open(configs, sourceContext));
+ source.close();
+ }
+
+ @Test(timeOut = 60000)
+ public void testBindToExchangeAndConsumeMessage() throws Exception {
+ String exchange = "test-exchange";
+ String queue = "test-bind-queue";
+ String routingKey = "test.key";
+
+ ConnectionFactory factory = new ConnectionFactory();
+ factory.setHost("localhost");
+ factory.setPort(rabbitMQBrokerManager.getPort());
+ factory.setUsername(rabbitMQBrokerManager.getUser());
+ factory.setPassword(rabbitMQBrokerManager.getPassword());
+ factory.setVirtualHost("default");
+
+ RabbitMQSource source = new RabbitMQSource();
+ SourceContext sourceContext = mock(SourceContext.class);
+
+ try (Connection connection = factory.newConnection();
+ Channel channel = connection.createChannel()) {
+ // the exchange must already exist before the source binds the
queue to it
+ channel.exchangeDeclare(exchange, "topic", true);
+
+ Map<String, Object> configs = baseConfigs(queue);
+ configs.put("exchangeName", exchange);
+ configs.put("routingKey", routingKey);
+ configs.put("durable", "true");
+
+ // rabbitmq service may need time to initialize
+
Awaitility.await().ignoreExceptions().pollDelay(Duration.ofSeconds(1))
+ .untilAsserted(() -> source.open(configs, sourceContext));
+
+ channel.basicPublish(exchange, routingKey, null,
"hello".getBytes(StandardCharsets.UTF_8));
+
+ Record<byte[]> record = source.read();
+ assertEquals("hello", new String(record.getValue(),
StandardCharsets.UTF_8));
+ assertEquals(queue, record.getProperties().get("queueName"));
+ record.ack();
+ } finally {
+ source.close();
+ }
Review Comment:
In this test, source.close() is always called in the finally block. If
source.open() never succeeds (e.g., Awaitility times out), close() can throw
(e.g., NPE due to uninitialized connection/channel) and obscure the real
failure. Make the test's cleanup resilient by swallowing close failures or by
tracking whether open succeeded.
##########
rabbitmq/src/main/java/org/apache/pulsar/io/rabbitmq/RabbitMQSource.java:
##########
@@ -64,19 +66,33 @@ public void open(Map<String, Object> config, SourceContext
sourceContext) throws
rabbitMQConnection.getPort()
);
rabbitMQChannel = rabbitMQConnection.createChannel();
+ AMQP.Queue.DeclareOk queueDeclaration;
if (rabbitMQSourceConfig.isPassive()) {
-
rabbitMQChannel.queueDeclarePassive(rabbitMQSourceConfig.getQueueName());
+ queueDeclaration =
rabbitMQChannel.queueDeclarePassive(rabbitMQSourceConfig.getQueueName());
} else {
- rabbitMQChannel.queueDeclare(rabbitMQSourceConfig.getQueueName(),
false, false, false, null);
+ queueDeclaration =
rabbitMQChannel.queueDeclare(rabbitMQSourceConfig.getQueueName(),
+
rabbitMQSourceConfig.isDurable(),
+
rabbitMQSourceConfig.isExclusive(),
+
rabbitMQSourceConfig.isAutoDelete(),
+ null
+ );
}
+ queueName = queueDeclaration.getQueue();
logger.info("Setting channel.basicQos({}, {}).",
rabbitMQSourceConfig.getPrefetchCount(),
rabbitMQSourceConfig.isPrefetchGlobal()
);
rabbitMQChannel.basicQos(rabbitMQSourceConfig.getPrefetchCount(),
rabbitMQSourceConfig.isPrefetchGlobal());
+ String exchange = rabbitMQSourceConfig.getExchangeName();
+ String routingKey = rabbitMQSourceConfig.getRoutingKey();
+ if (exchange != null && !exchange.isEmpty()) {
+ // note: the exchange is expected to already exist; it is not
declared here
+ rabbitMQChannel.queueBind(queueName, exchange, routingKey);
+ logger.info("Bound queue {} to exchange {} with routing key {}.",
queueName, exchange, routingKey);
+ }
com.rabbitmq.client.Consumer consumer = new RabbitMQConsumer(this,
rabbitMQChannel);
- rabbitMQChannel.basicConsume(rabbitMQSourceConfig.getQueueName(),
consumer);
- logger.info("A consumer for queue {} has been successfully started.",
rabbitMQSourceConfig.getQueueName());
+ rabbitMQChannel.basicConsume(queueName, consumer);
+ logger.info("A consumer for queue {} has been successfully started.",
queueName);
}
Review Comment:
If open() fails after partially initializing the connection/channel (e.g.,
queueDeclare/queueBind/basicConsume throws), the framework may still call
close(). close() currently dereferences rabbitMQChannel/rabbitMQConnection
without null checks, which can cause NPEs and mask the original failure. Making
close() null-safe (like RabbitMQSink.close()) improves reliability and test
cleanup.
--
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]
