This is an automated email from the ASF dual-hosted git repository.
github-merge-queue[bot] pushed a commit to branch dev
in repository https://gitbox.apache.org/repos/asf/seatunnel.git
The following commit(s) were added to refs/heads/dev by this push:
new 6c5b5b630f [Improve][Connector-V2][Neo4j] Migrate configuration
validation to OptionRule (#12047)
6c5b5b630f is described below
commit 6c5b5b630fa6c7b6322b522b0aa5229fd528b032
Author: 郑朝钦 <[email protected]>
AuthorDate: Thu Sep 3 13:50:18 2026 +0000
[Improve][Connector-V2][Neo4j] Migrate configuration validation to
OptionRule (#12047)
Signed-off-by: jjj-n <[email protected]>
---
docs/en/connectors/sink/Neo4j.md | 10 +-
docs/zh/connectors/sink/Neo4j.md | 10 +-
.../config/Neo4jAuthenticationConditions.java | 60 +++++++++
.../seatunnel/neo4j/config/Neo4jQueryInfo.java | 58 +-------
.../seatunnel/neo4j/config/Neo4jSinkQueryInfo.java | 31 +----
.../neo4j/config/Neo4jSourceQueryInfo.java | 4 +-
.../seatunnel/neo4j/sink/Neo4jSinkFactory.java | 26 +++-
.../seatunnel/neo4j/source/Neo4jSourceFactory.java | 12 +-
.../seatunnel/neo4j/Neo4jFactoryTest.java | 149 +++++++++++++++++++++
9 files changed, 254 insertions(+), 106 deletions(-)
diff --git a/docs/en/connectors/sink/Neo4j.md b/docs/en/connectors/sink/Neo4j.md
index dba46fc580..b3470d1214 100644
--- a/docs/en/connectors/sink/Neo4j.md
+++ b/docs/en/connectors/sink/Neo4j.md
@@ -32,7 +32,7 @@ It supports one-row-at-a-time writes and batch writes with
Cypher `UNWIND`.
| kerberos_ticket | String | No | - | Kerberos
ticket used for Neo4j authentication.
|
| database | String | Yes | - | Neo4j
database name.
|
| query | String | Yes | - | Cypher
statement used to write data. In `ONE_BY_ONE` mode, use placeholders such as
`$name`; in `BATCH` mode, use `UNWIND $batch AS row`. |
-| queryParamPosition | Object | Yes | - | Mapping
between Cypher parameter names and input row field positions. It is required by
the connector configuration. |
+| queryParamPosition | Object | ONE_BY_ONE only | - |
Mapping between Cypher parameter names and input row field positions. Required
when `write_mode = "ONE_BY_ONE"`. |
| max_batch_size | Integer | No | 500 | Maximum
number of rows written in one transaction when `write_mode = "BATCH"`. Must be
greater than 0. |
| write_mode | String | No | ONE_BY_ONE | Write mode.
Supported values are `ONE_BY_ONE` and `BATCH`.
|
| max_transaction_retry_time | Long | No | 30 | Maximum
transaction retry time, in seconds.
|
@@ -41,10 +41,9 @@ It supports one-row-at-a-time writes and batch writes with
Cypher `UNWIND`.
## Notes
-- Use exactly one authentication method: username/password, bearer token, or
Kerberos ticket.
+- Configure at least one authentication method: username/password, bearer
token, or Kerberos ticket. If several are configured, username/password takes
precedence, followed by bearer token and Kerberos ticket.
- In `ONE_BY_ONE` mode, `queryParamPosition` maps each Cypher placeholder to a
field position in the input row.
- In `BATCH` mode, the query should use `UNWIND $batch AS row`. The connector
passes the rows through the `batch` variable.
-- `queryParamPosition` is still required by the connector configuration in
`BATCH` mode, even though the batch query reads values from `row`.
- Field positions in `queryParamPosition` start from `0`, following the input
schema field order.
- In `BATCH` mode, each `row` entry uses the input field names, so the names
referenced in the Cypher statement must match the upstream schema.
@@ -104,11 +103,6 @@ sink {
max_transaction_retry_time = 3
max_connection_timeout = 10
- queryParamPosition = {
- name = 0
- age = 1
- }
-
query = "UNWIND $batch AS row CREATE (n:BatchLabel) SET n.name = row.name,
n.age = row.age"
}
}
diff --git a/docs/zh/connectors/sink/Neo4j.md b/docs/zh/connectors/sink/Neo4j.md
index b5c76ad722..eaf3fca2c3 100644
--- a/docs/zh/connectors/sink/Neo4j.md
+++ b/docs/zh/connectors/sink/Neo4j.md
@@ -32,7 +32,7 @@ Neo4j Sink 连接器通过执行 Cypher 语句把 SeaTunnel 数据写入 Neo4j
| kerberos_ticket | String | 否 | - | 用于 Neo4j 认证的
Kerberos ticket。
|
| database | String | 是 | - | Neo4j 数据库名。
|
| query | String | 是 | - | 写入数据使用的 Cypher
语句。`ONE_BY_ONE` 模式使用 `$name` 这类占位符；`BATCH` 模式使用 `UNWIND $batch AS row`。
|
-| queryParamPosition | Object | 是 | - | Cypher
参数名和输入行字段位置的映射。连接器配置校验要求必须填写。
|
+| queryParamPosition | Object | 仅 ONE_BY_ONE | - | Cypher
参数名和输入行字段位置的映射。`write_mode = "ONE_BY_ONE"` 时必须填写。
|
| max_batch_size | Integer | 否 | 500 | `write_mode =
"BATCH"` 时，单个事务最多写入的数据条数，必须大于 0。
|
| write_mode | String | 否 | ONE_BY_ONE | 写入模式。可选值为
`ONE_BY_ONE` 和 `BATCH`。
|
| max_transaction_retry_time | Long | 否 | 30 | 最大事务重试时间，单位为秒。
|
@@ -41,10 +41,9 @@ Neo4j Sink 连接器通过执行 Cypher 语句把 SeaTunnel 数据写入 Neo4j
## 注意事项
-- 认证方式只选一种：用户名密码、bearer token 或 Kerberos ticket。
+- 至少配置一种认证方式：用户名密码、bearer token 或 Kerberos
ticket。如果同时配置多种方式，优先级依次为用户名密码、bearer token、Kerberos ticket。
- `ONE_BY_ONE` 模式下，`queryParamPosition` 用来把 Cypher 占位符映射到输入行的字段位置。
- `BATCH` 模式下，查询语句应使用 `UNWIND $batch AS row`，连接器会通过 `batch` 变量传入一批数据。
-- `BATCH` 模式虽然从 `row` 中取值，但连接器配置校验仍要求填写 `queryParamPosition`。
- `queryParamPosition` 中的字段位置从 `0` 开始，顺序对应上游输入表结构。
- `BATCH` 模式下，每个 `row` 使用上游字段名取值，因此 Cypher 语句里的字段名需要和上游表结构一致。
@@ -104,11 +103,6 @@ sink {
max_transaction_retry_time = 3
max_connection_timeout = 10
- queryParamPosition = {
- name = 0
- age = 1
- }
-
query = "UNWIND $batch AS row CREATE (n:BatchLabel) SET n.name = row.name,
n.age = row.age"
}
}
diff --git
a/seatunnel-connectors-v2/connector-neo4j/src/main/java/org/apache/seatunnel/connectors/seatunnel/neo4j/config/Neo4jAuthenticationConditions.java
b/seatunnel-connectors-v2/connector-neo4j/src/main/java/org/apache/seatunnel/connectors/seatunnel/neo4j/config/Neo4jAuthenticationConditions.java
new file mode 100644
index 0000000000..6a28a4fd2d
--- /dev/null
+++
b/seatunnel-connectors-v2/connector-neo4j/src/main/java/org/apache/seatunnel/connectors/seatunnel/neo4j/config/Neo4jAuthenticationConditions.java
@@ -0,0 +1,60 @@
+/*
+ * 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.config;
+
+import org.apache.seatunnel.api.configuration.ReadonlyConfig;
+import org.apache.seatunnel.api.configuration.util.ConditionExtension;
+
+public final class Neo4jAuthenticationConditions {
+
+ public static final ConditionExtension<String> AUTHENTICATION_METHOD =
+ new AuthenticationMethodCondition();
+
+ public static final ConditionExtension<String> USERNAME_REQUIRES_PASSWORD =
+ new UsernamePasswordCondition();
+
+ private Neo4jAuthenticationConditions() {}
+
+ private static final class AuthenticationMethodCondition implements
ConditionExtension<String> {
+
+ @Override
+ public String description() {
+ return "at least one of 'username', 'bearer_token', or
'kerberos_ticket' must be configured";
+ }
+
+ @Override
+ public boolean evaluate(ReadonlyConfig config, String uri) {
+ return
config.getOptional(Neo4jBaseOptions.KEY_USERNAME).isPresent()
+ ||
config.getOptional(Neo4jBaseOptions.KEY_BEARER_TOKEN).isPresent()
+ ||
config.getOptional(Neo4jBaseOptions.KEY_KERBEROS_TICKET).isPresent();
+ }
+ }
+
+ private static final class UsernamePasswordCondition implements
ConditionExtension<String> {
+
+ @Override
+ public String description() {
+ return "requires 'password' to be configured when 'username' is
configured";
+ }
+
+ @Override
+ public boolean evaluate(ReadonlyConfig config, String username) {
+ return
config.getOptional(Neo4jBaseOptions.KEY_PASSWORD).isPresent();
+ }
+ }
+}
diff --git
a/seatunnel-connectors-v2/connector-neo4j/src/main/java/org/apache/seatunnel/connectors/seatunnel/neo4j/config/Neo4jQueryInfo.java
b/seatunnel-connectors-v2/connector-neo4j/src/main/java/org/apache/seatunnel/connectors/seatunnel/neo4j/config/Neo4jQueryInfo.java
index f3118d56ca..9e8c76cd3d 100644
---
a/seatunnel-connectors-v2/connector-neo4j/src/main/java/org/apache/seatunnel/connectors/seatunnel/neo4j/config/Neo4jQueryInfo.java
+++
b/seatunnel-connectors-v2/connector-neo4j/src/main/java/org/apache/seatunnel/connectors/seatunnel/neo4j/config/Neo4jQueryInfo.java
@@ -19,12 +19,6 @@ package
org.apache.seatunnel.connectors.seatunnel.neo4j.config;
import org.apache.seatunnel.shade.com.typesafe.config.Config;
-import org.apache.seatunnel.api.common.SeaTunnelAPIErrorCode;
-import org.apache.seatunnel.common.config.CheckConfigUtil;
-import org.apache.seatunnel.common.config.CheckResult;
-import org.apache.seatunnel.common.constants.PluginType;
-import
org.apache.seatunnel.connectors.seatunnel.neo4j.exception.Neo4jConnectorException;
-
import org.neo4j.driver.AuthTokens;
import lombok.Data;
@@ -41,7 +35,6 @@ import static
org.apache.seatunnel.connectors.seatunnel.neo4j.config.Neo4jBaseOp
import static
org.apache.seatunnel.connectors.seatunnel.neo4j.config.Neo4jBaseOptions.KEY_PASSWORD;
import static
org.apache.seatunnel.connectors.seatunnel.neo4j.config.Neo4jBaseOptions.KEY_QUERY;
import static
org.apache.seatunnel.connectors.seatunnel.neo4j.config.Neo4jBaseOptions.KEY_USERNAME;
-import static
org.apache.seatunnel.connectors.seatunnel.neo4j.config.Neo4jBaseOptions.PLUGIN_NAME;
/**
* Because Neo4jQueryInfo is one of the Neo4jSink's member variable, So
Neo4jQueryInfo need
@@ -52,50 +45,17 @@ public abstract class Neo4jQueryInfo implements
Serializable {
protected DriverBuilder driverBuilder;
protected String query;
- protected PluginType pluginType;
-
- public Neo4jQueryInfo(Config config, PluginType pluginType) {
- this.pluginType = pluginType;
- this.driverBuilder = prepareDriver(config, pluginType);
- this.query = prepareQuery(config, pluginType);
+ public Neo4jQueryInfo(Config config) {
+ this.driverBuilder = prepareDriver(config);
+ this.query = prepareQuery(config);
}
- // which is identical to the prepareDriver methods of the source and sink.
- // the only difference is the pluginType mentioned in the error messages.
- // so move code to here
- protected DriverBuilder prepareDriver(Config config, PluginType
pluginType) {
- final CheckResult uriConfigCheck =
- CheckConfigUtil.checkAllExists(config, KEY_NEO4J_URI.key(),
KEY_DATABASE.key());
- final CheckResult authConfigCheck =
- CheckConfigUtil.checkAtLeastOneExists(
- config,
- KEY_USERNAME.key(),
- KEY_BEARER_TOKEN.key(),
- KEY_KERBEROS_TICKET.key());
- final CheckResult mergedConfigCheck =
- CheckConfigUtil.mergeCheckResults(uriConfigCheck,
authConfigCheck);
- if (!mergedConfigCheck.isSuccess()) {
- throw new Neo4jConnectorException(
- SeaTunnelAPIErrorCode.CONFIG_VALIDATION_FAILED,
- String.format(
- "PluginName: %s, PluginType: %s, Message: %s",
- PLUGIN_NAME, pluginType,
mergedConfigCheck.getMsg()));
- }
-
+ protected DriverBuilder prepareDriver(Config config) {
final URI uri = URI.create(config.getString(KEY_NEO4J_URI.key()));
final DriverBuilder driverBuilder = DriverBuilder.create(uri);
if (config.hasPath(KEY_USERNAME.key())) {
- final CheckResult pwParamCheck =
- CheckConfigUtil.checkAllExists(config, KEY_PASSWORD.key());
- if (!pwParamCheck.isSuccess()) {
- throw new Neo4jConnectorException(
- SeaTunnelAPIErrorCode.CONFIG_VALIDATION_FAILED,
- String.format(
- "PluginName: %s, PluginType: %s, Message: %s",
- PLUGIN_NAME, pluginType,
pwParamCheck.getMsg()));
- }
final String username = config.getString(KEY_USERNAME.key());
final String password = config.getString(KEY_PASSWORD.key());
@@ -125,15 +85,7 @@ public abstract class Neo4jQueryInfo implements
Serializable {
return driverBuilder;
}
- private String prepareQuery(Config config, PluginType pluginType) {
- CheckResult queryConfigCheck = CheckConfigUtil.checkAllExists(config,
KEY_QUERY.key());
- if (!queryConfigCheck.isSuccess()) {
- throw new Neo4jConnectorException(
- SeaTunnelAPIErrorCode.CONFIG_VALIDATION_FAILED,
- String.format(
- "PluginName: %s, PluginType: %s, Message: %s",
- PLUGIN_NAME, pluginType,
queryConfigCheck.getMsg()));
- }
+ private String prepareQuery(Config config) {
return config.getString(KEY_QUERY.key());
}
}
diff --git
a/seatunnel-connectors-v2/connector-neo4j/src/main/java/org/apache/seatunnel/connectors/seatunnel/neo4j/config/Neo4jSinkQueryInfo.java
b/seatunnel-connectors-v2/connector-neo4j/src/main/java/org/apache/seatunnel/connectors/seatunnel/neo4j/config/Neo4jSinkQueryInfo.java
index 674b96442a..25b50cbd18 100644
---
a/seatunnel-connectors-v2/connector-neo4j/src/main/java/org/apache/seatunnel/connectors/seatunnel/neo4j/config/Neo4jSinkQueryInfo.java
+++
b/seatunnel-connectors-v2/connector-neo4j/src/main/java/org/apache/seatunnel/connectors/seatunnel/neo4j/config/Neo4jSinkQueryInfo.java
@@ -19,19 +19,13 @@ package
org.apache.seatunnel.connectors.seatunnel.neo4j.config;
import org.apache.seatunnel.shade.com.typesafe.config.Config;
-import org.apache.seatunnel.api.common.SeaTunnelAPIErrorCode;
-import org.apache.seatunnel.common.config.CheckConfigUtil;
-import org.apache.seatunnel.common.config.CheckResult;
-import org.apache.seatunnel.common.constants.PluginType;
import org.apache.seatunnel.connectors.seatunnel.neo4j.constants.SinkWriteMode;
-import
org.apache.seatunnel.connectors.seatunnel.neo4j.exception.Neo4jConnectorException;
import lombok.Getter;
import lombok.Setter;
import java.util.Map;
-import static
org.apache.seatunnel.connectors.seatunnel.neo4j.config.Neo4jBaseOptions.PLUGIN_NAME;
import static
org.apache.seatunnel.connectors.seatunnel.neo4j.config.Neo4jSinkOptions.MAX_BATCH_SIZE;
import static
org.apache.seatunnel.connectors.seatunnel.neo4j.config.Neo4jSinkOptions.QUERY_PARAM_POSITION;
import static
org.apache.seatunnel.connectors.seatunnel.neo4j.config.Neo4jSinkOptions.WRITE_MODE;
@@ -50,7 +44,7 @@ public class Neo4jSinkQueryInfo extends Neo4jQueryInfo {
}
public Neo4jSinkQueryInfo(Config config) {
- super(config, PluginType.SINK);
+ super(config);
this.writeMode = prepareWriteMode(config);
@@ -62,35 +56,14 @@ public class Neo4jSinkQueryInfo extends Neo4jQueryInfo {
}
private void prepareOneByOneConfig(Config config) {
-
- CheckResult queryConfigCheck =
- CheckConfigUtil.checkAllExists(config,
QUERY_PARAM_POSITION.key());
-
- if (!queryConfigCheck.isSuccess()) {
- throw new Neo4jConnectorException(
- SeaTunnelAPIErrorCode.CONFIG_VALIDATION_FAILED,
- String.format(
- "PluginName: %s, PluginType: %s, Message: %s",
- PLUGIN_NAME, PluginType.SINK,
queryConfigCheck.getMsg()));
- }
-
// set queryParamPosition
this.queryParamPosition =
config.getObject(QUERY_PARAM_POSITION.key()).unwrapped();
}
private void prepareBatchWriteConfig(Config config) {
-
// batch size
if (config.hasPath(MAX_BATCH_SIZE.key())) {
- int batchSize = config.getInt(MAX_BATCH_SIZE.key());
- if (batchSize <= 0) {
- throw new Neo4jConnectorException(
- SeaTunnelAPIErrorCode.CONFIG_VALIDATION_FAILED,
- String.format(
- "PluginName: %s, PluginType: %s, Message: %s",
- PLUGIN_NAME, PluginType.SINK, "maxBatchSize
must greater than 0"));
- }
- this.maxBatchSize = batchSize;
+ this.maxBatchSize = config.getInt(MAX_BATCH_SIZE.key());
} else {
this.maxBatchSize = MAX_BATCH_SIZE.defaultValue();
}
diff --git
a/seatunnel-connectors-v2/connector-neo4j/src/main/java/org/apache/seatunnel/connectors/seatunnel/neo4j/config/Neo4jSourceQueryInfo.java
b/seatunnel-connectors-v2/connector-neo4j/src/main/java/org/apache/seatunnel/connectors/seatunnel/neo4j/config/Neo4jSourceQueryInfo.java
index cd98f54f96..da3c3f89f6 100644
---
a/seatunnel-connectors-v2/connector-neo4j/src/main/java/org/apache/seatunnel/connectors/seatunnel/neo4j/config/Neo4jSourceQueryInfo.java
+++
b/seatunnel-connectors-v2/connector-neo4j/src/main/java/org/apache/seatunnel/connectors/seatunnel/neo4j/config/Neo4jSourceQueryInfo.java
@@ -19,11 +19,9 @@ package
org.apache.seatunnel.connectors.seatunnel.neo4j.config;
import org.apache.seatunnel.shade.com.typesafe.config.Config;
-import org.apache.seatunnel.common.constants.PluginType;
-
public class Neo4jSourceQueryInfo extends Neo4jQueryInfo {
public Neo4jSourceQueryInfo(Config pluginConfig) {
- super(pluginConfig, PluginType.SOURCE);
+ super(pluginConfig);
}
}
diff --git
a/seatunnel-connectors-v2/connector-neo4j/src/main/java/org/apache/seatunnel/connectors/seatunnel/neo4j/sink/Neo4jSinkFactory.java
b/seatunnel-connectors-v2/connector-neo4j/src/main/java/org/apache/seatunnel/connectors/seatunnel/neo4j/sink/Neo4jSinkFactory.java
index f2623c32f5..aef47bec3e 100644
---
a/seatunnel-connectors-v2/connector-neo4j/src/main/java/org/apache/seatunnel/connectors/seatunnel/neo4j/sink/Neo4jSinkFactory.java
+++
b/seatunnel-connectors-v2/connector-neo4j/src/main/java/org/apache/seatunnel/connectors/seatunnel/neo4j/sink/Neo4jSinkFactory.java
@@ -17,13 +17,16 @@
package org.apache.seatunnel.connectors.seatunnel.neo4j.sink;
+import org.apache.seatunnel.api.configuration.util.Conditions;
import org.apache.seatunnel.api.configuration.util.OptionRule;
import org.apache.seatunnel.api.table.connector.TableSink;
import org.apache.seatunnel.api.table.factory.Factory;
import org.apache.seatunnel.api.table.factory.TableSinkFactory;
import org.apache.seatunnel.api.table.factory.TableSinkFactoryContext;
+import
org.apache.seatunnel.connectors.seatunnel.neo4j.config.Neo4jAuthenticationConditions;
import org.apache.seatunnel.connectors.seatunnel.neo4j.config.Neo4jSinkOptions;
import
org.apache.seatunnel.connectors.seatunnel.neo4j.config.Neo4jSinkQueryInfo;
+import org.apache.seatunnel.connectors.seatunnel.neo4j.constants.SinkWriteMode;
import com.google.auto.service.AutoService;
@@ -39,16 +42,31 @@ public class Neo4jSinkFactory implements TableSinkFactory {
return OptionRule.builder()
.required(
Neo4jSinkOptions.KEY_NEO4J_URI,
- Neo4jSinkOptions.KEY_DATABASE,
- Neo4jSinkOptions.KEY_QUERY,
- Neo4jSinkOptions.QUERY_PARAM_POSITION)
+ Conditions.extension(
+ Neo4jSinkOptions.KEY_NEO4J_URI,
+
Neo4jAuthenticationConditions.AUTHENTICATION_METHOD))
+ .required(Neo4jSinkOptions.KEY_DATABASE,
Neo4jSinkOptions.KEY_QUERY)
.optional(
Neo4jSinkOptions.KEY_USERNAME,
+ Conditions.extension(
+ Neo4jSinkOptions.KEY_USERNAME,
+
Neo4jAuthenticationConditions.USERNAME_REQUIRES_PASSWORD))
+ .optional(
Neo4jSinkOptions.KEY_PASSWORD,
Neo4jSinkOptions.KEY_BEARER_TOKEN,
Neo4jSinkOptions.KEY_KERBEROS_TICKET,
Neo4jSinkOptions.KEY_MAX_CONNECTION_TIMEOUT,
- Neo4jSinkOptions.KEY_MAX_TRANSACTION_RETRY_TIME)
+ Neo4jSinkOptions.KEY_MAX_TRANSACTION_RETRY_TIME,
+ Neo4jSinkOptions.WRITE_MODE,
+ Neo4jSinkOptions.MAX_BATCH_SIZE)
+ .conditional(
+ Neo4jSinkOptions.WRITE_MODE,
+ SinkWriteMode.ONE_BY_ONE,
+ Neo4jSinkOptions.QUERY_PARAM_POSITION)
+ .conditional(
+ Neo4jSinkOptions.WRITE_MODE,
+ SinkWriteMode.BATCH,
+
Conditions.greaterThan(Neo4jSinkOptions.MAX_BATCH_SIZE, 0))
.build();
}
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 eae2e8fc79..b847b0e9a7 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
@@ -34,6 +34,7 @@ import org.apache.seatunnel.api.table.connector.TableSource;
import org.apache.seatunnel.api.table.factory.Factory;
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;
import
org.apache.seatunnel.connectors.seatunnel.neo4j.config.Neo4jSourceOptions;
import
org.apache.seatunnel.connectors.seatunnel.neo4j.config.Neo4jSourceQueryInfo;
import
org.apache.seatunnel.connectors.seatunnel.neo4j.exception.Neo4jConnectorException;
@@ -57,7 +58,12 @@ public class Neo4jSourceFactory implements
TableSourceFactory {
@Override
public OptionRule optionRule() {
return OptionRule.builder()
- .required(Neo4jSourceOptions.KEY_NEO4J_URI,
Neo4jSourceOptions.KEY_DATABASE)
+ .required(
+ Neo4jSourceOptions.KEY_NEO4J_URI,
+ Conditions.extension(
+ Neo4jSourceOptions.KEY_NEO4J_URI,
+
Neo4jAuthenticationConditions.AUTHENTICATION_METHOD))
+ .required(Neo4jSourceOptions.KEY_DATABASE)
.exclusive(Neo4jSourceOptions.KEY_QUERY,
ConnectorCommonOptions.TABLE_CONFIGS)
.optional(
Neo4jSourceOptions.KEY_QUERY,
@@ -71,6 +77,10 @@ public class Neo4jSourceFactory implements
TableSourceFactory {
ConnectorCommonOptions.TABLE_CONFIGS, new
TableConfigsValidator()))
.optional(
Neo4jSourceOptions.KEY_USERNAME,
+ Conditions.extension(
+ Neo4jSourceOptions.KEY_USERNAME,
+
Neo4jAuthenticationConditions.USERNAME_REQUIRES_PASSWORD))
+ .optional(
Neo4jSourceOptions.KEY_PASSWORD,
Neo4jSourceOptions.KEY_BEARER_TOKEN,
Neo4jSourceOptions.KEY_KERBEROS_TICKET,
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 4c6220a97f..0550d9bfca 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
@@ -22,6 +22,9 @@ 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.TableSourceFactoryContext;
+import org.apache.seatunnel.connectors.seatunnel.neo4j.config.Neo4jSinkOptions;
+import
org.apache.seatunnel.connectors.seatunnel.neo4j.config.Neo4jSourceOptions;
+import org.apache.seatunnel.connectors.seatunnel.neo4j.constants.SinkWriteMode;
import org.apache.seatunnel.connectors.seatunnel.neo4j.sink.Neo4jSinkFactory;
import org.apache.seatunnel.connectors.seatunnel.neo4j.source.Neo4jSource;
import
org.apache.seatunnel.connectors.seatunnel.neo4j.source.Neo4jSourceFactory;
@@ -143,6 +146,117 @@ class Neo4jFactoryTest {
() -> validateSource(duplicateTable)));
}
+ @Test
+ void sourceAndSinkOptionRulesRequireAuthentication() {
+ Map<String, Object> sourceConfig = validSourceConfig();
+ removeAuthentication(sourceConfig);
+
+ Map<String, Object> sinkConfig = validSinkConfig();
+ removeAuthentication(sinkConfig);
+
+ Assertions.assertAll(
+ () ->
+ Assertions.assertThrows(
+ OptionValidationException.class,
+ () -> validateSource(sourceConfig)),
+ () ->
+ Assertions.assertThrows(
+ OptionValidationException.class, () ->
validateSink(sinkConfig)));
+ }
+
+ @Test
+ void sourceAndSinkOptionRulesRequirePasswordWithUsername() {
+ Map<String, Object> sourceConfig = validSourceConfig();
+ sourceConfig.remove(Neo4jSourceOptions.KEY_PASSWORD.key());
+
+ Map<String, Object> sinkConfig = validSinkConfig();
+ sinkConfig.remove(Neo4jSinkOptions.KEY_PASSWORD.key());
+
+ Assertions.assertAll(
+ () ->
+ Assertions.assertThrows(
+ OptionValidationException.class,
+ () -> validateSource(sourceConfig)),
+ () ->
+ Assertions.assertThrows(
+ OptionValidationException.class, () ->
validateSink(sinkConfig)));
+ }
+
+ @Test
+ void sourceAndSinkOptionRulesAcceptTokenAuthentication() {
+ Map<String, Object> bearerSource = validSourceConfig();
+ useTokenAuthentication(bearerSource,
Neo4jSourceOptions.KEY_BEARER_TOKEN.key());
+ Map<String, Object> kerberosSource = validSourceConfig();
+ useTokenAuthentication(kerberosSource,
Neo4jSourceOptions.KEY_KERBEROS_TICKET.key());
+
+ Map<String, Object> bearerSink = validSinkConfig();
+ useTokenAuthentication(bearerSink,
Neo4jSinkOptions.KEY_BEARER_TOKEN.key());
+ Map<String, Object> kerberosSink = validSinkConfig();
+ useTokenAuthentication(kerberosSink,
Neo4jSinkOptions.KEY_KERBEROS_TICKET.key());
+
+ Assertions.assertAll(
+ () -> Assertions.assertDoesNotThrow(() ->
validateSource(bearerSource)),
+ () -> Assertions.assertDoesNotThrow(() ->
validateSource(kerberosSource)),
+ () -> Assertions.assertDoesNotThrow(() ->
validateSink(bearerSink)),
+ () -> Assertions.assertDoesNotThrow(() ->
validateSink(kerberosSink)));
+ }
+
+ @Test
+ void sourceAndSinkOptionRulesPreserveMultipleAuthenticationCompatibility()
{
+ Map<String, Object> sourceConfig = validSourceConfig();
+ sourceConfig.put(Neo4jSourceOptions.KEY_BEARER_TOKEN.key(),
"bearer-token");
+
+ Map<String, Object> sinkConfig = validSinkConfig();
+ sinkConfig.put(Neo4jSinkOptions.KEY_KERBEROS_TICKET.key(),
"kerberos-ticket");
+
+ Assertions.assertAll(
+ () -> Assertions.assertDoesNotThrow(() ->
validateSource(sourceConfig)),
+ () -> Assertions.assertDoesNotThrow(() ->
validateSink(sinkConfig)));
+ }
+
+ @Test
+ void sinkOptionRuleRequiresQueryParamPositionInDefaultMode() {
+ Map<String, Object> config = validSinkConfig();
+ config.remove(Neo4jSinkOptions.QUERY_PARAM_POSITION.key());
+
+ Assertions.assertThrows(OptionValidationException.class, () ->
validateSink(config));
+ }
+
+ @Test
+ void sinkOptionRuleDoesNotRequireQueryParamPositionInBatchMode() {
+ Map<String, Object> config = validSinkConfig();
+ config.put(Neo4jSinkOptions.WRITE_MODE.key(), SinkWriteMode.BATCH);
+ config.remove(Neo4jSinkOptions.QUERY_PARAM_POSITION.key());
+
+ Assertions.assertDoesNotThrow(() -> validateSink(config));
+ }
+
+ @Test
+ void sinkOptionRuleRequiresPositiveBatchSizeOnlyInBatchMode() {
+ Map<String, Object> zeroBatchSize = validSinkConfig();
+ zeroBatchSize.put(Neo4jSinkOptions.WRITE_MODE.key(),
SinkWriteMode.BATCH);
+ zeroBatchSize.put(Neo4jSinkOptions.MAX_BATCH_SIZE.key(), 0);
+ zeroBatchSize.remove(Neo4jSinkOptions.QUERY_PARAM_POSITION.key());
+
+ Map<String, Object> negativeBatchSize = validSinkConfig();
+ negativeBatchSize.put(Neo4jSinkOptions.WRITE_MODE.key(),
SinkWriteMode.BATCH);
+ negativeBatchSize.put(Neo4jSinkOptions.MAX_BATCH_SIZE.key(), -1);
+ negativeBatchSize.remove(Neo4jSinkOptions.QUERY_PARAM_POSITION.key());
+
+ Map<String, Object> oneByOne = validSinkConfig();
+ oneByOne.put(Neo4jSinkOptions.MAX_BATCH_SIZE.key(), 0);
+
+ Assertions.assertAll(
+ () ->
+ Assertions.assertThrows(
+ OptionValidationException.class, () ->
validateSink(zeroBatchSize)),
+ () ->
+ Assertions.assertThrows(
+ OptionValidationException.class,
+ () -> validateSink(negativeBatchSize)),
+ () -> Assertions.assertDoesNotThrow(() ->
validateSink(oneByOne)));
+ }
+
@Test
void sourceFactoryProducesCatalogTableForEachQuery() {
Map<String, Object> config = sourceConnectionConfig();
@@ -177,6 +291,36 @@ class Neo4jFactoryTest {
return config;
}
+ private static Map<String, Object> validSourceConfig() {
+ Map<String, Object> config = sourceConnectionConfig();
+ config.put(Neo4jSourceOptions.KEY_QUERY.key(), "MATCH (p:Person)
RETURN p.name");
+ config.put("schema", schema("people"));
+ return config;
+ }
+
+ private static Map<String, Object> validSinkConfig() {
+ Map<String, Object> config = sourceConnectionConfig();
+ config.put(Neo4jSinkOptions.KEY_QUERY.key(), "CREATE (p:Person {name:
$name, age: $age})");
+
+ Map<String, String> queryParamPosition = new HashMap<>();
+ queryParamPosition.put("name", "0");
+ queryParamPosition.put("age", "1");
+ config.put(Neo4jSinkOptions.QUERY_PARAM_POSITION.key(),
queryParamPosition);
+ return config;
+ }
+
+ private static void removeAuthentication(Map<String, Object> config) {
+ config.remove(Neo4jSourceOptions.KEY_USERNAME.key());
+ config.remove(Neo4jSourceOptions.KEY_PASSWORD.key());
+ config.remove(Neo4jSourceOptions.KEY_BEARER_TOKEN.key());
+ config.remove(Neo4jSourceOptions.KEY_KERBEROS_TICKET.key());
+ }
+
+ private static void useTokenAuthentication(Map<String, Object> config,
String tokenKey) {
+ removeAuthentication(config);
+ config.put(tokenKey, "token");
+ }
+
private static Map<String, Object> tableConfig(String table, String query)
{
Map<String, Object> tableConfig = new HashMap<>();
tableConfig.put("query", query);
@@ -198,4 +342,9 @@ class Neo4jFactoryTest {
ConfigValidator.of(ReadonlyConfig.fromMap(config))
.validate(new Neo4jSourceFactory().optionRule());
}
+
+ private static void validateSink(Map<String, Object> config) {
+ ConfigValidator.of(ReadonlyConfig.fromMap(config))
+ .validate(new Neo4jSinkFactory().optionRule());
+ }
}
