This is an automated email from the ASF dual-hosted git repository. github-merge-queue[bot] pushed a commit to branch gh-readonly-queue/dev/pr-12519-acb0b5739c62fe43dd68e0d923fed1143bb52ce8 in repository https://gitbox.apache.org/repos/asf/seatunnel.git
commit 463adf8c47aab1f9da8f135121917281870d9b97 Author: Yiğitcan Öztürk <[email protected]> AuthorDate: Tue Sep 29 13:26:31 2026 +0000 [Improve][Connector-V2] Validate RabbitMQ sink queue_name as nonblank (#12519) --- docs/en/connectors/sink/Rabbitmq.md | 2 +- docs/zh/connectors/sink/Rabbitmq.md | 2 +- .../rabbitmq/sink/RabbitmqSinkFactory.java | 7 +++++-- .../rabbitmq/RabbitmqSinkFactoryTest.java | 23 ++++++++++++++++++++++ 4 files changed, 30 insertions(+), 4 deletions(-) diff --git a/docs/en/connectors/sink/Rabbitmq.md b/docs/en/connectors/sink/Rabbitmq.md index bab1147b37..f5d431bd4f 100644 --- a/docs/en/connectors/sink/Rabbitmq.md +++ b/docs/en/connectors/sink/Rabbitmq.md @@ -85,7 +85,7 @@ When `url` uses an `amqps://` URI, the broker certificate is verified against th ### queue_name [string] -the queue to write the message to. If `routing_key` is not configured, the connector publishes messages to this queue through the default exchange. +the queue to write the message to. The value must not be empty or whitespace-only. If `routing_key` is not configured, the connector publishes messages to this queue through the default exchange. ### format [string] diff --git a/docs/zh/connectors/sink/Rabbitmq.md b/docs/zh/connectors/sink/Rabbitmq.md index 12b9222797..4d0674c862 100644 --- a/docs/zh/connectors/sink/Rabbitmq.md +++ b/docs/zh/connectors/sink/Rabbitmq.md @@ -86,7 +86,7 @@ virtual host，连接 broker 使用的 vhost ### queue_name [string] -数据写入的队列名。如果没有配置 `routing_key`，连接器会通过默认 exchange 将消息直接写入该队列。 +数据写入的队列名。该值不能为空或仅包含空白字符。如果没有配置 `routing_key`，连接器会通过默认 exchange 将消息直接写入该队列。 ### format [string] diff --git a/seatunnel-connectors-v2/connector-rabbitmq/src/main/java/org/apache/seatunnel/connectors/seatunnel/rabbitmq/sink/RabbitmqSinkFactory.java b/seatunnel-connectors-v2/connector-rabbitmq/src/main/java/org/apache/seatunnel/connectors/seatunnel/rabbitmq/sink/RabbitmqSinkFactory.java index c41a0ea930..ff991c8543 100644 --- a/seatunnel-connectors-v2/connector-rabbitmq/src/main/java/org/apache/seatunnel/connectors/seatunnel/rabbitmq/sink/RabbitmqSinkFactory.java +++ b/seatunnel-connectors-v2/connector-rabbitmq/src/main/java/org/apache/seatunnel/connectors/seatunnel/rabbitmq/sink/RabbitmqSinkFactory.java @@ -17,6 +17,7 @@ package org.apache.seatunnel.connectors.seatunnel.rabbitmq.sink; +import org.apache.seatunnel.api.configuration.util.Conditions; import org.apache.seatunnel.api.configuration.util.OptionRule; import org.apache.seatunnel.api.table.connector.TableSink; import org.apache.seatunnel.api.table.factory.Factory; @@ -43,8 +44,10 @@ public class RabbitmqSinkFactory implements TableSinkFactory { .required( RabbitmqSinkOptions.HOST, RabbitmqSinkOptions.PORT, - RabbitmqSinkOptions.VIRTUAL_HOST, - RabbitmqSinkOptions.QUEUE_NAME) + RabbitmqSinkOptions.VIRTUAL_HOST) + .required( + RabbitmqSinkOptions.QUEUE_NAME, + Conditions.notBlank(RabbitmqSinkOptions.QUEUE_NAME)) .bundled(RabbitmqSinkOptions.USERNAME, RabbitmqSinkOptions.PASSWORD) .optional(RabbitmqSinkOptions.FORMAT) .conditional( diff --git a/seatunnel-connectors-v2/connector-rabbitmq/src/test/java/org/apache/seatunnel/connectors/seatunnel/rabbitmq/RabbitmqSinkFactoryTest.java b/seatunnel-connectors-v2/connector-rabbitmq/src/test/java/org/apache/seatunnel/connectors/seatunnel/rabbitmq/RabbitmqSinkFactoryTest.java index 16840e374e..1b261eec4e 100644 --- a/seatunnel-connectors-v2/connector-rabbitmq/src/test/java/org/apache/seatunnel/connectors/seatunnel/rabbitmq/RabbitmqSinkFactoryTest.java +++ b/seatunnel-connectors-v2/connector-rabbitmq/src/test/java/org/apache/seatunnel/connectors/seatunnel/rabbitmq/RabbitmqSinkFactoryTest.java @@ -38,6 +38,29 @@ public class RabbitmqSinkFactoryTest { .validate(rabbitmqSinkFactory.optionRule()); } + @Test + public void testQueueNameValidation() { + Map<String, Object> validConfig = createValidConfig(); + Assertions.assertDoesNotThrow(() -> validate(validConfig)); + + Map<String, Object> missingQueueName = createValidConfig(); + missingQueueName.remove(RabbitmqSinkOptions.QUEUE_NAME.key()); + Assertions.assertThrows(OptionValidationException.class, () -> validate(missingQueueName)); + + Map<String, Object> emptyQueueName = createValidConfig(); + emptyQueueName.put(RabbitmqSinkOptions.QUEUE_NAME.key(), ""); + Assertions.assertThrows(OptionValidationException.class, () -> validate(emptyQueueName)); + + Map<String, Object> whitespaceQueueName = createValidConfig(); + whitespaceQueueName.put(RabbitmqSinkOptions.QUEUE_NAME.key(), " "); + Assertions.assertThrows( + OptionValidationException.class, () -> validate(whitespaceQueueName)); + + Map<String, Object> paddedQueueName = createValidConfig(); + paddedQueueName.put(RabbitmqSinkOptions.QUEUE_NAME.key(), " test_queue "); + Assertions.assertDoesNotThrow(() -> validate(paddedQueueName)); + } + @Test public void testValidProtobufConfig() { Map<String, Object> config = createValidConfig();
