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-12432-11e4f5554973a66ba2d63d4df18ee22f10be121f in repository https://gitbox.apache.org/repos/asf/seatunnel.git
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 =
