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-12433-6071eabae585977efbb7f15e066d0787f9c0bf2c in repository https://gitbox.apache.org/repos/asf/seatunnel.git
commit eb332e20fe5ccc1adee4c8b15fd6d5b189d7394b Author: Goutam Adwant <[email protected]> AuthorDate: Fri Oct 9 08:09:53 2026 +0000 [Feature][Connector-V2] Add Neo4j source connectivity dry-run (#12433) Signed-off-by: Goutam Adwant <[email protected]> --- docs/en/connectors/source/Neo4j.md | 12 ++ docs/en/engines/zeta/user-command.md | 1 + docs/zh/connectors/source/Neo4j.md | 12 ++ docs/zh/engines/zeta/user-command.md | 1 + .../neo4j/source/Neo4jSourceDryRunValidator.java | 95 +++++++++++++ .../seatunnel/neo4j/source/Neo4jSourceFactory.java | 67 ++++++++- .../seatunnel/neo4j/Neo4jFactoryTest.java | 50 +++++++ .../source/Neo4jSourceDryRunValidatorTest.java | 150 +++++++++++++++++++++ .../seatunnel/e2e/connector/neo4j/Neo4jIT.java | 148 ++++++++++++++++++++ 9 files changed, 531 insertions(+), 5 deletions(-) diff --git a/docs/en/connectors/source/Neo4j.md b/docs/en/connectors/source/Neo4j.md index 4b98715b7c..04ee0c9cc4 100644 --- a/docs/en/connectors/source/Neo4j.md +++ b/docs/en/connectors/source/Neo4j.md @@ -43,6 +43,18 @@ the returned fields to a SeaTunnel schema. | Map | MAP | | Null | NULL | +## Connectivity dry-run + +Run `bin/seatunnel.sh --config <your-job.conf> --dry-run connect` with your own job configuration. + +This source uses the existing driver connection/authentication configuration and returns the same configured single-table or `tables_configs` schemas as a normal source. + +- The temporary driver calls `verifyConnectivityAsync()`. It does not create a session, execute user Cypher, read graph records, or write data. +- Success validates driver connectivity and the authentication performed by the driver handshake. It does **not** validate the configured database, database/query permissions, Cypher syntax, result columns/types, or schema compatibility with stored data. In particular, an invalid query or a nonexistent `database` can still pass this connectivity-only check. +- Driver connection and verification waits are capped at 15 seconds, retaining a smaller positive `max_connection_timeout`; zero is bounded for preflight. Driver close is initiated even on failure and waited for at most 5 seconds. DNS resolution follows the JVM resolver. +- Retry/timeout limits apply only to the temporary driver. Normal source configuration, source execution and sink behavior are unchanged. +- Validation errors omit raw URIs, driver error text and secret-bearing exception causes. Sink connectivity remains unsupported. + ## Source Options | Name | Type | Required | Default | Description | diff --git a/docs/en/engines/zeta/user-command.md b/docs/en/engines/zeta/user-command.md index c4fcbffeff..c117864c05 100644 --- a/docs/en/engines/zeta/user-command.md +++ b/docs/en/engines/zeta/user-command.md @@ -84,6 +84,7 @@ The `--dry-run connect` option runs the static checks first, then uses connector | 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) | - | | 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 | | 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/Neo4j.md b/docs/zh/connectors/source/Neo4j.md index 1ebcef8d83..161d106cf4 100644 --- a/docs/zh/connectors/source/Neo4j.md +++ b/docs/zh/connectors/source/Neo4j.md @@ -43,6 +43,18 @@ Neo4j 源连接器通过执行 Cypher 查询从 Neo4j 读取数据，并把查 | Map | MAP | | Null | NULL | +## 连接预检查 + +使用自己的作业配置执行 `bin/seatunnel.sh --config <your-job.conf> --dry-run connect`。 + +Source 复用现有驱动的连接和认证配置，并返回与正常 Source 相同的单表或 `tables_configs` 配置 schema。 + +- 临时驱动调用 `verifyConnectivityAsync()`，不创建 session，不执行用户 Cypher，不读取图数据，也不写入数据。 +- 成功表示驱动连接及握手时执行的认证成功，**不代表**配置的数据库存在、拥有数据库或查询权限、Cypher 语法正确、结果字段/类型正确，或 schema 与存储数据匹配。无效查询或不存在的 `database` 仍可能通过此连接检查。 +- 驱动连接和验证等待各自最多 15 秒，保留更小的正数 `max_connection_timeout`；零值在预检查中也受到限制。失败后仍会发起驱动关闭，最多等待 5 秒。DNS 解析由 JVM 解析器控制。 +- 重试和超时限制仅作用于临时驱动。正常 Source 配置、执行及 Sink 行为保持不变。 +- 校验错误不包含原始 URI、驱动原始错误或含凭据的异常原因。Sink 连通性仍不支持。 + ## 源选项 | 名称 | 类型 | 是否必填 | 默认值 | 描述 | diff --git a/docs/zh/engines/zeta/user-command.md b/docs/zh/engines/zeta/user-command.md index 6dd61b4bbc..e2b056500b 100644 --- a/docs/zh/engines/zeta/user-command.md +++ b/docs/zh/engines/zeta/user-command.md @@ -100,6 +100,7 @@ bin/seatunnel.sh --config $SEATUNNEL_HOME/config/v2.batch.config.template --dry- | Kafka | 支持（[主题元数据 + 运行时输出 schema](../../connectors/source/Kafka.md#连通性-dry-run)，不含消费或消费组权限） | 不支持 | | FakeSource | 支持（仅 schema 推断，无外部系统） | - | | MongoDB | 不支持 | 支持（[连通性 + 配置的认证](../../connectors/sink/MongoDB.md#连通性-dry-run)，不检查集合、schema、写入权限或事务） | +| Neo4j | 支持（[驱动连通性 + 配置 schema，不含数据库/查询校验](../../connectors/source/Neo4j.md#连接预检查)） | 不支持 | | 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-neo4j/src/main/java/org/apache/seatunnel/connectors/seatunnel/neo4j/source/Neo4jSourceDryRunValidator.java b/seatunnel-connectors-v2/connector-neo4j/src/main/java/org/apache/seatunnel/connectors/seatunnel/neo4j/source/Neo4jSourceDryRunValidator.java new file mode 100644 index 0000000000..296f1949ae --- /dev/null +++ b/seatunnel-connectors-v2/connector-neo4j/src/main/java/org/apache/seatunnel/connectors/seatunnel/neo4j/source/Neo4jSourceDryRunValidator.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.neo4j.source; + +import org.apache.seatunnel.connectors.seatunnel.neo4j.config.DriverBuilder; + +import org.neo4j.driver.Driver; + +import java.io.IOException; +import java.util.concurrent.TimeUnit; + +/** Verifies the driver's connection only; never creates a session or executes configured Cypher. */ +final class Neo4jSourceDryRunValidator { + private static final long MAX_TIMEOUT_SECONDS = 15; + private static final long CLOSE_TIMEOUT_SECONDS = 5; + + private Neo4jSourceDryRunValidator() {} + + static void validate(DriverBuilder builder) throws Exception { + if (Thread.currentThread().isInterrupted()) { + throw new InterruptedException("Neo4j connect dry-run interrupted"); + } + Driver driver = null; + Exception failure = null; + boolean interrupted = false; + try { + Long configuredTimeout = builder.getMaxConnectionTimeoutSeconds(); + if (configuredTimeout != null && configuredTimeout < 0) { + throw new IllegalArgumentException("Invalid connection timeout"); + } + long timeout = + configuredTimeout == null || configuredTimeout == 0 + ? MAX_TIMEOUT_SECONDS + : Math.min(configuredTimeout, MAX_TIMEOUT_SECONDS); + // This builder belongs only to the preflight; normal source settings are unchanged. + builder.setMaxConnectionTimeoutSeconds(timeout); + builder.setMaxTransactionRetryTimeSeconds(0L); + driver = builder.build(); + // Also bound routing/handshake waits, not just socket connection establishment. + driver.verifyConnectivityAsync().toCompletableFuture().get(timeout, TimeUnit.SECONDS); + } catch (InterruptedException e) { + interrupted = true; + failure = new InterruptedException("Neo4j connect dry-run interrupted"); + } catch (Exception e) { + failure = + new IOException( + "Neo4j connect dry-run connectivity check failed; check URI, authentication and TLS settings"); + } finally { + interrupted |= Thread.interrupted(); + if (driver != null) { + try { + // Close even after a failed or timed-out handshake; do not wait indefinitely. + driver.closeAsync() + .toCompletableFuture() + .get(CLOSE_TIMEOUT_SECONDS, TimeUnit.SECONDS); + } catch (InterruptedException e) { + interrupted = true; + if (failure == null) { + failure = + new InterruptedException( + "Neo4j connect dry-run cleanup interrupted"); + } + } catch (Exception e) { + if (failure == null) { + failure = new IOException("Neo4j connect dry-run driver cleanup failed"); + } + } + } + if (interrupted) { + Thread.currentThread().interrupt(); + if (failure == null) { + failure = new InterruptedException("Neo4j connect dry-run interrupted"); + } + } + } + if (failure != null) { + throw failure; + } + } +} diff --git a/seatunnel-connectors-v2/connector-neo4j/src/main/java/org/apache/seatunnel/connectors/seatunnel/neo4j/source/Neo4jSourceFactory.java b/seatunnel-connectors-v2/connector-neo4j/src/main/java/org/apache/seatunnel/connectors/seatunnel/neo4j/source/Neo4jSourceFactory.java index b847b0e9a7..ab89e9aa03 100644 --- a/seatunnel-connectors-v2/connector-neo4j/src/main/java/org/apache/seatunnel/connectors/seatunnel/neo4j/source/Neo4jSourceFactory.java +++ b/seatunnel-connectors-v2/connector-neo4j/src/main/java/org/apache/seatunnel/connectors/seatunnel/neo4j/source/Neo4jSourceFactory.java @@ -23,6 +23,7 @@ import org.apache.seatunnel.api.common.SeaTunnelAPIErrorCode; import org.apache.seatunnel.api.configuration.ReadonlyConfig; import org.apache.seatunnel.api.configuration.util.ConditionExtension; 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.configuration.util.OptionValidationException; import org.apache.seatunnel.api.options.ConnectorCommonOptions; @@ -32,6 +33,7 @@ import org.apache.seatunnel.api.table.catalog.CatalogTable; import org.apache.seatunnel.api.table.catalog.CatalogTableUtil; 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.neo4j.config.Neo4jAuthenticationConditions; @@ -41,15 +43,45 @@ import org.apache.seatunnel.connectors.seatunnel.neo4j.exception.Neo4jConnectorE import com.google.auto.service.AutoService; +import java.io.IOException; import java.io.Serializable; import java.util.ArrayList; +import java.util.Collections; import java.util.HashSet; import java.util.List; import java.util.Map; import java.util.Set; @AutoService(Factory.class) -public class Neo4jSourceFactory implements TableSourceFactory { +public class Neo4jSourceFactory implements TableSourceFactory, SupportSourceDryRunValidation { + + /** Reuses runtime catalog construction without opening a reader or a network connection. */ + @Override + public List<CatalogTable> inferSchemaForDryRun(TableSourceFactoryContext context) + throws IOException { + return parseDryRunSourceConfig(context).catalogTables; + } + + private SourceConfiguration parseDryRunSourceConfig(TableSourceFactoryContext context) + throws IOException { + try { + ConfigValidator.of(context.getOptions()).validate(optionRule()); + return parseSourceConfig(context.getOptions()); + } catch (RuntimeException e) { + throw new IOException( + "Neo4j 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. + Neo4jSourceDryRunValidator.validate( + parseDryRunSourceConfig(context).connectionInfo.getDriverBuilder()); + } + @Override public String factoryIdentifier() { return Neo4jSourceOptions.PLUGIN_NAME; @@ -102,10 +134,20 @@ public class Neo4jSourceFactory implements TableSourceFactory { } private Neo4jSource createNeo4jSource(ReadonlyConfig config) { + SourceConfiguration parsed = parseSourceConfig(config); + return parsed.tableConfigs == null + ? new Neo4jSource(parsed.catalogTables.get(0), parsed.connectionInfo) + : new Neo4jSource(parsed.catalogTables, parsed.connectionInfo, parsed.tableConfigs); + } + + // Shared pure configuration parsing keeps preflight schemas identical to the runtime without + // creating a source, reader or driver. + private SourceConfiguration parseSourceConfig(ReadonlyConfig config) { if (!config.getOptional(ConnectorCommonOptions.TABLE_CONFIGS).isPresent()) { - return new Neo4jSource( - CatalogTableUtil.buildWithConfig(config), - new Neo4jSourceQueryInfo(config.toConfig())); + return new SourceConfiguration( + Collections.singletonList(CatalogTableUtil.buildWithConfig(config)), + new Neo4jSourceQueryInfo(config.toConfig()), + null); } List<Map<String, Object>> entries = config.get(ConnectorCommonOptions.TABLE_CONFIGS); @@ -155,7 +197,22 @@ public class Neo4jSourceFactory implements TableSourceFactory { Neo4jSourceOptions.KEY_QUERY.key(), ConfigValueFactory.fromAnyRef( tableConfigs.get(0).getQuery()))); - return new Neo4jSource(catalogTables, connectionInfo, tableConfigs); + return new SourceConfiguration(catalogTables, connectionInfo, tableConfigs); + } + + private static final class SourceConfiguration { + private final List<CatalogTable> catalogTables; + private final Neo4jSourceQueryInfo connectionInfo; + private final List<Neo4jSourceTableConfig> tableConfigs; + + private SourceConfiguration( + List<CatalogTable> catalogTables, + Neo4jSourceQueryInfo connectionInfo, + List<Neo4jSourceTableConfig> tableConfigs) { + this.catalogTables = catalogTables; + this.connectionInfo = connectionInfo; + this.tableConfigs = tableConfigs; + } } private static Neo4jConnectorException configError(String message) { diff --git a/seatunnel-connectors-v2/connector-neo4j/src/test/java/org/apache/seatunnel/connectors/seatunnel/neo4j/Neo4jFactoryTest.java b/seatunnel-connectors-v2/connector-neo4j/src/test/java/org/apache/seatunnel/connectors/seatunnel/neo4j/Neo4jFactoryTest.java index 0550d9bfca..682ae81d9a 100644 --- a/seatunnel-connectors-v2/connector-neo4j/src/test/java/org/apache/seatunnel/connectors/seatunnel/neo4j/Neo4jFactoryTest.java +++ b/seatunnel-connectors-v2/connector-neo4j/src/test/java/org/apache/seatunnel/connectors/seatunnel/neo4j/Neo4jFactoryTest.java @@ -21,6 +21,7 @@ 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.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.neo4j.config.Neo4jSinkOptions; import org.apache.seatunnel.connectors.seatunnel.neo4j.config.Neo4jSourceOptions; @@ -40,6 +41,11 @@ import java.util.Map; class Neo4jFactoryTest { + @Test + void supportsConnectivityDryRun() { + Assertions.assertTrue(new Neo4jSourceFactory() instanceof SupportSourceDryRunValidation); + } + @Test void optionRule() { Assertions.assertNotNull((new Neo4jSourceFactory()).optionRule()); @@ -282,6 +288,50 @@ class Neo4jFactoryTest { Assertions.assertEquals("companies", tables.get(1).getTableId().toTablePath().toString()); } + @Test + void dryRunSchemaMatchesRuntimeForSingleAndMultipleTables() throws Exception { + Map<String, Object> multi = sourceConnectionConfig(); + multi.put( + "tables_configs", + Arrays.asList( + tableConfig("people", "CREATE (:MustNotExecute)"), + tableConfig("companies", "THIS IS NOT VALID CYPHER"))); + Neo4jSourceFactory factory = new Neo4jSourceFactory(); + for (Map<String, Object> options : Arrays.asList(validSourceConfig(), multi)) { + TableSourceFactoryContext context = + new TableSourceFactoryContext( + ReadonlyConfig.fromMap(options), getClass().getClassLoader()); + List<CatalogTable> expected = + ((Neo4jSource) (Object) factory.createSource(context).createSource()) + .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 dryRunInvalidUriDoesNotEchoCredentials() { + Map<String, Object> options = validSourceConfig(); + options.put("uri", "neo4j://private-secret@[invalid"); + Exception failure = + Assertions.assertThrows( + java.io.IOException.class, + () -> + new Neo4jSourceFactory() + .inferSchemaForDryRun( + new TableSourceFactoryContext( + ReadonlyConfig.fromMap(options), + getClass().getClassLoader()))); + Assertions.assertFalse(failure.getMessage().contains("private-secret")); + Assertions.assertNull(failure.getCause()); + } + private static Map<String, Object> sourceConnectionConfig() { Map<String, Object> config = new HashMap<>(); config.put("uri", "neo4j://localhost:7687"); diff --git a/seatunnel-connectors-v2/connector-neo4j/src/test/java/org/apache/seatunnel/connectors/seatunnel/neo4j/source/Neo4jSourceDryRunValidatorTest.java b/seatunnel-connectors-v2/connector-neo4j/src/test/java/org/apache/seatunnel/connectors/seatunnel/neo4j/source/Neo4jSourceDryRunValidatorTest.java new file mode 100644 index 0000000000..775421a872 --- /dev/null +++ b/seatunnel-connectors-v2/connector-neo4j/src/test/java/org/apache/seatunnel/connectors/seatunnel/neo4j/source/Neo4jSourceDryRunValidatorTest.java @@ -0,0 +1,150 @@ +/* + * 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.neo4j.source; + +import org.apache.seatunnel.connectors.seatunnel.neo4j.config.DriverBuilder; + +import org.junit.jupiter.api.Test; +import org.neo4j.driver.Driver; + +import java.io.IOException; +import java.util.concurrent.CompletableFuture; + +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.mock; +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 Neo4jSourceDryRunValidatorTest { + @Test + void verifiesConnectionAndClosesWithoutCreatingSession() throws Exception { + DriverBuilder builder = mock(DriverBuilder.class); + Driver driver = mock(Driver.class); + when(builder.build()).thenReturn(driver); + when(driver.verifyConnectivityAsync()).thenReturn(CompletableFuture.completedFuture(null)); + when(driver.closeAsync()).thenReturn(CompletableFuture.completedFuture(null)); + + Neo4jSourceDryRunValidator.validate(builder); + + verify(driver).verifyConnectivityAsync(); + verify(driver).closeAsync(); + verifyNoMoreInteractions(driver); + verify(builder).setMaxConnectionTimeoutSeconds(15L); + verify(builder).setMaxTransactionRetryTimeSeconds(0L); + } + + @Test + void closesAfterAuthenticationFailureAndDoesNotExposeDriverText() { + DriverBuilder builder = mock(DriverBuilder.class); + Driver driver = mock(Driver.class); + CompletableFuture<Void> failed = new CompletableFuture<>(); + failed.completeExceptionally(new IllegalArgumentException("neo4j://user:secret@host")); + when(builder.build()).thenReturn(driver); + when(driver.verifyConnectivityAsync()).thenReturn(failed); + when(driver.closeAsync()).thenReturn(CompletableFuture.completedFuture(null)); + + IOException error = + assertThrows(IOException.class, () -> Neo4jSourceDryRunValidator.validate(builder)); + + assertFalse(error.getMessage().contains("secret")); + assertNull(error.getCause()); + assertEquals(0, error.getSuppressed().length); + verify(driver).closeAsync(); + } + + @Test + void honorsSmallerTimeoutAndClosesStalledHandshake() { + DriverBuilder builder = mock(DriverBuilder.class); + Driver driver = mock(Driver.class); + when(builder.getMaxConnectionTimeoutSeconds()).thenReturn(1L); + when(builder.build()).thenReturn(driver); + when(driver.verifyConnectivityAsync()).thenReturn(new CompletableFuture<>()); + when(driver.closeAsync()).thenReturn(CompletableFuture.completedFuture(null)); + + assertThrows(IOException.class, () -> Neo4jSourceDryRunValidator.validate(builder)); + verify(builder).setMaxConnectionTimeoutSeconds(1L); + verify(driver).closeAsync(); + } + + @Test + void rejectsNegativeTimeoutBeforeDriverCreation() { + DriverBuilder builder = mock(DriverBuilder.class); + when(builder.getMaxConnectionTimeoutSeconds()).thenReturn(-1L); + assertThrows(IOException.class, () -> Neo4jSourceDryRunValidator.validate(builder)); + verify(builder).getMaxConnectionTimeoutSeconds(); + verifyNoMoreInteractions(builder); + } + + @Test + void rejectsPreInterruptedThreadWithoutCreatingDriver() { + DriverBuilder builder = mock(DriverBuilder.class); + Thread.currentThread().interrupt(); + try { + assertThrows( + InterruptedException.class, () -> Neo4jSourceDryRunValidator.validate(builder)); + assertTrue(Thread.currentThread().isInterrupted()); + verifyNoInteractions(builder); + } finally { + Thread.interrupted(); + } + } + + @Test + void preservesInterruptionDuringVerificationAndStillCloses() { + DriverBuilder builder = mock(DriverBuilder.class); + Driver driver = mock(Driver.class); + when(builder.build()).thenReturn(driver); + when(driver.verifyConnectivityAsync()) + .thenAnswer( + ignored -> { + Thread.currentThread().interrupt(); + return new CompletableFuture<>(); + }); + when(driver.closeAsync()).thenReturn(CompletableFuture.completedFuture(null)); + try { + assertThrows( + InterruptedException.class, () -> Neo4jSourceDryRunValidator.validate(builder)); + assertTrue(Thread.currentThread().isInterrupted()); + verify(driver).closeAsync(); + } finally { + Thread.interrupted(); + } + } + + @Test + void cleanupFailureCannotTurnIntoSuccessOrExposeSecrets() { + DriverBuilder builder = mock(DriverBuilder.class); + Driver driver = mock(Driver.class); + when(builder.build()).thenReturn(driver); + when(driver.verifyConnectivityAsync()).thenReturn(CompletableFuture.completedFuture(null)); + CompletableFuture<Void> failedClose = new CompletableFuture<>(); + failedClose.completeExceptionally(new IllegalArgumentException("private-token")); + when(driver.closeAsync()).thenReturn(failedClose); + IOException failure = + assertThrows(IOException.class, () -> Neo4jSourceDryRunValidator.validate(builder)); + assertTrue(failure.getMessage().contains("cleanup")); + assertFalse(failure.getMessage().contains("private-token")); + assertNull(failure.getCause()); + } +} diff --git a/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-neo4j-e2e/src/test/java/org/apache/seatunnel/e2e/connector/neo4j/Neo4jIT.java b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-neo4j-e2e/src/test/java/org/apache/seatunnel/e2e/connector/neo4j/Neo4jIT.java index ad21d44672..40f7b3067a 100644 --- a/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-neo4j-e2e/src/test/java/org/apache/seatunnel/e2e/connector/neo4j/Neo4jIT.java +++ b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-neo4j-e2e/src/test/java/org/apache/seatunnel/e2e/connector/neo4j/Neo4jIT.java @@ -17,6 +17,17 @@ package org.apache.seatunnel.e2e.connector.neo4j; +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.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.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; @@ -24,7 +35,9 @@ import org.apache.seatunnel.e2e.common.container.TestContainer; 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.neo4j.driver.AuthTokens; import org.neo4j.driver.Driver; import org.neo4j.driver.GraphDatabase; @@ -46,9 +59,16 @@ import lombok.extern.slf4j.Slf4j; import java.io.IOException; import java.net.URI; +import java.nio.charset.StandardCharsets; +import java.nio.file.Files; +import java.nio.file.Path; import java.time.LocalDate; import java.time.LocalDateTime; import java.util.ArrayList; +import java.util.Arrays; +import java.util.Collections; +import java.util.HashMap; +import java.util.Map; import java.util.concurrent.TimeUnit; import java.util.stream.Stream; @@ -197,6 +217,134 @@ public class Neo4jIT extends TestSuiteBase implements TestResource { Assertions.assertEquals(0, execResult.getExitCode()); } + @Test + public void testDryRunDoesNotExecuteConfiguredCypher() throws Exception { + neo4jSession.run("MATCH (n:DryRunSentinel) DELETE n").consume(); + neo4jSession.run("CREATE (:DryRunSentinel {name:'unchanged'})").consume(); + Map<String, Object> options = dryRunOptions(); + options.put("query", "MATCH (n:DryRunSentinel) DELETE n"); + validateDryRun(options); + Assertions.assertEquals( + "unchanged", + neo4jSession + .run("MATCH (n:DryRunSentinel) RETURN n.name AS name") + .single() + .get("name") + .asString()); + options.put("query", "THIS IS DELIBERATELY INVALID CYPHER"); + validateDryRun(options); + } + + @Test + public void testDryRunMultiTableSchemaWithoutQueryExecution() throws Exception { + Map<String, Object> options = dryRunOptions(); + options.remove("query"); + options.remove("schema"); + Map<String, Object> first = new HashMap<>(); + first.put("query", "CREATE (:DryRunMustNotExist)"); + Map<String, Object> schema = new HashMap<>(); + schema.put("table", "first"); + schema.put("fields", Collections.singletonMap("name", "string")); + first.put("schema", schema); + Map<String, Object> second = new HashMap<>(); + second.put("query", "INVALID CYPHER"); + Map<String, Object> secondSchema = new HashMap<>(schema); + secondSchema.put("table", "second"); + second.put("schema", secondSchema); + options.put("tables_configs", Arrays.asList(first, second)); + validateDryRun(options); + Assertions.assertEquals( + 0, + neo4jSession + .run("MATCH (n:DryRunMustNotExist) RETURN count(n) AS count") + .single() + .get("count") + .asInt()); + } + + @Test + public void testDryRunRejectsWrongPasswordWithoutEchoingIt() { + Map<String, Object> options = dryRunOptions(); + options.put("password", "dry-run-private-password"); + Exception error = Assertions.assertThrows(IOException.class, () -> validateDryRun(options)); + Assertions.assertFalse(error.toString().contains("dry-run-private-password")); + Assertions.assertNull(error.getCause()); + } + + @Test + public void testDryRunDoesNotClaimDatabaseValidation() throws Exception { + Map<String, Object> options = dryRunOptions(); + options.put("database", "missing-preflight-database"); + // Driver connectivity does not select a database or prove query permissions. + validateDryRun(options); + } + + private Map<String, Object> dryRunOptions() { + Map<String, Object> options = new HashMap<>(); + options.put( + "uri", + String.format( + "bolt://%s:%s", container.getHost(), container.getMappedPort(BOLT_PORT))); + options.put("database", "neo4j"); + options.put("username", CONTAINER_NEO4J_USERNAME); + options.put("password", CONTAINER_NEO4J_PASSWORD); + options.put("query", "RETURN 'unused' AS name"); + options.put( + "schema", + Collections.singletonMap("fields", Collections.singletonMap("name", "string"))); + return options; + } + + @Test + public void testDryRunCommand(@TempDir Path directory) throws Exception { + Map<String, Object> options = dryRunOptions(); + options.put("query", "CREATE (:DryRunCommandMustNotExist)"); + checkDryRunCommand(options, directory); + Assertions.assertEquals( + 0, + neo4jSession + .run("MATCH (n:DryRunCommandMustNotExist) RETURN count(n) AS count") + .single() + .get("count") + .asInt()); + } + + private void checkDryRunCommand(Map<String, Object> options, Path directory) throws Exception { + options.put("plugin_name", "Neo4j"); + 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, "Neo4j"); + Assertions.assertTrue(factory instanceof SupportSourceDryRunValidation); + SupportSourceDryRunValidation validator = (SupportSourceDryRunValidation) factory; + TableSourceFactoryContext context = + new TableSourceFactoryContext(ReadonlyConfig.fromMap(options), loader); + validator.validateConnectionForDryRun(context, validator.inferSchemaForDryRun(context)); + } + @AfterAll @Override public void tearDown() {
