This is an automated email from the ASF dual-hosted git repository.
davidzollo 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 a7cde22db3 [Improve][Connector-V2] Migrate Redis validation to
declarative OptionRule (#11225)
a7cde22db3 is described below
commit a7cde22db3eab22a223b90bcf07a2305ae962949
Author: Shuai Liu <[email protected]>
AuthorDate: Fri Jul 3 18:07:36 2026 +0800
[Improve][Connector-V2] Migrate Redis validation to declarative OptionRule
(#11225)
Co-authored-by: Claude Opus 4.8 (1M context) <[email protected]>
---
.../seatunnel/redis/config/RedisBaseOptions.java | 4 +
.../redis/config/RedisNodesValidator.java | 72 ++++++
.../seatunnel/redis/config/RedisParameters.java | 12 -
.../config/RedisSingleTableDataTypeValidator.java} | 21 +-
.../seatunnel/redis/config/RedisTableConfig.java | 20 --
.../redis/config/RedisTableConfigsValidator.java | 51 ++++
.../seatunnel/redis/sink/RedisSinkFactory.java | 15 ++
.../seatunnel/redis/source/RedisSourceFactory.java | 28 +++
.../seatunnel/redis/RedisFactoryTest.java | 257 ++++++++++++++++++++-
.../redis/config/RedisTableConfigTest.java | 31 ---
.../e2e/connector/redis/RedisValidationIT.java | 58 +++++
.../redis-validation-cluster-malformed-node.conf | 48 ++++
.../redis-validation-single-port-out-of-range.conf | 47 ++++
13 files changed, 589 insertions(+), 75 deletions(-)
diff --git
a/seatunnel-connectors-v2/connector-redis/src/main/java/org/apache/seatunnel/connectors/seatunnel/redis/config/RedisBaseOptions.java
b/seatunnel-connectors-v2/connector-redis/src/main/java/org/apache/seatunnel/connectors/seatunnel/redis/config/RedisBaseOptions.java
index b834ebb487..690f69fd92 100644
---
a/seatunnel-connectors-v2/connector-redis/src/main/java/org/apache/seatunnel/connectors/seatunnel/redis/config/RedisBaseOptions.java
+++
b/seatunnel-connectors-v2/connector-redis/src/main/java/org/apache/seatunnel/connectors/seatunnel/redis/config/RedisBaseOptions.java
@@ -41,6 +41,10 @@ public class RedisBaseOptions extends ConnectorCommonOptions
{
public static final Option<Integer> PORT =
Options.key("port").intType().defaultValue(6379).withDescription("redis port");
+ public static final int MIN_PORT = 1;
+
+ public static final int MAX_PORT = 65535;
+
public static final Option<String> AUTH =
Options.key("auth")
.stringType()
diff --git
a/seatunnel-connectors-v2/connector-redis/src/main/java/org/apache/seatunnel/connectors/seatunnel/redis/config/RedisNodesValidator.java
b/seatunnel-connectors-v2/connector-redis/src/main/java/org/apache/seatunnel/connectors/seatunnel/redis/config/RedisNodesValidator.java
new file mode 100644
index 0000000000..b76fa1448d
--- /dev/null
+++
b/seatunnel-connectors-v2/connector-redis/src/main/java/org/apache/seatunnel/connectors/seatunnel/redis/config/RedisNodesValidator.java
@@ -0,0 +1,72 @@
+/*
+ * 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.redis.config;
+
+import org.apache.seatunnel.api.configuration.ReadonlyConfig;
+import org.apache.seatunnel.api.configuration.util.ConditionExtension;
+import org.apache.seatunnel.api.configuration.util.OptionValidationException;
+
+import java.util.List;
+import java.util.regex.Pattern;
+
+public class RedisNodesValidator implements ConditionExtension<List<String>> {
+
+ private static final Pattern NODE_PATTERN =
Pattern.compile("^[^:]+:\\d+$");
+
+ @Override
+ public String description() {
+ return "each node must be in 'host:port' format with port in ["
+ + RedisBaseOptions.MIN_PORT
+ + ", "
+ + RedisBaseOptions.MAX_PORT
+ + "]";
+ }
+
+ @Override
+ public boolean evaluate(ReadonlyConfig config, List<String> value)
+ throws OptionValidationException {
+ for (int i = 0; i < value.size(); i++) {
+ String node = value.get(i);
+ if (node == null || !NODE_PATTERN.matcher(node).matches()) {
+ throw new OptionValidationException(
+ "nodes[%d]: must be in 'host:port' format, got: %s",
i, node);
+ }
+ int colon = node.indexOf(':');
+ String host = node.substring(0, colon);
+ if (host.trim().isEmpty()) {
+ throw new OptionValidationException(
+ "nodes[%d]: host must not be blank in 'host:port',
got: %s", i, node);
+ }
+ String portStr = node.substring(colon + 1);
+ int port;
+ try {
+ port = Integer.parseInt(portStr);
+ } catch (NumberFormatException e) {
+ throw new OptionValidationException(
+ "nodes[%d]: port '%s' is out of range [%d, %d]",
+ i, portStr, RedisBaseOptions.MIN_PORT,
RedisBaseOptions.MAX_PORT);
+ }
+ if (port < RedisBaseOptions.MIN_PORT || port >
RedisBaseOptions.MAX_PORT) {
+ throw new OptionValidationException(
+ "nodes[%d]: port %d is out of range [%d, %d]",
+ i, port, RedisBaseOptions.MIN_PORT,
RedisBaseOptions.MAX_PORT);
+ }
+ }
+ return true;
+ }
+}
diff --git
a/seatunnel-connectors-v2/connector-redis/src/main/java/org/apache/seatunnel/connectors/seatunnel/redis/config/RedisParameters.java
b/seatunnel-connectors-v2/connector-redis/src/main/java/org/apache/seatunnel/connectors/seatunnel/redis/config/RedisParameters.java
index a53a3d2afa..cac06d2b47 100644
---
a/seatunnel-connectors-v2/connector-redis/src/main/java/org/apache/seatunnel/connectors/seatunnel/redis/config/RedisParameters.java
+++
b/seatunnel-connectors-v2/connector-redis/src/main/java/org/apache/seatunnel/connectors/seatunnel/redis/config/RedisParameters.java
@@ -39,8 +39,6 @@ import java.util.HashSet;
import java.util.List;
import static
org.apache.seatunnel.connectors.seatunnel.redis.exception.RedisErrorCode.GET_REDIS_VERSION_INFO_FAILED;
-import static
org.apache.seatunnel.connectors.seatunnel.redis.exception.RedisErrorCode.INVALID_CONFIG;
-import static
org.apache.seatunnel.connectors.seatunnel.redis.exception.RedisErrorCode.REDIS_NODE_EMPTY_ERROR;
@Data
@Slf4j
@@ -196,18 +194,8 @@ public class RedisParameters implements Serializable {
return jedis;
case CLUSTER:
HashSet<HostAndPort> nodes = new HashSet<>();
- if (redisNodes.isEmpty()) {
- throw new RedisConnectorException(
- REDIS_NODE_EMPTY_ERROR, "Redis nodes parameter
must not be empty");
- }
for (String redisNode : redisNodes) {
String[] splits = redisNode.split(":");
- if (splits.length != 2) {
- throw new RedisConnectorException(
- INVALID_CONFIG,
- "Invalid redis node information,"
- + "redis node information must like as
the following: [host:port]");
- }
HostAndPort hostAndPort =
new HostAndPort(splits[0],
Integer.parseInt(splits[1]));
nodes.add(hostAndPort);
diff --git
a/seatunnel-connectors-v2/connector-redis/src/test/java/org/apache/seatunnel/connectors/seatunnel/redis/RedisFactoryTest.java
b/seatunnel-connectors-v2/connector-redis/src/main/java/org/apache/seatunnel/connectors/seatunnel/redis/config/RedisSingleTableDataTypeValidator.java
similarity index 59%
copy from
seatunnel-connectors-v2/connector-redis/src/test/java/org/apache/seatunnel/connectors/seatunnel/redis/RedisFactoryTest.java
copy to
seatunnel-connectors-v2/connector-redis/src/main/java/org/apache/seatunnel/connectors/seatunnel/redis/config/RedisSingleTableDataTypeValidator.java
index ef2b3f0367..42cfa87f62 100644
---
a/seatunnel-connectors-v2/connector-redis/src/test/java/org/apache/seatunnel/connectors/seatunnel/redis/RedisFactoryTest.java
+++
b/seatunnel-connectors-v2/connector-redis/src/main/java/org/apache/seatunnel/connectors/seatunnel/redis/config/RedisSingleTableDataTypeValidator.java
@@ -15,19 +15,20 @@
* limitations under the License.
*/
-package org.apache.seatunnel.connectors.seatunnel.redis;
+package org.apache.seatunnel.connectors.seatunnel.redis.config;
-import org.apache.seatunnel.connectors.seatunnel.redis.sink.RedisSinkFactory;
-import
org.apache.seatunnel.connectors.seatunnel.redis.source.RedisSourceFactory;
+import org.apache.seatunnel.api.configuration.ReadonlyConfig;
+import org.apache.seatunnel.api.configuration.util.ConditionExtension;
-import org.junit.jupiter.api.Assertions;
-import org.junit.jupiter.api.Test;
+public class RedisSingleTableDataTypeValidator implements
ConditionExtension<String> {
-class RedisFactoryTest {
+ @Override
+ public String description() {
+ return "'data_type' must be configured when 'keys' is set";
+ }
- @Test
- void optionRule() {
- Assertions.assertNotNull((new RedisSourceFactory()).optionRule());
- Assertions.assertNotNull((new RedisSinkFactory()).optionRule());
+ @Override
+ public boolean evaluate(ReadonlyConfig config, String keys) {
+ return config.getOptional(RedisBaseOptions.DATA_TYPE).isPresent();
}
}
diff --git
a/seatunnel-connectors-v2/connector-redis/src/main/java/org/apache/seatunnel/connectors/seatunnel/redis/config/RedisTableConfig.java
b/seatunnel-connectors-v2/connector-redis/src/main/java/org/apache/seatunnel/connectors/seatunnel/redis/config/RedisTableConfig.java
index 2f90ff5aad..ab61339603 100644
---
a/seatunnel-connectors-v2/connector-redis/src/main/java/org/apache/seatunnel/connectors/seatunnel/redis/config/RedisTableConfig.java
+++
b/seatunnel-connectors-v2/connector-redis/src/main/java/org/apache/seatunnel/connectors/seatunnel/redis/config/RedisTableConfig.java
@@ -135,9 +135,6 @@ public class RedisTableConfig implements Serializable {
* @return Fully initialized RedisTableConfig with runtime objects
*/
private static RedisTableConfig buildTableConfig(ReadonlyConfig config) {
- // Validate required fields are present
- validateRequiredFields(config);
-
// Build catalog table and deserialization schema
TableConfigResult result = buildCatalogTableAndSchema(config,
config.get(KEY_PATTERN));
@@ -158,23 +155,6 @@ public class RedisTableConfig implements Serializable {
.build();
}
- /**
- * Validate required fields in a (table-level) configuration.
- *
- * @param config ReadonlyConfig to validate
- */
- private static void validateRequiredFields(ReadonlyConfig config) {
- String keys = config.getOptional(KEY_PATTERN).orElse(null);
- if (keys == null || keys.trim().isEmpty()) {
- throw new IllegalArgumentException(
- "Redis table configuration requires 'keys' parameter.");
- }
- if (!config.getOptional(DATA_TYPE).isPresent()) {
- throw new IllegalArgumentException(
- "Redis table configuration requires 'data_type'
parameter.");
- }
- }
-
/** Result class containing catalog table and deserialization schema. */
private static class TableConfigResult {
CatalogTable catalogTable;
diff --git
a/seatunnel-connectors-v2/connector-redis/src/main/java/org/apache/seatunnel/connectors/seatunnel/redis/config/RedisTableConfigsValidator.java
b/seatunnel-connectors-v2/connector-redis/src/main/java/org/apache/seatunnel/connectors/seatunnel/redis/config/RedisTableConfigsValidator.java
new file mode 100644
index 0000000000..0aeadd7c3a
--- /dev/null
+++
b/seatunnel-connectors-v2/connector-redis/src/main/java/org/apache/seatunnel/connectors/seatunnel/redis/config/RedisTableConfigsValidator.java
@@ -0,0 +1,51 @@
+/*
+ * 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.redis.config;
+
+import org.apache.seatunnel.api.configuration.ReadonlyConfig;
+import org.apache.seatunnel.api.configuration.util.ConditionExtension;
+import org.apache.seatunnel.api.configuration.util.OptionValidationException;
+
+import java.util.List;
+import java.util.Map;
+
+public class RedisTableConfigsValidator implements
ConditionExtension<List<Map<String, Object>>> {
+
+ @Override
+ public String description() {
+ return "each 'tables_configs' entry must declare a non-blank 'keys'
and a 'data_type'";
+ }
+
+ @Override
+ public boolean evaluate(ReadonlyConfig config, List<Map<String, Object>>
entries)
+ throws OptionValidationException {
+ for (int i = 0; i < entries.size(); i++) {
+ ReadonlyConfig entry = ReadonlyConfig.fromMap(entries.get(i));
+ String keys =
entry.getOptional(RedisBaseOptions.KEY_PATTERN).orElse(null);
+ if (keys == null || keys.trim().isEmpty()) {
+ throw new OptionValidationException(
+ "tables_configs[%d]: 'keys' must be configured and
non-blank", i);
+ }
+ if (!entry.getOptional(RedisBaseOptions.DATA_TYPE).isPresent()) {
+ throw new OptionValidationException(
+ "tables_configs[%d]: 'data_type' must be configured",
i);
+ }
+ }
+ return true;
+ }
+}
diff --git
a/seatunnel-connectors-v2/connector-redis/src/main/java/org/apache/seatunnel/connectors/seatunnel/redis/sink/RedisSinkFactory.java
b/seatunnel-connectors-v2/connector-redis/src/main/java/org/apache/seatunnel/connectors/seatunnel/redis/sink/RedisSinkFactory.java
index 2bfbd09733..7730fec61a 100644
---
a/seatunnel-connectors-v2/connector-redis/src/main/java/org/apache/seatunnel/connectors/seatunnel/redis/sink/RedisSinkFactory.java
+++
b/seatunnel-connectors-v2/connector-redis/src/main/java/org/apache/seatunnel/connectors/seatunnel/redis/sink/RedisSinkFactory.java
@@ -17,6 +17,7 @@
package org.apache.seatunnel.connectors.seatunnel.redis.sink;
+import org.apache.seatunnel.api.configuration.util.Conditions;
import org.apache.seatunnel.api.configuration.util.OptionRule;
import org.apache.seatunnel.api.options.SinkConnectorCommonOptions;
import org.apache.seatunnel.api.table.catalog.CatalogTable;
@@ -25,6 +26,7 @@ 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.redis.config.RedisBaseOptions;
+import
org.apache.seatunnel.connectors.seatunnel.redis.config.RedisNodesValidator;
import org.apache.seatunnel.connectors.seatunnel.redis.config.RedisSinkOptions;
import com.google.auto.service.AutoService;
@@ -69,6 +71,19 @@ public class RedisSinkFactory implements TableSinkFactory {
RedisBaseOptions.MODE,
RedisBaseOptions.RedisMode.CLUSTER,
RedisBaseOptions.NODES)
+ .conditional(
+ RedisBaseOptions.MODE,
+ RedisBaseOptions.RedisMode.SINGLE,
+ Conditions.notBlank(RedisBaseOptions.HOST),
+ Conditions.greaterOrEqual(RedisBaseOptions.PORT,
RedisBaseOptions.MIN_PORT)
+ .and(
+ Conditions.lessOrEqual(
+ RedisBaseOptions.PORT,
RedisBaseOptions.MAX_PORT)))
+ .conditional(
+ RedisBaseOptions.MODE,
+ RedisBaseOptions.RedisMode.CLUSTER,
+ Conditions.notEmpty(RedisBaseOptions.NODES),
+ Conditions.extension(RedisBaseOptions.NODES, new
RedisNodesValidator()))
.build();
}
}
diff --git
a/seatunnel-connectors-v2/connector-redis/src/main/java/org/apache/seatunnel/connectors/seatunnel/redis/source/RedisSourceFactory.java
b/seatunnel-connectors-v2/connector-redis/src/main/java/org/apache/seatunnel/connectors/seatunnel/redis/source/RedisSourceFactory.java
index ad4dd6a79c..c44700e56e 100644
---
a/seatunnel-connectors-v2/connector-redis/src/main/java/org/apache/seatunnel/connectors/seatunnel/redis/source/RedisSourceFactory.java
+++
b/seatunnel-connectors-v2/connector-redis/src/main/java/org/apache/seatunnel/connectors/seatunnel/redis/source/RedisSourceFactory.java
@@ -17,6 +17,7 @@
package org.apache.seatunnel.connectors.seatunnel.redis.source;
+import org.apache.seatunnel.api.configuration.util.Conditions;
import org.apache.seatunnel.api.configuration.util.OptionRule;
import org.apache.seatunnel.api.source.SeaTunnelSource;
import org.apache.seatunnel.api.source.SourceSplit;
@@ -25,7 +26,10 @@ 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.redis.config.RedisBaseOptions;
+import
org.apache.seatunnel.connectors.seatunnel.redis.config.RedisNodesValidator;
+import
org.apache.seatunnel.connectors.seatunnel.redis.config.RedisSingleTableDataTypeValidator;
import
org.apache.seatunnel.connectors.seatunnel.redis.config.RedisSourceOptions;
+import
org.apache.seatunnel.connectors.seatunnel.redis.config.RedisTableConfigsValidator;
import com.google.auto.service.AutoService;
@@ -48,6 +52,17 @@ public class RedisSourceFactory implements
TableSourceFactory {
public OptionRule optionRule() {
return OptionRule.builder()
.exclusive(RedisBaseOptions.TABLE_CONFIGS,
RedisBaseOptions.KEY_PATTERN)
+ .optional(
+ RedisBaseOptions.KEY_PATTERN,
+ Conditions.notBlank(RedisBaseOptions.KEY_PATTERN),
+ Conditions.extension(
+ RedisBaseOptions.KEY_PATTERN,
+ new RedisSingleTableDataTypeValidator()))
+ .optional(
+ RedisBaseOptions.TABLE_CONFIGS,
+ Conditions.notEmpty(RedisBaseOptions.TABLE_CONFIGS),
+ Conditions.extension(
+ RedisBaseOptions.TABLE_CONFIGS, new
RedisTableConfigsValidator()))
.optional(
RedisBaseOptions.DATA_TYPE,
RedisBaseOptions.MODE,
@@ -71,6 +86,19 @@ public class RedisSourceFactory implements
TableSourceFactory {
RedisBaseOptions.RedisMode.SINGLE,
RedisBaseOptions.HOST,
RedisBaseOptions.PORT)
+ .conditional(
+ RedisBaseOptions.MODE,
+ RedisBaseOptions.RedisMode.SINGLE,
+ Conditions.notBlank(RedisBaseOptions.HOST),
+ Conditions.greaterOrEqual(RedisBaseOptions.PORT,
RedisBaseOptions.MIN_PORT)
+ .and(
+ Conditions.lessOrEqual(
+ RedisBaseOptions.PORT,
RedisBaseOptions.MAX_PORT)))
+ .conditional(
+ RedisBaseOptions.MODE,
+ RedisBaseOptions.RedisMode.CLUSTER,
+ Conditions.notEmpty(RedisBaseOptions.NODES),
+ Conditions.extension(RedisBaseOptions.NODES, new
RedisNodesValidator()))
.conditional(
RedisSourceOptions.READ_KEY_ENABLED,
true,
diff --git
a/seatunnel-connectors-v2/connector-redis/src/test/java/org/apache/seatunnel/connectors/seatunnel/redis/RedisFactoryTest.java
b/seatunnel-connectors-v2/connector-redis/src/test/java/org/apache/seatunnel/connectors/seatunnel/redis/RedisFactoryTest.java
index ef2b3f0367..cf2ca61d79 100644
---
a/seatunnel-connectors-v2/connector-redis/src/test/java/org/apache/seatunnel/connectors/seatunnel/redis/RedisFactoryTest.java
+++
b/seatunnel-connectors-v2/connector-redis/src/test/java/org/apache/seatunnel/connectors/seatunnel/redis/RedisFactoryTest.java
@@ -17,17 +17,270 @@
package org.apache.seatunnel.connectors.seatunnel.redis;
+import org.apache.seatunnel.api.configuration.ReadonlyConfig;
+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.table.catalog.CatalogTable;
+import org.apache.seatunnel.api.table.catalog.PhysicalColumn;
+import org.apache.seatunnel.api.table.catalog.TableIdentifier;
+import org.apache.seatunnel.api.table.catalog.TableSchema;
+import org.apache.seatunnel.api.table.factory.TableSinkFactoryContext;
+import org.apache.seatunnel.api.table.factory.TableSourceFactoryContext;
+import org.apache.seatunnel.api.table.type.BasicType;
import org.apache.seatunnel.connectors.seatunnel.redis.sink.RedisSinkFactory;
import
org.apache.seatunnel.connectors.seatunnel.redis.source.RedisSourceFactory;
import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.Test;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.Arguments;
+import org.junit.jupiter.params.provider.MethodSource;
+
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.stream.Stream;
class RedisFactoryTest {
+ private static final OptionRule SOURCE_RULE = new
RedisSourceFactory().optionRule();
+ private static final OptionRule SINK_RULE = new
RedisSinkFactory().optionRule();
+
@Test
void optionRule() {
- Assertions.assertNotNull((new RedisSourceFactory()).optionRule());
- Assertions.assertNotNull((new RedisSinkFactory()).optionRule());
+ Assertions.assertNotNull(SOURCE_RULE);
+ Assertions.assertNotNull(SINK_RULE);
+ }
+
+ // accepted configs
+
+ @Test
+ void sourceSingleModeValid() {
+ Assertions.assertDoesNotThrow(() ->
validateSource(singleSourceConfig()));
+ }
+
+ @Test
+ void sourceClusterModeValid() {
+ Map<String, Object> config = clusterSourceConfig();
+ config.put("nodes", Arrays.asList("127.0.0.1:7000", "127.0.0.1:7001"));
+ Assertions.assertDoesNotThrow(() -> validateSource(config));
+ }
+
+ @Test
+ void sourceMultiTableValid() {
+ Assertions.assertDoesNotThrow(() ->
validateSource(multiTableSourceConfig()));
+ }
+
+ @Test
+ void sinkSingleModeValid() {
+ Assertions.assertDoesNotThrow(() -> validateSink(singleSinkConfig()));
+ }
+
+ // rejected configs
+
+ @ParameterizedTest(name = "source single-mode connection rejected: {0}")
+ @MethodSource("invalidSingleConnections")
+ void sourceSingleModeInvalidConnectionRejected(String name, Map<String,
Object> config) {
+ Assertions.assertThrows(OptionValidationException.class, () ->
validateSource(config));
+ }
+
+ @ParameterizedTest(name = "source cluster-mode nodes rejected: {0}")
+ @MethodSource("invalidClusterNodes")
+ void sourceClusterModeInvalidNodesRejected(String name, Map<String,
Object> config) {
+ Assertions.assertThrows(OptionValidationException.class, () ->
validateSource(config));
+ }
+
+ @ParameterizedTest(name = "source table config rejected: {0}")
+ @MethodSource("invalidTableConfigs")
+ void sourceTableConfigRejected(String name, Map<String, Object> config) {
+ Assertions.assertThrows(OptionValidationException.class, () ->
validateSource(config));
+ }
+
+ @ParameterizedTest(name = "sink config rejected: {0}")
+ @MethodSource("invalidSinkConfigs")
+ void sinkConfigRejected(String name, Map<String, Object> config) {
+ Assertions.assertThrows(OptionValidationException.class, () ->
validateSink(config));
+ }
+
+ @Test
+ void sourceSingleModeAggregatesMultipleErrors() {
+ Map<String, Object> config = singleSourceConfig();
+ config.put("host", " ");
+ config.put("port", 70000);
+ OptionValidationException ex =
+ Assertions.assertThrows(
+ OptionValidationException.class, () ->
validateSource(config));
+ String message = ex.getMessage();
+ Assertions.assertTrue(message.contains("(2 error"), message);
+ Assertions.assertTrue(message.contains("host"), message);
+ Assertions.assertTrue(message.contains("port"), message);
+ }
+
+ // factory-context regression
+
+ @Test
+ void sourceCreatedViaFactoryContext() {
+ ReadonlyConfig config = ReadonlyConfig.fromMap(singleSourceConfig());
+ ConfigValidator.of(config).validate(SOURCE_RULE);
+ TableSourceFactoryContext context =
+ new TableSourceFactoryContext(
+ config,
Thread.currentThread().getContextClassLoader());
+ Assertions.assertDoesNotThrow(
+ () -> new
RedisSourceFactory().createSource(context).createSource());
+ }
+
+ @Test
+ void sinkCreatedViaFactoryContext() {
+ ReadonlyConfig config = ReadonlyConfig.fromMap(singleSinkConfig());
+ ConfigValidator.of(config).validate(SINK_RULE);
+ TableSinkFactoryContext context =
+ new TableSinkFactoryContext(
+ catalogTable(), config,
Thread.currentThread().getContextClassLoader());
+ Assertions.assertDoesNotThrow(
+ () -> new RedisSinkFactory().createSink(context).createSink());
+ }
+
+ // parameterized-case providers
+
+ static Stream<Arguments> invalidSingleConnections() {
+ Map<String, Object> blankHost = singleSourceConfig();
+ blankHost.put("host", " ");
+ Map<String, Object> portTooSmall = singleSourceConfig();
+ portTooSmall.put("port", 0);
+ Map<String, Object> portTooLarge = singleSourceConfig();
+ portTooLarge.put("port", 70000);
+ return Stream.of(
+ Arguments.of("blank host", blankHost),
+ Arguments.of("port below range", portTooSmall),
+ Arguments.of("port above range", portTooLarge));
+ }
+
+ static Stream<Arguments> invalidClusterNodes() {
+ return Stream.of(
+ Arguments.of("empty nodes",
clusterNodes(Collections.emptyList())),
+ Arguments.of(
+ "malformed node",
+ clusterNodes(Arrays.asList("127.0.0.1:7000",
"host-without-port"))),
+ Arguments.of(
+ "port out of range",
+ clusterNodes(Arrays.asList("127.0.0.1:7000",
"127.0.0.1:99999"))),
+ Arguments.of(
+ "blank host in node",
+ clusterNodes(Arrays.asList("127.0.0.1:7000", "
:7001"))));
+ }
+
+ static Stream<Arguments> invalidTableConfigs() {
+ Map<String, Object> singleBlankKeys = singleSourceConfig();
+ singleBlankKeys.put("keys", " ");
+ Map<String, Object> singleMissingDataType = singleSourceConfig();
+ singleMissingDataType.remove("data_type");
+ Map<String, Object> multiBlankKeysInEntry = multiTableSourceConfig();
+ multiBlankKeysInEntry.put(
+ "tables_configs", Arrays.asList(tableEntry("key_a*"),
tableEntry("")));
+ Map<String, Object> entryMissingDataType = new HashMap<>();
+ entryMissingDataType.put("keys", "key_a*");
+ Map<String, Object> multiMissingDataType = multiTableSourceConfig();
+ multiMissingDataType.put("tables_configs",
Collections.singletonList(entryMissingDataType));
+ Map<String, Object> multiEmptyList = multiTableSourceConfig();
+ multiEmptyList.put("tables_configs", Collections.emptyList());
+ return Stream.of(
+ Arguments.of("single: blank keys", singleBlankKeys),
+ Arguments.of("single: missing data_type",
singleMissingDataType),
+ Arguments.of("multi: blank keys in entry",
multiBlankKeysInEntry),
+ Arguments.of("multi: missing data_type in entry",
multiMissingDataType),
+ Arguments.of("multi: empty list", multiEmptyList));
+ }
+
+ static Stream<Arguments> invalidSinkConfigs() {
+ Map<String, Object> portOutOfRange = singleSinkConfig();
+ portOutOfRange.put("port", -1);
+ Map<String, Object> clusterMalformedNode = singleSinkConfig();
+ clusterMalformedNode.remove("host");
+ clusterMalformedNode.remove("port");
+ clusterMalformedNode.put("mode", "CLUSTER");
+ clusterMalformedNode.put("nodes",
Collections.singletonList("bad_node"));
+ return Stream.of(
+ Arguments.of("single: port out of range", portOutOfRange),
+ Arguments.of("cluster: malformed node", clusterMalformedNode));
+ }
+
+ // helpers
+
+ private static void validateSource(Map<String, Object> config) {
+
ConfigValidator.of(ReadonlyConfig.fromMap(config)).validate(SOURCE_RULE);
+ }
+
+ private static void validateSink(Map<String, Object> config) {
+ ConfigValidator.of(ReadonlyConfig.fromMap(config)).validate(SINK_RULE);
+ }
+
+ private static Map<String, Object> clusterNodes(List<String> nodes) {
+ Map<String, Object> config = clusterSourceConfig();
+ config.put("nodes", nodes);
+ return config;
+ }
+
+ private static CatalogTable catalogTable() {
+ TableSchema schema =
+ TableSchema.builder()
+ .column(PhysicalColumn.of("id", BasicType.LONG_TYPE,
22, false, null, "id"))
+ .column(
+ PhysicalColumn.of(
+ "name", BasicType.STRING_TYPE, 128,
false, null, "name"))
+ .build();
+ return CatalogTable.of(
+ TableIdentifier.of("catalog", "default", null, "test_table"),
+ schema,
+ new HashMap<>(),
+ new ArrayList<>(),
+ null,
+ "catalog");
+ }
+
+ private static Map<String, Object> singleSourceConfig() {
+ Map<String, Object> config = new HashMap<>();
+ config.put("mode", "SINGLE");
+ config.put("host", "localhost");
+ config.put("port", 6379);
+ config.put("keys", "key_test*");
+ config.put("data_type", "KEY");
+ return config;
+ }
+
+ private static Map<String, Object> clusterSourceConfig() {
+ Map<String, Object> config = new HashMap<>();
+ config.put("mode", "CLUSTER");
+ config.put("keys", "key_test*");
+ config.put("data_type", "KEY");
+ return config;
+ }
+
+ private static Map<String, Object> multiTableSourceConfig() {
+ Map<String, Object> config = new HashMap<>();
+ config.put("host", "localhost");
+ config.put("port", 6379);
+ config.put("tables_configs", Arrays.asList(tableEntry("key_a*"),
tableEntry("key_b*")));
+ return config;
+ }
+
+ private static Map<String, Object> tableEntry(String keys) {
+ Map<String, Object> entry = new HashMap<>();
+ entry.put("keys", keys);
+ entry.put("data_type", "KEY");
+ return entry;
+ }
+
+ private static Map<String, Object> singleSinkConfig() {
+ Map<String, Object> config = new HashMap<>();
+ config.put("mode", "SINGLE");
+ config.put("host", "localhost");
+ config.put("port", 6379);
+ config.put("key", "my_key");
+ config.put("data_type", "KEY");
+ return config;
}
}
diff --git
a/seatunnel-connectors-v2/connector-redis/src/test/java/org/apache/seatunnel/connectors/seatunnel/redis/config/RedisTableConfigTest.java
b/seatunnel-connectors-v2/connector-redis/src/test/java/org/apache/seatunnel/connectors/seatunnel/redis/config/RedisTableConfigTest.java
index 8d9edfb35f..70c96f6e60 100644
---
a/seatunnel-connectors-v2/connector-redis/src/test/java/org/apache/seatunnel/connectors/seatunnel/redis/config/RedisTableConfigTest.java
+++
b/seatunnel-connectors-v2/connector-redis/src/test/java/org/apache/seatunnel/connectors/seatunnel/redis/config/RedisTableConfigTest.java
@@ -23,7 +23,6 @@ import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.Test;
import java.util.Arrays;
-import java.util.Collections;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
@@ -47,36 +46,6 @@ class RedisTableConfigTest {
Assertions.assertEquals(RedisDataType.STRING,
tables.get(0).getDataType());
}
- @Test
- void singleTableMissingDataTypeThrows() {
- Map<String, Object> map = new HashMap<>();
- map.put("keys", "key_test*");
-
- Assertions.assertThrows(
- IllegalArgumentException.class, () ->
RedisTableConfig.of(config(map)));
- }
-
- @Test
- void singleTableBlankKeysThrows() {
- Map<String, Object> map = new HashMap<>();
- map.put("keys", " ");
- map.put("data_type", "string");
-
- Assertions.assertThrows(
- IllegalArgumentException.class, () ->
RedisTableConfig.of(config(map)));
- }
-
- @Test
- void multiTableAppliesRequiredFieldValidationPerItem() {
- Map<String, Object> tableItem = new HashMap<>();
- tableItem.put("keys", "key_test*");
- Map<String, Object> map = new HashMap<>();
- map.put("tables_configs", Collections.singletonList(tableItem));
-
- Assertions.assertThrows(
- IllegalArgumentException.class, () ->
RedisTableConfig.of(config(map)));
- }
-
@Test
void multiTableBuildsEachItem() {
Map<String, Object> t1 = new HashMap<>();
diff --git
a/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-redis-e2e/src/test/java/org/apache/seatunnel/e2e/connector/redis/RedisValidationIT.java
b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-redis-e2e/src/test/java/org/apache/seatunnel/e2e/connector/redis/RedisValidationIT.java
new file mode 100644
index 0000000000..f53064ac83
--- /dev/null
+++
b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-redis-e2e/src/test/java/org/apache/seatunnel/e2e/connector/redis/RedisValidationIT.java
@@ -0,0 +1,58 @@
+/*
+ * 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.e2e.connector.redis;
+
+import org.apache.seatunnel.e2e.common.TestSuiteBase;
+import org.apache.seatunnel.e2e.common.container.TestContainer;
+
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.TestTemplate;
+import org.testcontainers.containers.Container;
+
+import lombok.extern.slf4j.Slf4j;
+
+import java.io.IOException;
+
+@Slf4j
+public class RedisValidationIT extends TestSuiteBase {
+
+ @TestTemplate
+ public void testSingleModePortOutOfRangeRejected(TestContainer container)
+ throws IOException, InterruptedException {
+ Container.ExecResult execResult =
+
container.executeJob("/redis-validation-single-port-out-of-range.conf");
+ Assertions.assertNotEquals(0, execResult.getExitCode());
+ String stderr = execResult.getStderr();
+ Assertions.assertNotNull(stderr, "stderr should not be null");
+ Assertions.assertTrue(
+ stderr.contains("Option validation failed") &&
stderr.contains("port"),
+ "stderr should report the failed port validation, but was: " +
stderr);
+ }
+
+ @TestTemplate
+ public void testClusterModeMalformedNodeRejected(TestContainer container)
+ throws IOException, InterruptedException {
+ Container.ExecResult execResult =
+
container.executeJob("/redis-validation-cluster-malformed-node.conf");
+ Assertions.assertNotEquals(0, execResult.getExitCode());
+ String stderr = execResult.getStderr();
+ Assertions.assertNotNull(stderr, "stderr should not be null");
+ Assertions.assertTrue(
+ stderr.contains("nodes") && stderr.contains("host:port"),
+ "stderr should report the failed nodes validation, but was: "
+ stderr);
+ }
+}
diff --git
a/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-redis-e2e/src/test/resources/redis-validation-cluster-malformed-node.conf
b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-redis-e2e/src/test/resources/redis-validation-cluster-malformed-node.conf
new file mode 100644
index 0000000000..dd587c8353
--- /dev/null
+++
b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-redis-e2e/src/test/resources/redis-validation-cluster-malformed-node.conf
@@ -0,0 +1,48 @@
+#
+# 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.
+#
+
+# Negative case for declarative validation (#11007): cluster-mode `nodes`
contains a
+# malformed entry (no `host:port`), so the job must be rejected at submission
time by
+# RedisSourceFactory.optionRule() (RedisNodesValidator) before any connection
is
+# attempted. No Redis instance is required.
+
+env {
+ parallelism = 1
+ job.mode = "BATCH"
+}
+
+source {
+ Redis {
+ nodes = ["redis-cluster-0:6379", "host-without-port"]
+ mode = "CLUSTER"
+ auth = "SeaTunnel"
+ keys = "key_test*"
+ data_type = string
+ batch_size = 33
+ }
+}
+
+sink {
+ Redis {
+ nodes = ["redis-cluster-0:6379", "redis-cluster-1:6379"]
+ mode = "CLUSTER"
+ auth = "SeaTunnel"
+ key = "key_list"
+ data_type = list
+ batch_size = 33
+ }
+}
diff --git
a/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-redis-e2e/src/test/resources/redis-validation-single-port-out-of-range.conf
b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-redis-e2e/src/test/resources/redis-validation-single-port-out-of-range.conf
new file mode 100644
index 0000000000..f021b6ab6e
--- /dev/null
+++
b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-redis-e2e/src/test/resources/redis-validation-single-port-out-of-range.conf
@@ -0,0 +1,47 @@
+#
+# 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.
+#
+
+# Negative case for declarative validation (#11007): single-mode `port` is out
of the
+# [1, 65535] range, so the job must be rejected at submission time by
RedisSourceFactory
+# .optionRule() before any connection is attempted. No Redis instance is
required.
+
+env {
+ parallelism = 1
+ job.mode = "BATCH"
+}
+
+source {
+ Redis {
+ host = "redis-e2e"
+ port = 70000
+ auth = "U2VhVHVubmVs"
+ keys = "key_test*"
+ data_type = string
+ batch_size = 33
+ }
+}
+
+sink {
+ Redis {
+ host = "redis-e2e"
+ port = 6379
+ auth = "U2VhVHVubmVs"
+ key = "key_list"
+ data_type = list
+ batch_size = 33
+ }
+}
