david-streamlio commented on code in PR #90:
URL: https://github.com/apache/pulsar-connectors/pull/90#discussion_r3572630196
##########
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:
Good catch — this one still stands on the current head. `close()` calls
`rabbitMQChannel.close()` / `rabbitMQConnection.close()` with no null guards,
so if `open()` throws after partial init the framework's `close()` will NPE and
mask the original failure. Mirroring `RabbitMQSink.close()`'s null-safe pattern
would fix it:
```java
@Override
public void close() throws Exception {
if (rabbitMQChannel != null) {
rabbitMQChannel.close();
}
if (rabbitMQConnection != null) {
rabbitMQConnection.close();
}
}
```
##########
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:
Still applies to `testBindToExchangeAndConsumeMessage`: the `finally {
source.close(); }` runs even when the Awaitility `open()` never succeeds, and
since `close()` isn't null-safe (see the `RabbitMQSource.close()` thread) it
will NPE and hide the real timeout. Making `close()` null-safe resolves this
too; alternatively guard the cleanup in the test. The other test
(`testOpenAndWriteSink`) is fine since its `close()` isn't in a `finally`.
--
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]
