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-12686-8ceebc754c6f4b63efb925b56a29cdcafc06b5b6 in repository https://gitbox.apache.org/repos/asf/seatunnel.git
commit 0f8913a8c3f30dc5bb8cab07bda91365a8811cb5 Author: Goutam Adwant <[email protected]> AuthorDate: Sun Oct 11 09:46:04 2026 +0000 [Feature][Connector-V2] Add ClickHouse sink connectivity dry-run (#12686) Signed-off-by: Goutam Adwant <[email protected]> --- docs/en/connectors/sink/Clickhouse.md | 24 +++ docs/en/engines/zeta/user-command.md | 1 + docs/zh/connectors/sink/Clickhouse.md | 19 ++ docs/zh/engines/zeta/user-command.md | 1 + .../exception/ClickhouseConnectorErrorCode.java | 3 +- .../sink/client/ClickhouseSinkFactory.java | 118 ++++++++++- .../clickhouse/ClickhouseFactoryTest.java | 227 +++++++++++++++++++++ .../seatunnel/clickhouse/ClickhouseIT.java | 128 ++++++++++++ 8 files changed, 519 insertions(+), 2 deletions(-) diff --git a/docs/en/connectors/sink/Clickhouse.md b/docs/en/connectors/sink/Clickhouse.md index 123eb0ee3b..0a248fa4f3 100644 --- a/docs/en/connectors/sink/Clickhouse.md +++ b/docs/en/connectors/sink/Clickhouse.md @@ -24,6 +24,30 @@ import ChangeLog from '../changelog/connector-clickhouse.md'; Used to write data to Clickhouse. +## Connectivity dry-run + +`--dry-run connect` checks the first configured `host`, using the sink's connection +settings and a read-only query against `system.tables`. It does not create a writer, +read business rows, run `custom_sql`, or execute schema/data save modes. + +- With `schema_save_mode = ERROR_WHEN_SCHEMA_NOT_EXIST`, the target table must be + visible in `system.tables`. A missing or invisible target fails validation. +- Other schema save modes allow a missing table. A missing database is also allowed + for creation modes; validation does not create either object or verify creation permissions. +- Connect and socket timeouts are capped at 10 seconds, preserving shorter positive + values. Server query time is capped at 10 seconds, retries/failover are disabled, + and the check does not reuse a configured session. These overrides apply only to dry-run. +- `clickhouse.config.custom_http_params` is not supported by dry-run because it can + add SQL to the request. A custom `X-ClickHouse-Query-Id` header is also rejected to + avoid reusing a running query's identifier. An empty username, or a non-default username with an empty + password, is rejected: the current sink cannot resolve those credentials without + falling back to the driver's default account. The default user with an empty password + is supported. + +Successful validation confirms this metadata connection, not field/type compatibility, +write or DDL permissions, `ERROR_WHEN_DATA_EXISTS`, custom SQL correctness, or connectivity +to every shard in `split_mode`. Clickhouse source and ClickhouseFile sink are unchanged. + ## Supported DataSource Info In order to use the Clickhouse connector, the following dependencies are required. diff --git a/docs/en/engines/zeta/user-command.md b/docs/en/engines/zeta/user-command.md index 3ddb44734f..40ba238697 100644 --- a/docs/en/engines/zeta/user-command.md +++ b/docs/en/engines/zeta/user-command.md @@ -88,6 +88,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) | Yes ([metadata + local serializer checks](../../connectors/sink/Kafka.md#connectivity-dry-run); dynamic topics have limited checks, Produce/transaction permissions are not checked) | | FakeSource | Yes (schema inference only, no external system) | - | +| Clickhouse | No | Yes ([initial-host connectivity + target metadata](../../connectors/sink/Clickhouse.md#connectivity-dry-run); no write permission or field compatibility check) | | MongoDB | No | Yes ([connectivity + configured authentication](../../connectors/sink/MongoDB.md#connectivity-dry-run); no collection, schema, write permission or transaction checks) | | Neo4j | Yes ([driver connectivity + configured schema; not database/query validation](../../connectors/source/Neo4j.md#connectivity-dry-run)) | No | | RabbitMQ | Yes ([existing-queue metadata + configured schema; not consumer permissions](../../connectors/source/Rabbitmq.md#connectivity-dry-run)) | No | diff --git a/docs/zh/connectors/sink/Clickhouse.md b/docs/zh/connectors/sink/Clickhouse.md index a92b131d13..b825f8ea45 100644 --- a/docs/zh/connectors/sink/Clickhouse.md +++ b/docs/zh/connectors/sink/Clickhouse.md @@ -24,6 +24,25 @@ import ChangeLog from '../changelog/connector-clickhouse.md'; 用于将数据写入 Clickhouse。 +## 连通性 dry-run + +`--dry-run connect` 使用 sink 的连接配置检查第一个 `host`，并对 `system.tables` +执行只读元数据查询。它不会创建 writer、读取业务数据、执行 `custom_sql` 或执行任何表结构/数据保存模式。 + +- 当 `schema_save_mode = ERROR_WHEN_SCHEMA_NOT_EXIST` 时，目标表必须在 `system.tables` + 中可见。表不存在或不可见都会导致校验失败。 +- 其他表结构保存模式允许目标表不存在。创建模式也允许数据库不存在；校验不会创建数据库或表，也不验证创建权限。 +- 连接和 socket 超时最多为 10 秒，保留更短的正值配置。服务端查询超时最多为 10 秒， + 禁用重试和故障转移，不复用配置的 session。这些限制仅适用于 dry-run。 +- dry-run 不支持 `clickhouse.config.custom_http_params`，因为它可以在请求中附加 SQL。 + 也不支持自定义 `X-ClickHouse-Query-Id` 请求头，避免复用正在执行的查询标识。 + 空用户名或密码为空的非 default 用户也会被拒绝：当前 sink 无法解析这些凭据而不回退到驱动的默认账户。 + 支持密码为空的 default 用户。 + +校验成功仅确认元数据连接可用，不保证字段/类型兼容性、写入或 DDL 权限、 +`ERROR_WHEN_DATA_EXISTS` 条件、自定义 SQL 的正确性，或 `split_mode` 下所有分片的连通性。 +Clickhouse source 和 ClickhouseFile sink 的行为不变。 + ## 支持的数据源信息 为了使用 Clickhouse 连接器，需要以下依赖项。它们可以通过 install-plugin.sh 或从 Maven 中央存储库下载。 diff --git a/docs/zh/engines/zeta/user-command.md b/docs/zh/engines/zeta/user-command.md index 87f66b5493..c206b90198 100644 --- a/docs/zh/engines/zeta/user-command.md +++ b/docs/zh/engines/zeta/user-command.md @@ -102,6 +102,7 @@ bin/seatunnel.sh --config $SEATUNNEL_HOME/config/v2.batch.config.template --dry- | Jdbc | 支持（连通性 + schema 推断） | 支持（连通性 + 表存在性 + 字段兼容性） | | Kafka | 支持（[主题元数据 + 运行时输出 schema](../../connectors/source/Kafka.md#连通性-dry-run)，不含消费或消费组权限） | 支持（[元数据 + 本地序列化检查](../../connectors/sink/Kafka.md#连通性-dry-run)；动态主题仅执行有限检查，不检查 Produce/事务权限） | | FakeSource | 支持（仅 schema 推断，无外部系统） | - | +| Clickhouse | 不支持 | 支持（[首个 host 的连通性 + 目标元数据](../../connectors/sink/Clickhouse.md#连通性-dry-run)，不校验写入权限或字段兼容性） | | MongoDB | 不支持 | 支持（[连通性 + 配置的认证](../../connectors/sink/MongoDB.md#连通性-dry-run)，不检查集合、schema、写入权限或事务） | | Neo4j | 支持（[驱动连通性 + 配置 schema，不含数据库/查询校验](../../connectors/source/Neo4j.md#连接预检查)） | 不支持 | | RabbitMQ | 支持（[已有队列元数据 + 配置 schema，不含消费权限](../../connectors/source/Rabbitmq.md#连接预检查)） | 不支持 | diff --git a/seatunnel-connectors-v2/connector-clickhouse/src/main/java/org/apache/seatunnel/connectors/seatunnel/clickhouse/exception/ClickhouseConnectorErrorCode.java b/seatunnel-connectors-v2/connector-clickhouse/src/main/java/org/apache/seatunnel/connectors/seatunnel/clickhouse/exception/ClickhouseConnectorErrorCode.java index b9b99657e7..3794b748e5 100644 --- a/seatunnel-connectors-v2/connector-clickhouse/src/main/java/org/apache/seatunnel/connectors/seatunnel/clickhouse/exception/ClickhouseConnectorErrorCode.java +++ b/seatunnel-connectors-v2/connector-clickhouse/src/main/java/org/apache/seatunnel/connectors/seatunnel/clickhouse/exception/ClickhouseConnectorErrorCode.java @@ -43,7 +43,8 @@ public enum ClickhouseConnectorErrorCode implements SeaTunnelErrorCode { COMPLEX_SQL_NOT_SUPPORT_PARALLEL_ERROR( "CLICKHOUSE-15", "Complex sql not support parallel read."), ROW_BATCH_GET_FAILED("CLICKHOUSE-16", "Row batch get error"), - GET_TABLE_LIST_CONFIG_ERROR("CLICKHOUSE-17", "Get table list config error."); + GET_TABLE_LIST_CONFIG_ERROR("CLICKHOUSE-17", "Get table list config error."), + DRY_RUN_VALIDATION_FAILED("CLICKHOUSE-18", "Connect dry-run validation failed."); private final String code; private final String description; diff --git a/seatunnel-connectors-v2/connector-clickhouse/src/main/java/org/apache/seatunnel/connectors/seatunnel/clickhouse/sink/client/ClickhouseSinkFactory.java b/seatunnel-connectors-v2/connector-clickhouse/src/main/java/org/apache/seatunnel/connectors/seatunnel/clickhouse/sink/client/ClickhouseSinkFactory.java index d01982307e..c24197e15b 100644 --- a/seatunnel-connectors-v2/connector-clickhouse/src/main/java/org/apache/seatunnel/connectors/seatunnel/clickhouse/sink/client/ClickhouseSinkFactory.java +++ b/seatunnel-connectors-v2/connector-clickhouse/src/main/java/org/apache/seatunnel/connectors/seatunnel/clickhouse/sink/client/ClickhouseSinkFactory.java @@ -17,18 +17,34 @@ package org.apache.seatunnel.connectors.seatunnel.clickhouse.sink.client; +import org.apache.seatunnel.api.common.SeaTunnelAPIErrorCode; import org.apache.seatunnel.api.configuration.ReadonlyConfig; import org.apache.seatunnel.api.configuration.util.OptionRule; import org.apache.seatunnel.api.options.SinkConnectorCommonOptions; import org.apache.seatunnel.api.sink.DataSaveMode; +import org.apache.seatunnel.api.sink.SchemaSaveMode; import org.apache.seatunnel.api.table.catalog.CatalogTable; import org.apache.seatunnel.api.table.connector.TableSink; import org.apache.seatunnel.api.table.factory.Factory; +import org.apache.seatunnel.api.table.factory.SupportSinkDryRunValidation; import org.apache.seatunnel.api.table.factory.TableSinkFactory; import org.apache.seatunnel.api.table.factory.TableSinkFactoryContext; +import org.apache.seatunnel.connectors.seatunnel.clickhouse.exception.ClickhouseConnectorErrorCode; +import org.apache.seatunnel.connectors.seatunnel.clickhouse.exception.ClickhouseConnectorException; +import org.apache.seatunnel.connectors.seatunnel.clickhouse.util.ClickhouseUtil; +import com.clickhouse.client.ClickHouseClient; +import com.clickhouse.client.ClickHouseFormat; +import com.clickhouse.client.ClickHouseNode; +import com.clickhouse.client.ClickHouseResponse; +import com.clickhouse.client.ClickHouseUtils; +import com.clickhouse.client.config.ClickHouseClientOption; +import com.clickhouse.client.http.config.ClickHouseHttpOption; import com.google.auto.service.AutoService; +import java.util.HashMap; +import java.util.Map; + import static org.apache.seatunnel.connectors.seatunnel.clickhouse.config.ClickhouseBaseOptions.CLICKHOUSE_CONFIG; import static org.apache.seatunnel.connectors.seatunnel.clickhouse.config.ClickhouseBaseOptions.DATABASE; import static org.apache.seatunnel.connectors.seatunnel.clickhouse.config.ClickhouseBaseOptions.HOST; @@ -48,7 +64,9 @@ import static org.apache.seatunnel.connectors.seatunnel.clickhouse.config.Clickh import static org.apache.seatunnel.connectors.seatunnel.clickhouse.config.ClickhouseSinkOptions.TABLE; @AutoService(Factory.class) -public class ClickhouseSinkFactory implements TableSinkFactory { +public class ClickhouseSinkFactory implements TableSinkFactory, SupportSinkDryRunValidation { + private static final int DRY_RUN_TIMEOUT_MILLIS = 10_000; + @Override public String factoryIdentifier() { return "Clickhouse"; @@ -61,6 +79,104 @@ public class ClickhouseSinkFactory implements TableSinkFactory { return () -> new ClickhouseSink(catalogTable, readonlyConfig); } + /** Checks the runtime's initial server using system metadata, without invoking save modes. */ + @Override + public void validateConnectionForDryRun(TableSinkFactoryContext context) { + if (Thread.currentThread().isInterrupted()) { + throw new ClickhouseConnectorException( + ClickhouseConnectorErrorCode.DRY_RUN_VALIDATION_FAILED, + "Clickhouse sink dry-run was interrupted before connecting."); + } + ReadonlyConfig config = context.getOptions(); + String username = config.get(USERNAME); + String password = config.get(PASSWORD); + // createNodes only installs credentials when both fields are nonempty. Do not report + // validation of a named account when the runtime would fall back to the default account. + if (username.isEmpty() || (!"default".equals(username) && password.isEmpty())) { + throw new ClickhouseConnectorException( + ClickhouseConnectorErrorCode.DRY_RUN_VALIDATION_FAILED, + "Clickhouse sink dry-run requires a nonempty username and a nonempty password" + + " for non-default users; the sink cannot resolve these credentials."); + } + Map<String, String> options = new HashMap<>(config.get(CLICKHOUSE_CONFIG)); + if (!options.getOrDefault(ClickHouseHttpOption.CUSTOM_PARAMS.getKey(), "").isEmpty()) { + // HTTP parameters can supply an additional query, bypassing our metadata-only query. + throw new ClickhouseConnectorException( + ClickhouseConnectorErrorCode.DRY_RUN_VALIDATION_FAILED, + "Clickhouse sink dry-run does not support clickhouse.config.custom_http_params."); + } + if (ClickHouseUtils.getKeyValuePairs( + options.getOrDefault(ClickHouseHttpOption.CUSTOM_HEADERS.getKey(), "")) + .keySet().stream() + .anyMatch("X-ClickHouse-Query-Id"::equalsIgnoreCase)) { + throw new ClickhouseConnectorException( + ClickhouseConnectorErrorCode.DRY_RUN_VALIDATION_FAILED, + "Clickhouse sink dry-run does not support X-ClickHouse-Query-Id in custom_http_headers."); + } + boolean tableExists; + try { + capDryRunTimeout( + options, ClickHouseClientOption.CONNECTION_TIMEOUT, DRY_RUN_TIMEOUT_MILLIS); + capDryRunTimeout( + options, ClickHouseClientOption.SOCKET_TIMEOUT, DRY_RUN_TIMEOUT_MILLIS); + capDryRunTimeout(options, ClickHouseClientOption.MAX_EXECUTION_TIME, 10); + options.put(ClickHouseClientOption.RETRY.getKey(), "0"); + options.put(ClickHouseClientOption.FAILOVER.getKey(), "0"); + options.put(ClickHouseClientOption.REPEAT_ON_SESSION_LOCK.getKey(), "false"); + options.put(ClickHouseClientOption.SESSION_ID.getKey(), ""); + options.put(ClickHouseClientOption.ASYNC.getKey(), "false"); + // Do not select the target database: a create-schema mode can create it later. + options.put(ClickHouseClientOption.DATABASE.getKey(), ""); + ClickHouseNode node = + ClickhouseUtil.createNodes( + config.get(HOST), + "", + config.get(SERVER_TIME_ZONE), + username, + password, + options) + .get(0); + try (ClickHouseClient client = ClickHouseClient.newInstance(node.getProtocol()); + ClickHouseResponse response = + client.connect(node) + .format(ClickHouseFormat.RowBinaryWithNamesAndTypes) + .query( + "SELECT count() FROM system.tables" + + " WHERE database = :database AND name = :table") + // Object[] binds literal values; the String[] overload accepts + // SQL. + .params(new Object[] {config.get(DATABASE), config.get(TABLE)}) + .executeAndWait()) { + tableExists = response.firstRecord().getValue(0).asInteger() > 0; + } + } catch (Exception e) { + // Driver messages and suppressed close failures may include credentials or HTTP + // options. + throw new ClickhouseConnectorException( + ClickhouseConnectorErrorCode.DRY_RUN_VALIDATION_FAILED, + "Clickhouse sink dry-run failed. Check host, username, password, TLS settings," + + " timeouts, and access to system.tables."); + } + if (!tableExists + && config.get(SCHEMA_SAVE_MODE) == SchemaSaveMode.ERROR_WHEN_SCHEMA_NOT_EXIST) { + throw new ClickhouseConnectorException( + SeaTunnelAPIErrorCode.SINK_TABLE_NOT_EXIST, + "Clickhouse sink table is missing or not visible in system.tables and" + + " schema_save_mode is ERROR_WHEN_SCHEMA_NOT_EXIST." + + " Check the database and table options."); + } + } + + private static void capDryRunTimeout( + Map<String, String> options, ClickHouseClientOption option, int maximum) { + int configured = + Integer.parseInt( + options.getOrDefault(option.getKey(), option.getDefaultValue().toString())); + options.put( + option.getKey(), + Integer.toString(configured > 0 ? Math.min(configured, maximum) : maximum)); + } + @Override public OptionRule optionRule() { return OptionRule.builder() diff --git a/seatunnel-connectors-v2/connector-clickhouse/src/test/java/org/apache/seatunnel/connectors/seatunnel/clickhouse/ClickhouseFactoryTest.java b/seatunnel-connectors-v2/connector-clickhouse/src/test/java/org/apache/seatunnel/connectors/seatunnel/clickhouse/ClickhouseFactoryTest.java index 7d68a83add..2afdc559f3 100644 --- a/seatunnel-connectors-v2/connector-clickhouse/src/test/java/org/apache/seatunnel/connectors/seatunnel/clickhouse/ClickhouseFactoryTest.java +++ b/seatunnel-connectors-v2/connector-clickhouse/src/test/java/org/apache/seatunnel/connectors/seatunnel/clickhouse/ClickhouseFactoryTest.java @@ -20,19 +20,246 @@ package org.apache.seatunnel.connectors.seatunnel.clickhouse; import org.apache.seatunnel.api.configuration.ReadonlyConfig; import org.apache.seatunnel.api.configuration.util.ConfigValidator; import org.apache.seatunnel.api.configuration.util.OptionValidationException; +import org.apache.seatunnel.api.sink.SchemaSaveMode; +import org.apache.seatunnel.api.table.factory.SupportSinkDryRunValidation; +import org.apache.seatunnel.api.table.factory.TableSinkFactoryContext; import org.apache.seatunnel.connectors.seatunnel.clickhouse.config.ClickhouseBaseOptions; +import org.apache.seatunnel.connectors.seatunnel.clickhouse.exception.ClickhouseConnectorException; import org.apache.seatunnel.connectors.seatunnel.clickhouse.sink.client.ClickhouseSinkFactory; import org.apache.seatunnel.connectors.seatunnel.clickhouse.sink.file.ClickhouseFileSinkFactory; import org.apache.seatunnel.connectors.seatunnel.clickhouse.source.ClickhouseSourceFactory; import org.junit.jupiter.api.Assertions; import org.junit.jupiter.api.Test; +import org.mockito.ArgumentCaptor; +import org.mockito.MockedStatic; +import org.mockito.Mockito; +import com.clickhouse.client.ClickHouseClient; +import com.clickhouse.client.ClickHouseConfig; +import com.clickhouse.client.ClickHouseNode; +import com.clickhouse.client.ClickHouseRecord; +import com.clickhouse.client.ClickHouseRequest; +import com.clickhouse.client.ClickHouseResponse; +import com.clickhouse.client.ClickHouseValue; + +import java.io.PrintWriter; +import java.io.StringWriter; +import java.util.Collections; import java.util.HashMap; import java.util.Map; +import java.util.concurrent.CompletableFuture; +import java.util.concurrent.TimeUnit; + +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.ArgumentMatchers.anyLong; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.never; +import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.when; public class ClickhouseFactoryTest { + @Test + public void testSinkSupportsConnectDryRun() { + Assertions.assertTrue(new ClickhouseSinkFactory() instanceof SupportSinkDryRunValidation); + } + + @Test + public void testSinkDryRunUsesMetadataAndClosesResources() throws Exception { + Map<String, Object> config = createValidSinkConfig(); + config.put("table", "quoted'\\table"); + Map<String, String> properties = new HashMap<>(); + properties.put("connect_timeout", "250"); + properties.put("socket_timeout", "0"); + properties.put("retry", "100"); + properties.put("max_execution_time", "2"); + properties.put("failover", "100"); + properties.put("ssl", "true"); + properties.put("sslrootcert", "/test-only/ca.pem"); + properties.put("session_id", "existing-session"); + config.put("clickhouse.config", properties); + ClickHouseClient client = mock(ClickHouseClient.class); + ClickHouseRequest<?> request = mock(ClickHouseRequest.class, Mockito.RETURNS_SELF); + ClickHouseResponse response = metadataResponse(1); + when(client.connect(any(ClickHouseNode.class))).thenAnswer(invocation -> request); + when(request.executeAndWait()).thenReturn(response); + try (MockedStatic<ClickHouseClient> clients = Mockito.mockStatic(ClickHouseClient.class)) { + clients.when(() -> ClickHouseClient.newInstance(any())).thenReturn(client); + validateSinkDryRun(config); + } + ArgumentCaptor<ClickHouseNode> node = ArgumentCaptor.forClass(ClickHouseNode.class); + verify(client).connect(node.capture()); + Assertions.assertEquals("localhost", node.getValue().getHost()); + Assertions.assertEquals(8123, node.getValue().getPort()); + Assertions.assertEquals("250", node.getValue().getOptions().get("connect_timeout")); + Assertions.assertEquals("10000", node.getValue().getOptions().get("socket_timeout")); + Assertions.assertEquals("0", node.getValue().getOptions().get("retry")); + Assertions.assertEquals("2", node.getValue().getOptions().get("max_execution_time")); + Assertions.assertEquals("0", node.getValue().getOptions().get("failover")); + Assertions.assertEquals("", node.getValue().getOptions().get("session_id")); + Assertions.assertEquals("true", node.getValue().getOptions().get("ssl")); + Assertions.assertEquals( + "/test-only/ca.pem", node.getValue().getOptions().get("sslrootcert")); + Assertions.assertEquals("100", properties.get("retry")); + verify(request) + .query( + "SELECT count() FROM system.tables WHERE database = :database AND name = :table"); + verify(request).params(new Object[] {"default", "quoted'\\table"}); + verify(response).close(); + verify(client).close(); + } + + @Test + public void testSinkDryRunRespectsSchemaSaveModesWithoutCreatingTables() throws Exception { + for (SchemaSaveMode mode : SchemaSaveMode.values()) { + Map<String, Object> config = createValidSinkConfig(); + config.put("schema_save_mode", mode.name()); + config.put("data_save_mode", "CUSTOM_PROCESSING"); + config.put("custom_sql", "DROP TABLE must_not_execute"); + ClickHouseClient client = mock(ClickHouseClient.class); + ClickHouseRequest<?> request = mock(ClickHouseRequest.class, Mockito.RETURNS_SELF); + ClickHouseResponse response = metadataResponse(0); + when(client.connect(any(ClickHouseNode.class))).thenAnswer(invocation -> request); + when(request.executeAndWait()).thenReturn(response); + try (MockedStatic<ClickHouseClient> clients = + Mockito.mockStatic(ClickHouseClient.class)) { + clients.when(() -> ClickHouseClient.newInstance(any())).thenReturn(client); + if (mode == SchemaSaveMode.ERROR_WHEN_SCHEMA_NOT_EXIST) { + ClickhouseConnectorException error = + Assertions.assertThrows( + ClickhouseConnectorException.class, + () -> validateSinkDryRun(config)); + Assertions.assertTrue( + error.getMessage().contains("ERROR_WHEN_SCHEMA_NOT_EXIST")); + } else { + validateSinkDryRun(config); + } + } + verify(request, never()).query("DROP TABLE must_not_execute"); + verify(response).close(); + verify(client).close(); + } + } + + @Test + public void testSinkDryRunSanitizesQueryAndCloseFailures() throws Exception { + for (boolean failQuery : new boolean[] {true, false}) { + ClickHouseClient client = mock(ClickHouseClient.class); + ClickHouseRequest<?> request = mock(ClickHouseRequest.class, Mockito.RETURNS_SELF); + ClickHouseResponse response = metadataResponse(0); + when(client.connect(any(ClickHouseNode.class))).thenAnswer(invocation -> request); + if (failQuery) { + when(request.executeAndWait()) + .thenThrow(new IllegalStateException("private-password")); + } else { + when(request.executeAndWait()).thenReturn(response); + Mockito.doThrow(new IllegalStateException("private-password")) + .when(response) + .close(); + } + Mockito.doThrow(new IllegalStateException("private-token")).when(client).close(); + try (MockedStatic<ClickHouseClient> clients = + Mockito.mockStatic(ClickHouseClient.class)) { + clients.when(() -> ClickHouseClient.newInstance(any())).thenReturn(client); + ClickhouseConnectorException error = + Assertions.assertThrows( + ClickhouseConnectorException.class, + () -> validateSinkDryRun(createValidSinkConfig())); + StringWriter trace = new StringWriter(); + error.printStackTrace(new PrintWriter(trace)); + Assertions.assertFalse(trace.toString().contains("private-password")); + Assertions.assertFalse(trace.toString().contains("private-token")); + Assertions.assertTrue(error.getMessage().contains("access to system.tables")); + } + verify(client).close(); + } + } + + @Test + public void testSinkDryRunRejectsUnsafeOptionsAndUnresolvedCredentialsBeforeConnecting() { + try (MockedStatic<ClickHouseClient> clients = Mockito.mockStatic(ClickHouseClient.class)) { + Map<String, Object> config = createValidSinkConfig(); + config.put( + "clickhouse.config", + Collections.singletonMap("custom_http_params", "query=SELECT 1")); + Assertions.assertThrows( + ClickhouseConnectorException.class, () -> validateSinkDryRun(config)); + config.put( + "clickhouse.config", + Collections.singletonMap( + "custom_http_headers", "x-clickhouse-query-id=existing-query")); + Assertions.assertThrows( + ClickhouseConnectorException.class, () -> validateSinkDryRun(config)); + config.remove("clickhouse.config"); + config.put("username", "non-default-user"); + config.put("password", ""); + Assertions.assertThrows( + ClickhouseConnectorException.class, () -> validateSinkDryRun(config)); + config.put("username", ""); + config.put("password", "password"); + Assertions.assertThrows( + ClickhouseConnectorException.class, () -> validateSinkDryRun(config)); + clients.verifyNoInteractions(); + } + } + + @Test + public void testSinkDryRunPreservesInterruption() throws Exception { + try (MockedStatic<ClickHouseClient> clients = Mockito.mockStatic(ClickHouseClient.class)) { + Thread.currentThread().interrupt(); + try { + Assertions.assertThrows( + ClickhouseConnectorException.class, + () -> validateSinkDryRun(createValidSinkConfig())); + Assertions.assertTrue(Thread.currentThread().isInterrupted()); + clients.verifyNoInteractions(); + } finally { + Thread.interrupted(); + } + ClickHouseClient client = mock(ClickHouseClient.class, Mockito.CALLS_REAL_METHODS); + when(client.getConfig()).thenReturn(new ClickHouseConfig()); + CompletableFuture<ClickHouseResponse> response = mock(CompletableFuture.class); + when(response.get(anyLong(), any(TimeUnit.class))) + .thenThrow(new InterruptedException("interrupted")); + when(client.execute(any(ClickHouseRequest.class))).thenReturn(response); + clients.when(() -> ClickHouseClient.newInstance(any())).thenReturn(client); + try { + Assertions.assertThrows( + ClickhouseConnectorException.class, + () -> validateSinkDryRun(createValidSinkConfig())); + Assertions.assertTrue(Thread.currentThread().isInterrupted()); + } finally { + Thread.interrupted(); + } + verify(client).close(); + } + } + + private ClickHouseResponse metadataResponse(int count) { + ClickHouseResponse response = mock(ClickHouseResponse.class); + ClickHouseRecord record = mock(ClickHouseRecord.class); + ClickHouseValue value = mock(ClickHouseValue.class); + when(response.firstRecord()).thenReturn(record); + when(record.getValue(0)).thenReturn(value); + when(value.asInteger()).thenReturn(count); + return response; + } + + private void validateSinkDryRun(Map<String, Object> config) { + new ClickhouseSinkFactory() + .validateConnectionForDryRun( + new TableSinkFactoryContext( + null, ReadonlyConfig.fromMap(config), getClass().getClassLoader())); + } + + private Map<String, Object> createValidSinkConfig() { + Map<String, Object> config = createValidSourceConfig(); + config.put("database", "default"); + config.put("table", "target"); + return config; + } + private void validateSource(Map<String, Object> configMap) { ClickhouseSourceFactory factory = new ClickhouseSourceFactory(); ConfigValidator.of(ReadonlyConfig.fromMap(configMap)).validate(factory.optionRule()); diff --git a/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-clickhouse-e2e/src/test/java/org/apache/seatunnel/connectors/seatunnel/clickhouse/ClickhouseIT.java b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-clickhouse-e2e/src/test/java/org/apache/seatunnel/connectors/seatunnel/clickhouse/ClickhouseIT.java index 87fe215152..b0a9b51dd4 100644 --- a/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-clickhouse-e2e/src/test/java/org/apache/seatunnel/connectors/seatunnel/clickhouse/ClickhouseIT.java +++ b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-clickhouse-e2e/src/test/java/org/apache/seatunnel/connectors/seatunnel/clickhouse/ClickhouseIT.java @@ -18,9 +18,11 @@ package org.apache.seatunnel.connectors.seatunnel.clickhouse; import org.apache.seatunnel.api.configuration.ReadonlyConfig; +import org.apache.seatunnel.api.sink.SchemaSaveMode; import org.apache.seatunnel.api.table.catalog.CatalogTable; import org.apache.seatunnel.api.table.catalog.Column; import org.apache.seatunnel.api.table.catalog.TablePath; +import org.apache.seatunnel.api.table.factory.TableSinkFactoryContext; import org.apache.seatunnel.api.table.type.ArrayType; import org.apache.seatunnel.api.table.type.BasicType; import org.apache.seatunnel.api.table.type.DecimalType; @@ -30,6 +32,8 @@ import org.apache.seatunnel.api.table.type.SeaTunnelDataType; import org.apache.seatunnel.api.table.type.SeaTunnelRow; import org.apache.seatunnel.api.table.type.SeaTunnelRowType; import org.apache.seatunnel.connectors.seatunnel.clickhouse.catalog.ClickhouseCatalog; +import org.apache.seatunnel.connectors.seatunnel.clickhouse.exception.ClickhouseConnectorException; +import org.apache.seatunnel.connectors.seatunnel.clickhouse.sink.client.ClickhouseSinkFactory; import org.apache.seatunnel.e2e.common.TestResource; import org.apache.seatunnel.e2e.common.TestSuiteBase; import org.apache.seatunnel.e2e.common.container.TestContainer; @@ -39,6 +43,7 @@ import org.awaitility.Awaitility; 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.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -57,6 +62,8 @@ import java.io.File; import java.io.IOException; import java.io.InputStream; import java.math.BigDecimal; +import java.net.InetAddress; +import java.net.ServerSocket; import java.nio.charset.StandardCharsets; import java.sql.Array; import java.sql.Connection; @@ -67,6 +74,7 @@ import java.sql.ResultSet; import java.sql.SQLException; import java.sql.Statement; import java.sql.Timestamp; +import java.time.Duration; import java.time.LocalDate; import java.time.LocalDateTime; import java.util.ArrayList; @@ -110,6 +118,125 @@ public class ClickhouseIT extends TestSuiteBase implements TestResource { private static final String FIX_PARTITION_DATE = "2025-06-17"; + @Test + public void testSinkConnectDryRunDoesNotExecuteSaveModes() throws Exception { + String table = "dry_run_target"; + try (Statement statement = connection.createStatement()) { + statement.execute("CREATE TABLE default." + table + " (id Int32) ENGINE = Memory"); + statement.execute("INSERT INTO default." + table + " VALUES (7)"); + try { + for (SchemaSaveMode mode : SchemaSaveMode.values()) { + Map<String, Object> options = dryRunSinkOptions(table); + options.put("schema_save_mode", mode.name()); + options.put("data_save_mode", "DROP_DATA"); + validateSinkDryRun(options); + options.put("data_save_mode", "CUSTOM_PROCESSING"); + options.put("custom_sql", "DROP TABLE default." + table); + validateSinkDryRun(options); + Assertions.assertEquals(1, countData(table)); + } + try (ResultSet rows = statement.executeQuery("SELECT id FROM default." + table)) { + Assertions.assertTrue(rows.next()); + Assertions.assertEquals(7, rows.getInt(1)); + Assertions.assertFalse(rows.next()); + } + } finally { + statement.execute("DROP TABLE default." + table); + } + } + } + + @Test + public void testSinkConnectDryRunAllowsCreationWithoutCreatingTarget() throws Exception { + Map<String, Object> options = dryRunSinkOptions("dry_run_missing_table"); + for (SchemaSaveMode mode : + Arrays.asList( + SchemaSaveMode.CREATE_SCHEMA_WHEN_NOT_EXIST, + SchemaSaveMode.RECREATE_SCHEMA, + SchemaSaveMode.IGNORE)) { + options.put("schema_save_mode", mode.name()); + validateSinkDryRun(options); + } + options.put("schema_save_mode", SchemaSaveMode.ERROR_WHEN_SCHEMA_NOT_EXIST.name()); + ClickhouseConnectorException error = + Assertions.assertThrows( + ClickhouseConnectorException.class, () -> validateSinkDryRun(options)); + Assertions.assertTrue(error.getMessage().contains("ERROR_WHEN_SCHEMA_NOT_EXIST")); + try (Statement statement = connection.createStatement(); + ResultSet tables = + statement.executeQuery("EXISTS TABLE default.dry_run_missing_table")) { + Assertions.assertTrue(tables.next()); + Assertions.assertEquals(0, tables.getInt(1)); + } + options.put("database", "dry_run_missing_database"); + options.put("schema_save_mode", SchemaSaveMode.CREATE_SCHEMA_WHEN_NOT_EXIST.name()); + validateSinkDryRun(options); + try (Statement statement = connection.createStatement(); + ResultSet databases = + statement.executeQuery( + "SELECT count() FROM system.databases WHERE name = 'dry_run_missing_database'")) { + Assertions.assertTrue(databases.next()); + Assertions.assertEquals(0, databases.getInt(1)); + } + } + + @Test + public void testSinkConnectDryRunRejectsAuthenticationAndConnectionFailures() throws Exception { + Map<String, Object> options = dryRunSinkOptions(SINK_TABLE); + try (Statement statement = connection.createStatement()) { + statement.execute( + "CREATE USER seatunnel_dry_run_reader IDENTIFIED BY 'test-only-password'"); + try { + statement.execute("GRANT SELECT ON *.* TO seatunnel_dry_run_reader"); + options.put("username", "seatunnel_dry_run_reader"); + options.put("password", "test-only-password"); + options.put("schema_save_mode", SchemaSaveMode.ERROR_WHEN_SCHEMA_NOT_EXIST.name()); + // Metadata access does not require INSERT/CREATE/DROP privileges. + validateSinkDryRun(options); + } finally { + statement.execute("DROP USER seatunnel_dry_run_reader"); + } + } + options.put("username", container.getUsername()); + options.put("password", "dry-run-invalid-password"); + ClickhouseConnectorException error = + Assertions.assertThrows( + ClickhouseConnectorException.class, () -> validateSinkDryRun(options)); + Assertions.assertFalse(error.getMessage().contains("dry-run-invalid-password")); + Assertions.assertNull(error.getCause()); + try (ServerSocket unavailable = new ServerSocket(0, 1, InetAddress.getLoopbackAddress())) { + options.put("host", "localhost:" + unavailable.getLocalPort()); + options.put("clickhouse.config", Collections.singletonMap("socket_timeout", "200")); + // An accepting-but-unresponsive endpoint exercises the socket deadline without port + // races. + Assertions.assertTimeoutPreemptively( + Duration.ofSeconds(5), + () -> + Assertions.assertThrows( + ClickhouseConnectorException.class, + () -> validateSinkDryRun(options))); + } + } + + private Map<String, Object> dryRunSinkOptions(String table) { + Map<String, Object> options = new HashMap<>(); + options.put("host", container.getHost() + ":" + container.getMappedPort(8123)); + options.put("username", container.getUsername()); + options.put("password", container.getPassword()); + options.put("database", DATABASE); + options.put("table", table); + return options; + } + + private void validateSinkDryRun(Map<String, Object> options) { + new ClickhouseSinkFactory() + .validateConnectionForDryRun( + new TableSinkFactoryContext( + null, + ReadonlyConfig.fromMap(options), + getClass().getClassLoader())); + } + @TestTemplate public void testClickhouse(TestContainer container) throws Exception { Container.ExecResult execResult = container.executeJob(CLICKHOUSE_JOB_CONFIG); @@ -507,6 +634,7 @@ public class ClickhouseIT extends TestSuiteBase implements TestResource { public void startUp() throws Exception { this.container = new ClickHouseContainer(CLICKHOUSE_DOCKER_IMAGE) + .withEnv("CLICKHOUSE_DEFAULT_ACCESS_MANAGEMENT", "1") .withNetwork(NETWORK) .withNetworkAliases(HOST) .withLogConsumer(
