This is an automated email from the ASF dual-hosted git repository.
github-merge-queue[bot] pushed a commit to branch dev
in repository https://gitbox.apache.org/repos/asf/seatunnel.git
The following commit(s) were added to refs/heads/dev by this push:
new 760c33a944 [Feature][Connector-V2] Add RabbitMQ source connectivity
dry-run (#12432)
760c33a944 is described below
commit 760c33a94433db85b974ae4d6622d05b726a4a47
Author: Goutam Adwant <[email protected]>
AuthorDate: Wed Oct 7 07:37:01 2026 +0000
[Feature][Connector-V2] Add RabbitMQ source connectivity dry-run (#12432)
---
docs/en/connectors/source/Rabbitmq.md | 13 ++
docs/en/engines/zeta/user-command.md | 1 +
docs/zh/connectors/source/Rabbitmq.md | 13 ++
docs/zh/engines/zeta/user-command.md | 1 +
.../seatunnel/rabbitmq/client/RabbitmqClient.java | 7 +-
.../client/RabbitmqSourceDryRunValidator.java | 128 ++++++++++++++
.../seatunnel/rabbitmq/source/RabbitmqSource.java | 59 +------
.../rabbitmq/source/RabbitmqSourceFactory.java | 40 ++++-
.../rabbitmq/source/RabbitmqSourceSchema.java | 95 +++++++++++
.../rabbitmq/RabbitmqSourceFactoryTest.java | 47 ++++++
.../rabbitmq/client/RabbitmqClientTest.java | 187 +++++++++++++++++++++
.../e2e/connector/rabbitmq/RabbitmqIT.java | 184 ++++++++++++++++++++
12 files changed, 714 insertions(+), 61 deletions(-)
diff --git a/docs/en/connectors/source/Rabbitmq.md
b/docs/en/connectors/source/Rabbitmq.md
index 1c50b4b0ba..f677459313 100644
--- a/docs/en/connectors/source/Rabbitmq.md
+++ b/docs/en/connectors/source/Rabbitmq.md
@@ -30,6 +30,19 @@ The source must be non-parallel (parallelism set to 1) in
order to achieve exact
:::
+## Connectivity dry-run
+
+Run `bin/seatunnel.sh --config <your-job.conf> --dry-run connect` with your
own job configuration.
+
+This source supports connection/authentication and existing-queue metadata
checks. It uses the same URI precedence, virtual host, TLS configuration and
configured single-table or `tables_configs` schemas as the normal source,
without creating a reader.
+
+- Each configured queue is checked with a passive declaration. No queue is
created or deleted, no consumer is registered, and no message is read,
acknowledged or published.
+- Queues must already exist, even when `passive = false`. A normal job may
create a missing queue, but this dry-run deliberately cannot. Exclusive queues
owned by another connection cannot be checked.
+- A successful check does **not** prove consumer/read permissions, queue
declaration compatibility, payload format/schema compatibility, or runtime
delivery guarantees. Passive declarations may refresh a queue's expiration
timer.
+- Socket connection, AMQP handshake and channel RPC waits are capped at 10
seconds each; a smaller positive `connection_timeout` is retained. Zero is
bounded for this preflight. Queue checks are sequential, so this is not a
10-second limit for the entire command. DNS resolution follows the JVM resolver.
+- Automatic/topology recovery is disabled only on the temporary connection.
Cleanup aborts that connection with a one-second wait limit, including after
failure. Normal source settings and behavior are unchanged.
+- Validation errors do not include raw connection strings, broker text or
secret-bearing exception causes. Sink connectivity remains unsupported.
+
## Options
| name | type | required | default value |
diff --git a/docs/en/engines/zeta/user-command.md
b/docs/en/engines/zeta/user-command.md
index 53e0ea4141..0b822ff845 100644
--- a/docs/en/engines/zeta/user-command.md
+++ b/docs/en/engines/zeta/user-command.md
@@ -83,6 +83,7 @@ The `--dry-run connect` option runs the static checks first,
then uses connector
| Jdbc | Yes (connectivity + schema inference) | Yes (connectivity +
table existence + field compatibility) |
| Kafka | Yes ([topic metadata + runtime output
schema](../../connectors/source/Kafka.md#connectivity-dry-run), not
consumer/group permissions) | No |
| FakeSource | Yes (schema inference only, no external system) | - |
+| RabbitMQ | Yes ([existing-queue metadata + configured schema; not consumer
permissions](../../connectors/source/Rabbitmq.md#connectivity-dry-run)) | No |
| S3File | Yes (metadata connectivity + inline schema, single-table
text/csv/json/xml; see [supported
scope](../../connectors/source/S3File.md#connectivity-dry-run)) | - |
| Redis | Yes ([connectivity +
authentication](../../connectors/source/Redis.md#connectivity-dry-run) +
configured schema; no key access) | Yes ([connectivity +
authentication](../../connectors/sink/Redis.md#connectivity-dry-run); no field
compatibility or write permission check) |
diff --git a/docs/zh/connectors/source/Rabbitmq.md
b/docs/zh/connectors/source/Rabbitmq.md
index 5975930117..d53e2595ae 100644
--- a/docs/zh/connectors/source/Rabbitmq.md
+++ b/docs/zh/connectors/source/Rabbitmq.md
@@ -30,6 +30,19 @@ import ChangeLog from '../changelog/connector-rabbitmq.md';
:::
+## 连接预检查
+
+使用自己的作业配置执行 `bin/seatunnel.sh --config <your-job.conf> --dry-run connect`。
+
+Source 支持连接、身份认证及已有队列的元数据检查。预检查复用正常 Source 的 URI 优先级、虚拟主机、TLS 配置，以及单表或
`tables_configs` 的 schema 构建逻辑，不创建 reader。
+
+- 对每个配置队列执行被动声明，不创建或删除队列，不注册消费者，不读取、确认或发布消息。
+- 即使 `passive = false`，队列也必须事先存在。正常作业可以创建缺失队列，但预检查不会创建。其他连接持有的独占队列无法检查。
+- 检查成功**不代表**具备消费/读取权限、队列声明属性兼容、消息格式与 schema 匹配，或运行时交付保证。被动声明可能刷新队列的过期计时器。
+- Socket 连接、AMQP 握手及 channel RPC 各自最多等待 10 秒；保留更小的正数
`connection_timeout`，零值在预检查中也受到限制。队列按顺序检查，因此整个命令并非最多 10 秒。DNS 解析由 JVM 解析器控制。
+- 仅临时连接禁用自动恢复和拓扑恢复。成功或失败后均终止该连接，清理等待上限为 1 秒。正常 Source 的配置及行为不变。
+- 校验错误不包含原始连接字符串、服务端原始错误或含凭据的异常原因。Sink 连通性仍不支持。
+
## 选项
| 参数名 | 类型 | 必须 | 默认值 | 描述
|
diff --git a/docs/zh/engines/zeta/user-command.md
b/docs/zh/engines/zeta/user-command.md
index c491ac9cdd..134939bfaa 100644
--- a/docs/zh/engines/zeta/user-command.md
+++ b/docs/zh/engines/zeta/user-command.md
@@ -99,6 +99,7 @@ bin/seatunnel.sh --config
$SEATUNNEL_HOME/config/v2.batch.config.template --dry-
| Jdbc | 支持（连通性 + schema 推断） | 支持（连通性 + 表存在性 + 字段兼容性） |
| Kafka | 支持（[主题元数据 + 运行时输出
schema](../../connectors/source/Kafka.md#连通性-dry-run)，不含消费或消费组权限） | 不支持 |
| FakeSource | 支持（仅 schema 推断，无外部系统） | - |
+| RabbitMQ | 支持（[已有队列元数据 + 配置
schema，不含消费权限](../../connectors/source/Rabbitmq.md#连接预检查)） | 不支持 |
| S3File | 支持（元数据连通性 + 内联 schema，仅单表
text/csv/json/xml；参见[支持范围](../../connectors/source/S3File.md#连接预检查)） | - |
| Redis | 支持（[连通性 + 认证](../../connectors/source/Redis.md#连通性-dry-run) + 配置的
schema，不访问 key） | 支持（[连通性 +
认证](../../connectors/sink/Redis.md#连通性-dry-run)，不校验字段兼容性和写入权限） |
diff --git
a/seatunnel-connectors-v2/connector-rabbitmq/src/main/java/org/apache/seatunnel/connectors/seatunnel/rabbitmq/client/RabbitmqClient.java
b/seatunnel-connectors-v2/connector-rabbitmq/src/main/java/org/apache/seatunnel/connectors/seatunnel/rabbitmq/client/RabbitmqClient.java
index af7ca2ef62..78dddc2d43 100644
---
a/seatunnel-connectors-v2/connector-rabbitmq/src/main/java/org/apache/seatunnel/connectors/seatunnel/rabbitmq/client/RabbitmqClient.java
+++
b/seatunnel-connectors-v2/connector-rabbitmq/src/main/java/org/apache/seatunnel/connectors/seatunnel/rabbitmq/client/RabbitmqClient.java
@@ -62,7 +62,7 @@ public class RabbitmqClient implements AutoCloseable {
public RabbitmqClient(RabbitmqConfig config) {
this.config = config;
try {
- this.connectionFactory = createConnectionFactory();
+ this.connectionFactory = createConnectionFactory(config);
this.connection = connectionFactory.newConnection();
this.channel = connection.createChannel();
@@ -102,7 +102,8 @@ public class RabbitmqClient implements AutoCloseable {
return new QueueingConsumer(channel, queue, splitId);
}
- private ConnectionFactory createConnectionFactory() {
+ // Shared by runtime and preflight so URI precedence, authentication and
TLS stay identical.
+ static ConnectionFactory createConnectionFactory(RabbitmqConfig config) {
ConnectionFactory factory = new ConnectionFactory();
if (StringUtils.isNotEmpty(config.getUri())) {
try {
@@ -159,7 +160,7 @@ public class RabbitmqClient implements AutoCloseable {
}
/** Configures TLS with the JVM trust store and hostname verification
enabled. */
- private void configureSsl(ConnectionFactory factory) {
+ private static void configureSsl(ConnectionFactory factory) {
try {
factory.useSslProtocol(SSLContext.getDefault());
factory.enableHostnameVerification();
diff --git
a/seatunnel-connectors-v2/connector-rabbitmq/src/main/java/org/apache/seatunnel/connectors/seatunnel/rabbitmq/client/RabbitmqSourceDryRunValidator.java
b/seatunnel-connectors-v2/connector-rabbitmq/src/main/java/org/apache/seatunnel/connectors/seatunnel/rabbitmq/client/RabbitmqSourceDryRunValidator.java
new file mode 100644
index 0000000000..8d73e796b8
--- /dev/null
+++
b/seatunnel-connectors-v2/connector-rabbitmq/src/main/java/org/apache/seatunnel/connectors/seatunnel/rabbitmq/client/RabbitmqSourceDryRunValidator.java
@@ -0,0 +1,128 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.seatunnel.connectors.seatunnel.rabbitmq.client;
+
+import org.apache.seatunnel.api.configuration.ReadonlyConfig;
+import
org.apache.seatunnel.connectors.seatunnel.rabbitmq.config.RabbitmqConfig;
+import
org.apache.seatunnel.connectors.seatunnel.rabbitmq.config.RabbitmqSourceOptions;
+
+import com.rabbitmq.client.Channel;
+import com.rabbitmq.client.Connection;
+import com.rabbitmq.client.ConnectionFactory;
+
+import java.io.IOException;
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+import java.util.function.Function;
+
+/** Checks existing queue metadata using an independently owned,
non-recovering connection. */
+public final class RabbitmqSourceDryRunValidator {
+ private static final int MAX_TIMEOUT_MS = 10_000;
+ private static final int CLOSE_TIMEOUT_MS = 1_000;
+
+ private RabbitmqSourceDryRunValidator() {}
+
+ public static void validate(ReadonlyConfig options) throws Exception {
+ validate(options, RabbitmqClient::createConnectionFactory);
+ }
+
+ static void validate(
+ ReadonlyConfig options, Function<RabbitmqConfig,
ConnectionFactory> factoryProvider)
+ throws Exception {
+ if (Thread.currentThread().isInterrupted()) {
+ throw new InterruptedException("RabbitMQ connect dry-run
interrupted");
+ }
+ Connection connection = null;
+ Exception failure = null;
+ String stage = "connection configuration";
+ boolean interrupted = false;
+ try {
+ RabbitmqConfig config = new RabbitmqConfig(options);
+ ConnectionFactory factory = factoryProvider.apply(config);
+ // Zero means infinite in the client. Cap preflight waits without
changing job options.
+ int configuredTimeout = factory.getConnectionTimeout();
+ if (configuredTimeout < 0) {
+ throw new IllegalArgumentException("Invalid connection
timeout");
+ }
+ int timeout =
+ configuredTimeout == 0
+ ? MAX_TIMEOUT_MS
+ : Math.min(configuredTimeout, MAX_TIMEOUT_MS);
+ factory.setConnectionTimeout(timeout);
+ factory.setHandshakeTimeout(timeout);
+ factory.setChannelRpcTimeout(timeout);
+ factory.setShutdownTimeout(CLOSE_TIMEOUT_MS);
+ factory.setAutomaticRecoveryEnabled(false);
+ factory.setTopologyRecoveryEnabled(false);
+
+ stage = "connection or authentication";
+ connection = factory.newConnection();
+ stage = "channel creation";
+ Channel channel = connection.createChannel();
+ if (channel == null) {
+ throw new IOException("No channel available");
+ }
+ List<Map<String, Object>> tables =
+ options.getOptional(RabbitmqSourceOptions.TABLE_CONFIGS)
+ .orElse(
+ Collections.singletonList(
+ Collections.singletonMap(
+
RabbitmqSourceOptions.QUEUE_NAME.key(),
+ config.getQueueName())));
+ stage = "existing queue metadata (pre-create configured queues
before dry-run)";
+ for (Map<String, Object> table : tables) {
+ if (Thread.currentThread().isInterrupted()) {
+ throw new InterruptedException("RabbitMQ connect dry-run
interrupted");
+ }
+ // Never use the runtime declaration path: passive=false may
create queues.
+ channel.queueDeclarePassive(
+
ReadonlyConfig.fromMap(table).get(RabbitmqSourceOptions.QUEUE_NAME));
+ }
+ } catch (InterruptedException e) {
+ interrupted = true;
+ failure = new InterruptedException("RabbitMQ connect dry-run
interrupted");
+ } catch (Exception e) {
+ // Driver/configuration errors can contain URI credentials or
broker-controlled text.
+ failure = new IOException("RabbitMQ connect dry-run could not
validate " + stage);
+ } finally {
+ interrupted |= Thread.interrupted();
+ if (connection != null) {
+ try {
+ // Abort owns channel cleanup too, including a channel
closed by a failed RPC.
+ connection.abort(CLOSE_TIMEOUT_MS);
+ } catch (Exception e) {
+ if (failure == null) {
+ failure =
+ new IOException(
+ "RabbitMQ connect dry-run connection
cleanup failed");
+ }
+ }
+ }
+ if (interrupted) {
+ Thread.currentThread().interrupt();
+ if (failure == null) {
+ failure = new InterruptedException("RabbitMQ connect
dry-run interrupted");
+ }
+ }
+ }
+ if (failure != null) {
+ throw failure;
+ }
+ }
+}
diff --git
a/seatunnel-connectors-v2/connector-rabbitmq/src/main/java/org/apache/seatunnel/connectors/seatunnel/rabbitmq/source/RabbitmqSource.java
b/seatunnel-connectors-v2/connector-rabbitmq/src/main/java/org/apache/seatunnel/connectors/seatunnel/rabbitmq/source/RabbitmqSource.java
index bbf045c82a..003ee25cb6 100644
---
a/seatunnel-connectors-v2/connector-rabbitmq/src/main/java/org/apache/seatunnel/connectors/seatunnel/rabbitmq/source/RabbitmqSource.java
+++
b/seatunnel-connectors-v2/connector-rabbitmq/src/main/java/org/apache/seatunnel/connectors/seatunnel/rabbitmq/source/RabbitmqSource.java
@@ -20,18 +20,14 @@ package
org.apache.seatunnel.connectors.seatunnel.rabbitmq.source;
import org.apache.seatunnel.api.common.JobContext;
import org.apache.seatunnel.api.common.SeaTunnelAPIErrorCode;
import org.apache.seatunnel.api.configuration.ReadonlyConfig;
-import org.apache.seatunnel.api.options.ConnectorCommonOptions;
import org.apache.seatunnel.api.source.Boundedness;
import org.apache.seatunnel.api.source.SeaTunnelSource;
import org.apache.seatunnel.api.source.SourceReader;
import org.apache.seatunnel.api.source.SourceSplitEnumerator;
import org.apache.seatunnel.api.source.SupportParallelism;
import org.apache.seatunnel.api.table.catalog.CatalogTable;
-import org.apache.seatunnel.api.table.catalog.CatalogTableUtil;
-import org.apache.seatunnel.api.table.catalog.TableIdentifier;
import org.apache.seatunnel.api.table.type.SeaTunnelRow;
import org.apache.seatunnel.common.constants.JobMode;
-import
org.apache.seatunnel.connectors.seatunnel.rabbitmq.config.RabbitmqBaseOptions;
import
org.apache.seatunnel.connectors.seatunnel.rabbitmq.config.RabbitmqConfig;
import
org.apache.seatunnel.connectors.seatunnel.rabbitmq.exception.RabbitmqConnectorException;
import org.apache.seatunnel.connectors.seatunnel.rabbitmq.split.RabbitmqSplit;
@@ -64,60 +60,9 @@ public class RabbitmqSource
initializeCatalogTables(config);
}
- /**
- * Parses the configuration to initialize the CatalogTables. Determines
whether the source is
- * operating in Single-Table or Multi-Table mode.
- *
- * @param config The plugin configuration.
- */
private void initializeCatalogTables(ReadonlyConfig config) {
- boolean hasTableConfigs =
-
config.getOptional(ConnectorCommonOptions.TABLE_CONFIGS).isPresent();
- boolean hasSchema =
config.getOptional(ConnectorCommonOptions.SCHEMA).isPresent();
-
- if (hasTableConfigs) {
- // Multi-Table Mode: Parse multiple queue configurations
- List<Map<String, Object>> tableConfigList =
- config.get(ConnectorCommonOptions.TABLE_CONFIGS);
- for (Map<String, Object> item : tableConfigList) {
- ReadonlyConfig tableConfig = ReadonlyConfig.fromMap(item);
- CatalogTable table = buildCatalogTable(tableConfig);
- String queueName =
tableConfig.get(RabbitmqBaseOptions.QUEUE_NAME);
-
- this.catalogTables.add(table);
- this.queueToTableMap.put(queueName, table);
- }
- } else if (hasSchema) {
- CatalogTable table = buildCatalogTable(config);
- String queueName = config.get(RabbitmqBaseOptions.QUEUE_NAME);
- if (queueName == null) {
- queueName = rabbitmqConfig.getQueueName();
- }
- this.catalogTables.add(table);
- this.queueToTableMap.put(queueName, table);
- }
- }
-
- private CatalogTable buildCatalogTable(ReadonlyConfig config) {
- CatalogTable catalogTable = CatalogTableUtil.buildWithConfig(config);
- Map<String, String> options = new HashMap<>(catalogTable.getOptions());
- options.put(
- RabbitmqBaseOptions.FORMAT.key(),
config.get(RabbitmqBaseOptions.FORMAT).name());
- config.getOptional(RabbitmqBaseOptions.PROTOBUF_SCHEMA)
- .ifPresent(value ->
options.put(RabbitmqBaseOptions.PROTOBUF_SCHEMA.key(), value));
- config.getOptional(RabbitmqBaseOptions.PROTOBUF_MESSAGE_NAME)
- .ifPresent(
- value ->
- options.put(
-
RabbitmqBaseOptions.PROTOBUF_MESSAGE_NAME.key(), value));
- return CatalogTable.of(
- TableIdentifier.of(catalogTable.getCatalogName(),
catalogTable.getTablePath()),
- catalogTable.getTableSchema(),
- options,
- catalogTable.getPartitionKeys(),
- catalogTable.getComment(),
- catalogTable.getCatalogName(),
- catalogTable.getMetadataSchema());
+ RabbitmqSourceSchema.initializeCatalogTables(
+ config, rabbitmqConfig, catalogTables, queueToTableMap);
}
@Override
diff --git
a/seatunnel-connectors-v2/connector-rabbitmq/src/main/java/org/apache/seatunnel/connectors/seatunnel/rabbitmq/source/RabbitmqSourceFactory.java
b/seatunnel-connectors-v2/connector-rabbitmq/src/main/java/org/apache/seatunnel/connectors/seatunnel/rabbitmq/source/RabbitmqSourceFactory.java
index 969a3880b8..2a2c4b73ed 100644
---
a/seatunnel-connectors-v2/connector-rabbitmq/src/main/java/org/apache/seatunnel/connectors/seatunnel/rabbitmq/source/RabbitmqSourceFactory.java
+++
b/seatunnel-connectors-v2/connector-rabbitmq/src/main/java/org/apache/seatunnel/connectors/seatunnel/rabbitmq/source/RabbitmqSourceFactory.java
@@ -18,14 +18,19 @@
package org.apache.seatunnel.connectors.seatunnel.rabbitmq.source;
import org.apache.seatunnel.api.configuration.util.Conditions;
+import org.apache.seatunnel.api.configuration.util.ConfigValidator;
import org.apache.seatunnel.api.configuration.util.OptionRule;
import org.apache.seatunnel.api.source.SeaTunnelSource;
import org.apache.seatunnel.api.source.SourceSplit;
+import org.apache.seatunnel.api.table.catalog.CatalogTable;
import org.apache.seatunnel.api.table.connector.TableSource;
import org.apache.seatunnel.api.table.factory.Factory;
+import org.apache.seatunnel.api.table.factory.SupportSourceDryRunValidation;
import org.apache.seatunnel.api.table.factory.TableSourceFactory;
import org.apache.seatunnel.api.table.factory.TableSourceFactoryContext;
+import
org.apache.seatunnel.connectors.seatunnel.rabbitmq.client.RabbitmqSourceDryRunValidator;
import
org.apache.seatunnel.connectors.seatunnel.rabbitmq.config.RabbitmqBaseOptions;
+import
org.apache.seatunnel.connectors.seatunnel.rabbitmq.config.RabbitmqConfig;
import
org.apache.seatunnel.connectors.seatunnel.rabbitmq.config.RabbitmqMessageFormat;
import
org.apache.seatunnel.connectors.seatunnel.rabbitmq.config.RabbitmqSingleTableValidator;
import
org.apache.seatunnel.connectors.seatunnel.rabbitmq.config.RabbitmqSinkOptions;
@@ -34,10 +39,43 @@ import
org.apache.seatunnel.connectors.seatunnel.rabbitmq.config.RabbitmqTableCo
import com.google.auto.service.AutoService;
+import java.io.IOException;
import java.io.Serializable;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
@AutoService(Factory.class)
-public class RabbitmqSourceFactory implements TableSourceFactory {
+public class RabbitmqSourceFactory implements TableSourceFactory,
SupportSourceDryRunValidation {
+
+ /** Reuses runtime catalog construction without opening a reader or a
network connection. */
+ @Override
+ public List<CatalogTable> inferSchemaForDryRun(TableSourceFactoryContext
context)
+ throws IOException {
+ try {
+ ConfigValidator.of(context.getOptions()).validate(optionRule());
+ List<CatalogTable> tables = new ArrayList<>();
+ RabbitmqSourceSchema.initializeCatalogTables(
+ context.getOptions(),
+ new RabbitmqConfig(context.getOptions()),
+ tables,
+ new HashMap<>());
+ return tables;
+ } catch (RuntimeException e) {
+ throw new IOException(
+ "RabbitMQ connect dry-run source configuration or schema
is invalid");
+ }
+ }
+
+ /** Checks connectivity only, without starting the source's data-reading
lifecycle. */
+ @Override
+ public void validateConnectionForDryRun(
+ TableSourceFactoryContext context, List<CatalogTable>
catalogTables) throws Exception {
+ // The connection hook can be called directly, without the preceding
schema hook.
+ inferSchemaForDryRun(context);
+ RabbitmqSourceDryRunValidator.validate(context.getOptions());
+ }
+
@Override
public String factoryIdentifier() {
return "RabbitMQ";
diff --git
a/seatunnel-connectors-v2/connector-rabbitmq/src/main/java/org/apache/seatunnel/connectors/seatunnel/rabbitmq/source/RabbitmqSourceSchema.java
b/seatunnel-connectors-v2/connector-rabbitmq/src/main/java/org/apache/seatunnel/connectors/seatunnel/rabbitmq/source/RabbitmqSourceSchema.java
new file mode 100644
index 0000000000..5a72c0cf76
--- /dev/null
+++
b/seatunnel-connectors-v2/connector-rabbitmq/src/main/java/org/apache/seatunnel/connectors/seatunnel/rabbitmq/source/RabbitmqSourceSchema.java
@@ -0,0 +1,95 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.seatunnel.connectors.seatunnel.rabbitmq.source;
+
+import org.apache.seatunnel.api.configuration.ReadonlyConfig;
+import org.apache.seatunnel.api.options.ConnectorCommonOptions;
+import org.apache.seatunnel.api.table.catalog.CatalogTable;
+import org.apache.seatunnel.api.table.catalog.CatalogTableUtil;
+import org.apache.seatunnel.api.table.catalog.TableIdentifier;
+import
org.apache.seatunnel.connectors.seatunnel.rabbitmq.config.RabbitmqBaseOptions;
+import
org.apache.seatunnel.connectors.seatunnel.rabbitmq.config.RabbitmqConfig;
+
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+/** Pure schema construction shared by the source runtime and connectivity
preflight. */
+final class RabbitmqSourceSchema {
+ private RabbitmqSourceSchema() {}
+
+ /**
+ * Parses the configuration to initialize the CatalogTables. Determines
whether the source is
+ * operating in Single-Table or Multi-Table mode.
+ *
+ * @param config The plugin configuration.
+ */
+ static void initializeCatalogTables(
+ ReadonlyConfig config,
+ RabbitmqConfig rabbitmqConfig,
+ List<CatalogTable> catalogTables,
+ Map<String, CatalogTable> queueToTableMap) {
+ boolean hasTableConfigs =
+
config.getOptional(ConnectorCommonOptions.TABLE_CONFIGS).isPresent();
+ boolean hasSchema =
config.getOptional(ConnectorCommonOptions.SCHEMA).isPresent();
+
+ if (hasTableConfigs) {
+ // Multi-Table Mode: Parse multiple queue configurations
+ List<Map<String, Object>> tableConfigList =
+ config.get(ConnectorCommonOptions.TABLE_CONFIGS);
+ for (Map<String, Object> item : tableConfigList) {
+ ReadonlyConfig tableConfig = ReadonlyConfig.fromMap(item);
+ CatalogTable table = buildCatalogTable(tableConfig);
+ String queueName =
tableConfig.get(RabbitmqBaseOptions.QUEUE_NAME);
+
+ catalogTables.add(table);
+ queueToTableMap.put(queueName, table);
+ }
+ } else if (hasSchema) {
+ CatalogTable table = buildCatalogTable(config);
+ String queueName = config.get(RabbitmqBaseOptions.QUEUE_NAME);
+ if (queueName == null) {
+ queueName = rabbitmqConfig.getQueueName();
+ }
+ catalogTables.add(table);
+ queueToTableMap.put(queueName, table);
+ }
+ }
+
+ private static CatalogTable buildCatalogTable(ReadonlyConfig config) {
+ CatalogTable catalogTable = CatalogTableUtil.buildWithConfig(config);
+ Map<String, String> options = new HashMap<>(catalogTable.getOptions());
+ options.put(
+ RabbitmqBaseOptions.FORMAT.key(),
config.get(RabbitmqBaseOptions.FORMAT).name());
+ config.getOptional(RabbitmqBaseOptions.PROTOBUF_SCHEMA)
+ .ifPresent(value ->
options.put(RabbitmqBaseOptions.PROTOBUF_SCHEMA.key(), value));
+ config.getOptional(RabbitmqBaseOptions.PROTOBUF_MESSAGE_NAME)
+ .ifPresent(
+ value ->
+ options.put(
+
RabbitmqBaseOptions.PROTOBUF_MESSAGE_NAME.key(), value));
+ return CatalogTable.of(
+ TableIdentifier.of(catalogTable.getCatalogName(),
catalogTable.getTablePath()),
+ catalogTable.getTableSchema(),
+ options,
+ catalogTable.getPartitionKeys(),
+ catalogTable.getComment(),
+ catalogTable.getCatalogName(),
+ catalogTable.getMetadataSchema());
+ }
+}
diff --git
a/seatunnel-connectors-v2/connector-rabbitmq/src/test/java/org/apache/seatunnel/connectors/seatunnel/rabbitmq/RabbitmqSourceFactoryTest.java
b/seatunnel-connectors-v2/connector-rabbitmq/src/test/java/org/apache/seatunnel/connectors/seatunnel/rabbitmq/RabbitmqSourceFactoryTest.java
index 4588e83d30..537379f26f 100644
---
a/seatunnel-connectors-v2/connector-rabbitmq/src/test/java/org/apache/seatunnel/connectors/seatunnel/rabbitmq/RabbitmqSourceFactoryTest.java
+++
b/seatunnel-connectors-v2/connector-rabbitmq/src/test/java/org/apache/seatunnel/connectors/seatunnel/rabbitmq/RabbitmqSourceFactoryTest.java
@@ -23,13 +23,18 @@ import
org.apache.seatunnel.api.configuration.util.OptionRule;
import org.apache.seatunnel.api.configuration.util.OptionValidationException;
import org.apache.seatunnel.api.configuration.util.RequiredOption;
import org.apache.seatunnel.api.options.table.TableSchemaOptions;
+import org.apache.seatunnel.api.table.catalog.CatalogTable;
+import org.apache.seatunnel.api.table.factory.SupportSourceDryRunValidation;
+import org.apache.seatunnel.api.table.factory.TableSourceFactoryContext;
import
org.apache.seatunnel.connectors.seatunnel.rabbitmq.config.RabbitmqMessageFormat;
import
org.apache.seatunnel.connectors.seatunnel.rabbitmq.config.RabbitmqSourceOptions;
+import
org.apache.seatunnel.connectors.seatunnel.rabbitmq.source.RabbitmqSource;
import
org.apache.seatunnel.connectors.seatunnel.rabbitmq.source.RabbitmqSourceFactory;
import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.Test;
+import java.util.Arrays;
import java.util.Collections;
import java.util.HashMap;
import java.util.List;
@@ -37,6 +42,48 @@ import java.util.Map;
public class RabbitmqSourceFactoryTest {
+ @Test
+ void dryRunSchemaMatchesRuntimeForAllFormatsAndTableModes() throws
Exception {
+ RabbitmqSourceFactory factory = new RabbitmqSourceFactory();
+ for (RabbitmqMessageFormat format : RabbitmqMessageFormat.values()) {
+ for (Map<String, Object> options :
+ Arrays.asList(createValidSingleTableConfig(),
createValidMultiTableConfig())) {
+ Map<String, Object> table =
+ options.containsKey("tables_configs")
+ ? getFirstTableConfig(options)
+ : options;
+ table.put("format", format);
+ if (format == RabbitmqMessageFormat.PROTOBUF) {
+ table.put(
+ "protobuf_schema",
+ "syntax = \"proto3\"; message Record { int32 id =
1; }");
+ table.put("protobuf_message_name", "Record");
+ }
+ TableSourceFactoryContext context =
+ new TableSourceFactoryContext(
+ ReadonlyConfig.fromMap(options),
getClass().getClassLoader());
+ List<CatalogTable> expected =
+ new
RabbitmqSource(context.getOptions()).getProducedCatalogTables();
+ List<CatalogTable> actual =
factory.inferSchemaForDryRun(context);
+ Assertions.assertEquals(expected.size(), actual.size());
+ for (int i = 0; i < expected.size(); i++) {
+ Assertions.assertEquals(
+ expected.get(i).getTableId(),
actual.get(i).getTableId());
+ Assertions.assertEquals(
+ expected.get(i).getSeaTunnelRowType(),
+ actual.get(i).getSeaTunnelRowType());
+ Assertions.assertEquals(
+ expected.get(i).getOptions(),
actual.get(i).getOptions());
+ }
+ }
+ }
+ }
+
+ @Test
+ void supportsConnectivityDryRun() {
+ Assertions.assertTrue(new RabbitmqSourceFactory() instanceof
SupportSourceDryRunValidation);
+ }
+
@Test
public void testFactoryIdentifier() {
RabbitmqSourceFactory factory = new RabbitmqSourceFactory();
diff --git
a/seatunnel-connectors-v2/connector-rabbitmq/src/test/java/org/apache/seatunnel/connectors/seatunnel/rabbitmq/client/RabbitmqClientTest.java
b/seatunnel-connectors-v2/connector-rabbitmq/src/test/java/org/apache/seatunnel/connectors/seatunnel/rabbitmq/client/RabbitmqClientTest.java
index a2ab086521..c0504fbe8c 100644
---
a/seatunnel-connectors-v2/connector-rabbitmq/src/test/java/org/apache/seatunnel/connectors/seatunnel/rabbitmq/client/RabbitmqClientTest.java
+++
b/seatunnel-connectors-v2/connector-rabbitmq/src/test/java/org/apache/seatunnel/connectors/seatunnel/rabbitmq/client/RabbitmqClientTest.java
@@ -17,25 +17,212 @@
package org.apache.seatunnel.connectors.seatunnel.rabbitmq.client;
+import org.apache.seatunnel.api.configuration.ReadonlyConfig;
import
org.apache.seatunnel.connectors.seatunnel.rabbitmq.config.RabbitmqConfig;
import
org.apache.seatunnel.connectors.seatunnel.rabbitmq.exception.RabbitmqConnectorException;
import org.junit.jupiter.api.Test;
import com.rabbitmq.client.Channel;
+import com.rabbitmq.client.Connection;
+import com.rabbitmq.client.ConnectionFactory;
import java.io.IOException;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.Map;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertNull;
import static org.junit.jupiter.api.Assertions.assertThrows;
import static org.junit.jupiter.api.Assertions.assertTrue;
import static org.mockito.Mockito.doThrow;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.never;
import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.verifyNoInteractions;
+import static org.mockito.Mockito.verifyNoMoreInteractions;
import static org.mockito.Mockito.when;
class RabbitmqClientTest {
+ @Test
+ void dryRunOnlyPassivelyChecksAllQueuesAndClosesConnection() throws
Exception {
+ ConnectionFactory factory = mock(ConnectionFactory.class);
+ Connection connection = mock(Connection.class);
+ Channel channel = mock(Channel.class);
+ when(factory.newConnection()).thenReturn(connection);
+ when(connection.createChannel()).thenReturn(channel);
+ Map<String, Object> options = dryRunOptions();
+ options.remove("queue_name");
+ options.put(
+ "tables_configs",
+ Arrays.asList(
+ Collections.singletonMap("queue_name", "first"),
+ Collections.singletonMap("queue_name", "second")));
+
+
RabbitmqSourceDryRunValidator.validate(ReadonlyConfig.fromMap(options), ignored
-> factory);
+
+ verify(channel).queueDeclarePassive("first");
+ verify(channel).queueDeclarePassive("second");
+ verifyNoMoreInteractions(channel);
+ verify(connection).createChannel();
+ verify(connection).abort(1000);
+ verifyNoMoreInteractions(connection);
+ verify(factory).setConnectionTimeout(10000);
+ verify(factory).setHandshakeTimeout(10000);
+ verify(factory).setChannelRpcTimeout(10000);
+ verify(factory).setAutomaticRecoveryEnabled(false);
+ verify(factory).setTopologyRecoveryEnabled(false);
+ assertFalse((Boolean) options.get("passive"));
+ }
+
+ @Test
+ void dryRunClosesConnectionWhenChannelCreationFailsWithoutExposingCause()
throws Exception {
+ ConnectionFactory factory = mock(ConnectionFactory.class);
+ Connection connection = mock(Connection.class);
+ when(factory.newConnection()).thenReturn(connection);
+ when(connection.createChannel()).thenThrow(new
IOException("amqp://user:secret@host"));
+ IOException failure =
+ assertThrows(
+ IOException.class,
+ () ->
+ RabbitmqSourceDryRunValidator.validate(
+
ReadonlyConfig.fromMap(dryRunOptions()),
+ ignored -> factory));
+ assertTrue(failure.getMessage().contains("channel creation"));
+ assertFalse(failure.getMessage().contains("secret"));
+ assertNull(failure.getCause());
+ assertEquals(0, failure.getSuppressed().length);
+ verify(connection).abort(1000);
+ }
+
+ @Test
+ void dryRunMissingQueueFailsWithoutCreatingItAndClosesConnection() throws
Exception {
+ ConnectionFactory factory = mock(ConnectionFactory.class);
+ Connection connection = mock(Connection.class);
+ Channel channel = mock(Channel.class);
+ when(factory.newConnection()).thenReturn(connection);
+ when(connection.createChannel()).thenReturn(channel);
+ when(channel.queueDeclarePassive("existing"))
+ .thenThrow(new IOException("private broker text"));
+ IOException failure =
+ assertThrows(
+ IOException.class,
+ () ->
+ RabbitmqSourceDryRunValidator.validate(
+
ReadonlyConfig.fromMap(dryRunOptions()),
+ ignored -> factory));
+ assertTrue(failure.getMessage().contains("pre-create"));
+ assertNull(failure.getCause());
+ verify(channel).queueDeclarePassive("existing");
+ verifyNoMoreInteractions(channel);
+ verify(connection).abort(1000);
+ }
+
+ @Test
+ void dryRunHonorsSmallerTimeoutAndRejectsPreInterruptedThread() throws
Exception {
+ ConnectionFactory factory = mock(ConnectionFactory.class);
+ when(factory.getConnectionTimeout()).thenReturn(250);
+ when(factory.newConnection()).thenThrow(new
IOException("unavailable"));
+ assertThrows(
+ IOException.class,
+ () ->
+ RabbitmqSourceDryRunValidator.validate(
+ ReadonlyConfig.fromMap(dryRunOptions()),
ignored -> factory));
+ verify(factory).setConnectionTimeout(250);
+ verify(factory).setHandshakeTimeout(250);
+ verify(factory).setChannelRpcTimeout(250);
+
+ ConnectionFactory unused = mock(ConnectionFactory.class);
+ Thread.currentThread().interrupt();
+ try {
+ assertThrows(
+ InterruptedException.class,
+ () ->
+ RabbitmqSourceDryRunValidator.validate(
+ ReadonlyConfig.fromMap(dryRunOptions()),
ignored -> unused));
+ assertTrue(Thread.currentThread().isInterrupted());
+ verifyNoInteractions(unused);
+ } finally {
+ Thread.interrupted();
+ }
+ }
+
+ @Test
+ void sharedFactoryPreservesUriPrecedenceAndRecoverySettings() {
+ Map<String, Object> options = dryRunOptions();
+ options.put("url",
"amqp://uri-user:uri-password@uri-host:1234/my-vhost");
+ options.put("AUTOMATIC_RECOVERY_ENABLED", true);
+ RabbitmqConfig config = new
RabbitmqConfig(ReadonlyConfig.fromMap(options));
+ ConnectionFactory factory =
RabbitmqClient.createConnectionFactory(config);
+ assertEquals("uri-host", factory.getHost());
+ assertEquals(1234, factory.getPort());
+ assertEquals("uri-user", factory.getUsername());
+ assertEquals("uri-password", factory.getPassword());
+ assertEquals("my-vhost", factory.getVirtualHost());
+ assertTrue(factory.isAutomaticRecoveryEnabled());
+ }
+
+ private static Map<String, Object> dryRunOptions() {
+ Map<String, Object> options = new HashMap<>();
+ options.put("host", "localhost");
+ options.put("port", 5672);
+ options.put("username", "guest");
+ options.put("password", "guest");
+ options.put("queue_name", "existing");
+ options.put("passive", false);
+ return options;
+ }
+
+ @Test
+ void dryRunCleanupFailureIsSanitizedAndCannotPass() throws Exception {
+ ConnectionFactory factory = mock(ConnectionFactory.class);
+ Connection connection = mock(Connection.class);
+ Channel channel = mock(Channel.class);
+ when(factory.newConnection()).thenReturn(connection);
+ when(connection.createChannel()).thenReturn(channel);
+ doThrow(new
IllegalStateException("private-secret")).when(connection).abort(1000);
+ IOException failure =
+ assertThrows(
+ IOException.class,
+ () ->
+ RabbitmqSourceDryRunValidator.validate(
+
ReadonlyConfig.fromMap(dryRunOptions()),
+ ignored -> factory));
+ assertTrue(failure.getMessage().contains("cleanup"));
+ assertFalse(failure.getMessage().contains("private-secret"));
+ assertNull(failure.getCause());
+ }
+
+ @Test
+ void dryRunRestoresInterruptAndClosesAfterQueueCheck() throws Exception {
+ ConnectionFactory factory = mock(ConnectionFactory.class);
+ Connection connection = mock(Connection.class);
+ Channel channel = mock(Channel.class);
+ when(factory.newConnection()).thenReturn(connection);
+ when(connection.createChannel()).thenReturn(channel);
+ when(channel.queueDeclarePassive("existing"))
+ .thenAnswer(
+ ignored -> {
+ Thread.currentThread().interrupt();
+ return null;
+ });
+ try {
+ assertThrows(
+ InterruptedException.class,
+ () ->
+ RabbitmqSourceDryRunValidator.validate(
+ ReadonlyConfig.fromMap(dryRunOptions()),
ignored -> factory));
+ assertTrue(Thread.currentThread().isInterrupted());
+ verify(connection).abort(1000);
+ } finally {
+ Thread.interrupted();
+ }
+ }
+
@Test
void declaresExistingQueuePassivelyWhenConfigured() throws Exception {
Channel channel = mock(Channel.class);
diff --git
a/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-rabbitmq-e2e/src/test/java/org/apache/seatunnel/e2e/connector/rabbitmq/RabbitmqIT.java
b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-rabbitmq-e2e/src/test/java/org/apache/seatunnel/e2e/connector/rabbitmq/RabbitmqIT.java
index faa1d58c3f..5ef10bdeed 100644
---
a/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-rabbitmq-e2e/src/test/java/org/apache/seatunnel/e2e/connector/rabbitmq/RabbitmqIT.java
+++
b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-rabbitmq-e2e/src/test/java/org/apache/seatunnel/e2e/connector/rabbitmq/RabbitmqIT.java
@@ -17,10 +17,18 @@
package org.apache.seatunnel.e2e.connector.rabbitmq;
+import org.apache.seatunnel.shade.com.typesafe.config.ConfigFactory;
+import org.apache.seatunnel.shade.com.typesafe.config.ConfigRenderOptions;
+
+import org.apache.seatunnel.api.configuration.ReadonlyConfig;
import org.apache.seatunnel.api.table.catalog.CatalogTable;
import org.apache.seatunnel.api.table.catalog.PhysicalColumn;
import org.apache.seatunnel.api.table.catalog.TableIdentifier;
import org.apache.seatunnel.api.table.catalog.TableSchema;
+import org.apache.seatunnel.api.table.factory.FactoryUtil;
+import org.apache.seatunnel.api.table.factory.SupportSourceDryRunValidation;
+import org.apache.seatunnel.api.table.factory.TableSourceFactory;
+import org.apache.seatunnel.api.table.factory.TableSourceFactoryContext;
import org.apache.seatunnel.api.table.type.ArrayType;
import org.apache.seatunnel.api.table.type.BasicType;
import org.apache.seatunnel.api.table.type.DecimalType;
@@ -33,6 +41,9 @@ import org.apache.seatunnel.api.table.type.SeaTunnelRowType;
import
org.apache.seatunnel.connectors.seatunnel.rabbitmq.client.RabbitmqClient;
import
org.apache.seatunnel.connectors.seatunnel.rabbitmq.config.RabbitmqConfig;
import
org.apache.seatunnel.connectors.seatunnel.rabbitmq.source.DeliveryMessage;
+import org.apache.seatunnel.core.starter.seatunnel.args.ClientCommandArgs;
+import
org.apache.seatunnel.core.starter.seatunnel.command.SeaTunnelConfValidateCommand;
+import org.apache.seatunnel.core.starter.utils.CommandLineUtils;
import org.apache.seatunnel.e2e.common.TestResource;
import org.apache.seatunnel.e2e.common.TestSuiteBase;
import org.apache.seatunnel.e2e.common.container.TestContainer;
@@ -43,7 +54,9 @@ import
org.apache.seatunnel.format.protobuf.ProtobufSerializationSchema;
import org.junit.jupiter.api.AfterAll;
import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.TestTemplate;
+import org.junit.jupiter.api.io.TempDir;
import org.testcontainers.containers.Container;
import org.testcontainers.containers.GenericContainer;
import org.testcontainers.containers.output.Slf4jLogConsumer;
@@ -53,13 +66,17 @@ import
org.testcontainers.shaded.org.apache.commons.lang3.tuple.Pair;
import org.testcontainers.utility.DockerImageName;
import org.testcontainers.utility.DockerLoggerFactory;
+import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
+import com.rabbitmq.client.ConnectionFactory;
import com.rabbitmq.client.DefaultConsumer;
import lombok.extern.slf4j.Slf4j;
import java.io.IOException;
import java.math.BigDecimal;
import java.nio.charset.StandardCharsets;
+import java.nio.file.Files;
+import java.nio.file.Path;
import java.time.Duration;
import java.time.LocalDate;
import java.time.LocalDateTime;
@@ -131,6 +148,173 @@ public class RabbitmqIT extends TestSuiteBase implements
TestResource {
}
}
+ @Test
+ public void testDryRunLeavesMessagesAndConsumersUnchanged() throws
Exception {
+ try (Connection connection = dryRunConnection();
+ Channel channel = connection.createChannel()) {
+ String queue =
+ channel.queueDeclare(
+ "dry-run-" + java.util.UUID.randomUUID(),
+ false,
+ false,
+ false,
+ null)
+ .getQueue();
+ channel.confirmSelect();
+ channel.basicPublish("", queue, null,
"unchanged".getBytes(StandardCharsets.UTF_8));
+ channel.waitForConfirmsOrDie(5000);
+ int before = channel.queueDeclarePassive(queue).getMessageCount();
+ Map<String, Object> options = dryRunOptions(queue);
+ // Even passive=false must not use the runtime queue creation path.
+ options.put("passive", false);
+ validateDryRun(options);
+ Assertions.assertEquals(before,
channel.queueDeclarePassive(queue).getMessageCount());
+ Assertions.assertEquals(0,
channel.queueDeclarePassive(queue).getConsumerCount());
+ Assertions.assertArrayEquals(
+ "unchanged".getBytes(StandardCharsets.UTF_8),
+ channel.basicGet(queue, true).getBody());
+ }
+ }
+
+ @Test
+ public void testDryRunMultiQueueAndUriPrecedence() throws Exception {
+ try (Connection connection = dryRunConnection();
+ Channel channel = connection.createChannel()) {
+ String first =
+ channel.queueDeclare(
+ "dry-run-" + java.util.UUID.randomUUID(),
+ false,
+ false,
+ false,
+ null)
+ .getQueue();
+ String second =
+ channel.queueDeclare(
+ "dry-run-" + java.util.UUID.randomUUID(),
+ false,
+ false,
+ false,
+ null)
+ .getQueue();
+ Map<String, Object> options = dryRunOptions(first);
+ Object schema = options.remove("schema");
+ options.remove("queue_name");
+ List<Map<String, Object>> tables = new ArrayList<>();
+ for (String queue : Arrays.asList(first, second)) {
+ Map<String, Object> table = new HashMap<>();
+ table.put("queue_name", queue);
+ table.put("schema", schema);
+ tables.add(table);
+ }
+ options.put("tables_configs", tables);
+ options.put(
+ "url",
+ String.format(
+ "amqp://guest:guest@%s:%s/%%2f",
+ rabbitmqContainer.getHost(),
rabbitmqContainer.getMappedPort(PORT)));
+ options.put("host", "unused.invalid");
+ options.put("port", 1);
+ validateDryRun(options);
+ }
+ }
+
+ @Test
+ public void testDryRunMissingQueueIsNotCreated() throws Exception {
+ String queue = "missing-dry-run-" + java.util.UUID.randomUUID();
+ Assertions.assertThrows(IOException.class, () ->
validateDryRun(dryRunOptions(queue)));
+ try (Connection connection = dryRunConnection()) {
+ // A failed passive declaration closes its channel; only the
connection needs closing.
+ Channel channel = connection.createChannel();
+ Assertions.assertThrows(IOException.class, () ->
channel.queueDeclarePassive(queue));
+ }
+ }
+
+ @Test
+ public void testDryRunRejectsInvalidCredentialsWithoutEchoingThem() {
+ Map<String, Object> options = dryRunOptions("unused");
+ options.put("password", "dry-run-private-password");
+ Exception failure =
+ Assertions.assertThrows(IOException.class, () ->
validateDryRun(options));
+
Assertions.assertFalse(failure.toString().contains("dry-run-private-password"));
+ Assertions.assertNull(failure.getCause());
+ }
+
+ private Connection dryRunConnection() throws Exception {
+ ConnectionFactory factory = new ConnectionFactory();
+ factory.setHost(rabbitmqContainer.getHost());
+ factory.setPort(rabbitmqContainer.getMappedPort(PORT));
+ factory.setUsername(USERNAME);
+ factory.setPassword(PASSWORD);
+ return factory.newConnection();
+ }
+
+ private Map<String, Object> dryRunOptions(String queue) {
+ Map<String, Object> options = new HashMap<>();
+ options.put("host", rabbitmqContainer.getHost());
+ options.put("port", rabbitmqContainer.getMappedPort(PORT));
+ options.put("username", USERNAME);
+ options.put("password", PASSWORD);
+ options.put("queue_name", queue);
+ options.put(
+ "schema",
+ Collections.singletonMap("fields",
Collections.singletonMap("name", "string")));
+ return options;
+ }
+
+ @Test
+ public void testDryRunCommand(@TempDir Path directory) throws Exception {
+ try (Connection connection = dryRunConnection();
+ Channel channel = connection.createChannel()) {
+ String queue =
+ channel.queueDeclare(
+ "cli-dry-run-" +
java.util.UUID.randomUUID(),
+ false,
+ false,
+ false,
+ null)
+ .getQueue();
+ Map<String, Object> options = dryRunOptions(queue);
+ checkDryRunCommand(options, directory);
+ Assertions.assertEquals(0,
channel.queueDeclarePassive(queue).getConsumerCount());
+ }
+ }
+
+ private void checkDryRunCommand(Map<String, Object> options, Path
directory) throws Exception {
+ options.put("plugin_name", "RabbitMQ");
+ options.put("plugin_output", "preflight");
+ Map<String, Object> sink = new HashMap<>();
+ sink.put("plugin_name", "Console");
+ sink.put("plugin_input", "preflight");
+ Map<String, Object> job = new HashMap<>();
+ job.put("source", Collections.singletonList(options));
+ job.put("sink", Collections.singletonList(sink));
+ Path file = directory.resolve("dry-run.json");
+ Files.write(
+ file,
+ ConfigFactory.parseMap(job)
+ .root()
+ .render(ConfigRenderOptions.concise())
+ .getBytes(StandardCharsets.UTF_8));
+ ClientCommandArgs args =
+ CommandLineUtils.parse(
+ new String[] {"-c", file.toString(), "--dry-run",
"connect"},
+ new ClientCommandArgs(),
+ "seatunnel.sh",
+ true);
+ new SeaTunnelConfValidateCommand(args).execute();
+ }
+
+ private void validateDryRun(Map<String, Object> options) throws Exception {
+ ClassLoader loader = getClass().getClassLoader();
+ TableSourceFactory factory =
+ FactoryUtil.discoverFactory(loader, TableSourceFactory.class,
"RabbitMQ");
+ Assertions.assertTrue(factory instanceof
SupportSourceDryRunValidation);
+ SupportSourceDryRunValidation validator =
(SupportSourceDryRunValidation) factory;
+ TableSourceFactoryContext context =
+ new TableSourceFactoryContext(ReadonlyConfig.fromMap(options),
loader);
+ validator.validateConnectionForDryRun(context,
validator.inferSchemaForDryRun(context));
+ }
+
private static Pair<SeaTunnelRowType, List<SeaTunnelRow>>
generateTestDataSet() {
SeaTunnelRowType rowType =
