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 7a591a0411 [Improve][Connector-V2][Redis] Align OptionRule with docs
and dedupe config validation (#11195)
7a591a0411 is described below
commit 7a591a04115a2dc15660e6774546cf8b679f47f4
Author: Shuai Liu <[email protected]>
AuthorDate: Sat Jun 27 21:10:49 2026 +0800
[Improve][Connector-V2][Redis] Align OptionRule with docs and dedupe config
validation (#11195)
---
.../seatunnel/redis/config/RedisTableConfig.java | 101 ++++++---------------
.../seatunnel/redis/sink/RedisSinkFactory.java | 4 +-
.../seatunnel/redis/source/RedisSourceFactory.java | 1 -
.../redis/config/RedisTableConfigTest.java | 97 ++++++++++++++++++++
4 files changed, 126 insertions(+), 77 deletions(-)
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 525106d96b..2f90ff5aad 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
@@ -114,44 +114,44 @@ public class RedisTableConfig implements Serializable {
* @return List of RedisTableConfig
*/
public static List<RedisTableConfig> of(ReadonlyConfig config) {
- // Check if using multi-table mode
+ // Multi-table mode: each table_configs item is parsed as an
independent table.
if (config.getOptional(RedisBaseOptions.TABLE_CONFIGS).isPresent()) {
List<Map<String, Object>> tableConfigMaps =
config.get(RedisBaseOptions.TABLE_CONFIGS);
return tableConfigMaps.stream()
.map(ReadonlyConfig::fromMap)
- .map(RedisTableConfig::buildFromConfig)
+ .map(RedisTableConfig::buildTableConfig)
.collect(Collectors.toList());
- } else {
- // Single table mode (backward compatibility)
- return Collections.singletonList(buildSingleTableConfig(config));
}
+ // Single-table mode: top-level options describe one table.
+ return Collections.singletonList(buildTableConfig(config));
}
/**
- * Build RedisTableConfig from ReadonlyConfig (for multi-table mode).
+ * Build a fully initialized {@link RedisTableConfig} from a single
table-level {@link
+ * ReadonlyConfig}. Shared by both single-table (top-level config) and
multi-table (each {@code
+ * tables_configs} item) modes.
*
- * @param tableConfig ReadonlyConfig for a single table (table-level
config)
+ * @param config ReadonlyConfig describing one table
* @return Fully initialized RedisTableConfig with runtime objects
*/
- private static RedisTableConfig buildFromConfig(ReadonlyConfig
tableConfig) {
- // Validate required fields
- validateRequiredFields(tableConfig);
+ private static RedisTableConfig buildTableConfig(ReadonlyConfig config) {
+ // Validate required fields are present
+ validateRequiredFields(config);
// Build catalog table and deserialization schema
- TableConfigResult result =
- buildCatalogTableAndSchema(tableConfig,
tableConfig.get(KEY_PATTERN));
+ TableConfigResult result = buildCatalogTableAndSchema(config,
config.get(KEY_PATTERN));
return RedisTableConfig.builder()
- .keys(tableConfig.get(KEY_PATTERN))
- .dataType(tableConfig.get(DATA_TYPE))
- .batchSize(tableConfig.get(BATCH_SIZE))
- .format(tableConfig.get(FORMAT))
-
.schema(tableConfig.getOptional(ConnectorCommonOptions.SCHEMA).orElse(null))
- .hashKeyParseMode(tableConfig.get(HASH_KEY_PARSE_MODE))
- .readKeyEnabled(tableConfig.get(READ_KEY_ENABLED))
+ .keys(config.get(KEY_PATTERN))
+ .dataType(config.get(DATA_TYPE))
+ .batchSize(config.get(BATCH_SIZE))
+ .format(config.get(FORMAT))
+
.schema(config.getOptional(ConnectorCommonOptions.SCHEMA).orElse(null))
+ .hashKeyParseMode(config.get(HASH_KEY_PARSE_MODE))
+ .readKeyEnabled(config.get(READ_KEY_ENABLED))
.keyFieldName(result.keyFieldName)
-
.singleFieldName(tableConfig.getOptional(SINGLE_FIELD_NAME).orElse(null))
- .fieldDelimiter(tableConfig.get(FIELD_DELIMITER))
+
.singleFieldName(config.getOptional(SINGLE_FIELD_NAME).orElse(null))
+ .fieldDelimiter(config.get(FIELD_DELIMITER))
.tablePath(result.tablePath)
.catalogTable(result.catalogTable)
.deserializationSchema(result.deserializationSchema)
@@ -159,52 +159,19 @@ public class RedisTableConfig implements Serializable {
}
/**
- * Build a single table configuration from global config (backward
compatibility).
- *
- * @param config ReadonlyConfig instance
- * @return RedisTableConfig
- */
- private static RedisTableConfig buildSingleTableConfig(ReadonlyConfig
config) {
- // Validate that required fields for single table mode are present
- validateRequiredFields(config);
-
- // Build catalog table and deserialization schema
- TableConfigResult result = buildCatalogTableAndSchema(config,
config.get(KEY_PATTERN));
-
- RedisTableConfig tableConfig =
- RedisTableConfig.builder()
- .keys(config.get(KEY_PATTERN))
- .dataType(config.get(DATA_TYPE))
- .batchSize(config.get(BATCH_SIZE))
- .format(config.get(FORMAT))
-
.schema(config.getOptional(ConnectorCommonOptions.SCHEMA).orElse(null))
- .hashKeyParseMode(config.get(HASH_KEY_PARSE_MODE))
- .readKeyEnabled(config.get(READ_KEY_ENABLED))
- .keyFieldName(result.keyFieldName)
-
.singleFieldName(config.getOptional(SINGLE_FIELD_NAME).orElse(null))
- .fieldDelimiter(config.get(FIELD_DELIMITER))
- .tablePath(result.tablePath)
- .catalogTable(result.catalogTable)
- .deserializationSchema(result.deserializationSchema)
- .build();
-
- validateTableConfig(tableConfig);
- return tableConfig;
- }
-
- /**
- * Validate required fields in configuration.
+ * Validate required fields in a (table-level) configuration.
*
* @param config ReadonlyConfig to validate
*/
private static void validateRequiredFields(ReadonlyConfig config) {
- if (!config.getOptional(KEY_PATTERN).isPresent()) {
+ String keys = config.getOptional(KEY_PATTERN).orElse(null);
+ if (keys == null || keys.trim().isEmpty()) {
throw new IllegalArgumentException(
- "Redis table configuration requires 'keys' parameter. ");
+ "Redis table configuration requires 'keys' parameter.");
}
if (!config.getOptional(DATA_TYPE).isPresent()) {
throw new IllegalArgumentException(
- "Redis table configuration requires 'data_type' parameter.
");
+ "Redis table configuration requires 'data_type'
parameter.");
}
}
@@ -264,20 +231,4 @@ public class RedisTableConfig implements Serializable {
return result;
}
-
- /**
- * Validate a single table configuration.
- *
- * @param tableConfig RedisTableConfig to validate
- */
- private static void validateTableConfig(RedisTableConfig tableConfig) {
- if (tableConfig.getKeys() == null ||
tableConfig.getKeys().trim().isEmpty()) {
- throw new IllegalArgumentException(
- "Redis configuration must specify 'keys' parameter.");
- }
- if (tableConfig.getDataType() == null) {
- throw new IllegalArgumentException(
- "Redis configuration must specify 'data_type' parameter.");
- }
- }
}
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 53c5b62b22..2bfbd09733 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
@@ -50,8 +50,10 @@ public class RedisSinkFactory implements TableSinkFactory {
RedisBaseOptions.MODE,
RedisBaseOptions.AUTH,
RedisBaseOptions.USER,
- RedisBaseOptions.KEY_PATTERN,
+ RedisBaseOptions.DB_NUM,
+ RedisBaseOptions.BATCH_SIZE,
RedisBaseOptions.FORMAT,
+ RedisBaseOptions.FIELD_DELIMITER,
RedisSinkOptions.EXPIRE,
RedisSinkOptions.SUPPORT_CUSTOM_KEY,
RedisSinkOptions.VALUE_FIELD,
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 4007cc9395..ad4dd6a79c 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
@@ -54,7 +54,6 @@ public class RedisSourceFactory implements TableSourceFactory
{
RedisSourceOptions.HASH_KEY_PARSE_MODE,
RedisBaseOptions.AUTH,
RedisBaseOptions.USER,
- RedisBaseOptions.KEY,
RedisSourceOptions.READ_KEY_ENABLED,
RedisSourceOptions.SINGLE_FIELD_NAME,
RedisSourceOptions.KEY_FIELD_NAME,
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
new file mode 100644
index 0000000000..8d9edfb35f
--- /dev/null
+++
b/seatunnel-connectors-v2/connector-redis/src/test/java/org/apache/seatunnel/connectors/seatunnel/redis/config/RedisTableConfigTest.java
@@ -0,0 +1,97 @@
+/*
+ * 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.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;
+
+class RedisTableConfigTest {
+
+ private static ReadonlyConfig config(Map<String, Object> map) {
+ return ReadonlyConfig.fromMap(map);
+ }
+
+ @Test
+ void singleTableBuildsOneTable() {
+ Map<String, Object> map = new HashMap<>();
+ map.put("keys", "key_test*");
+ map.put("data_type", "string");
+
+ List<RedisTableConfig> tables = RedisTableConfig.of(config(map));
+
+ Assertions.assertEquals(1, tables.size());
+ Assertions.assertEquals("key_test*", tables.get(0).getKeys());
+ 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<>();
+ t1.put("keys", "k1*");
+ t1.put("data_type", "string");
+ Map<String, Object> t2 = new HashMap<>();
+ t2.put("keys", "k2*");
+ t2.put("data_type", "hash");
+ Map<String, Object> map = new HashMap<>();
+ map.put("tables_configs", Arrays.asList(t1, t2));
+
+ List<RedisTableConfig> tables = RedisTableConfig.of(config(map));
+
+ Assertions.assertEquals(2, tables.size());
+ Assertions.assertEquals(RedisDataType.STRING,
tables.get(0).getDataType());
+ Assertions.assertEquals(RedisDataType.HASH,
tables.get(1).getDataType());
+ }
+}
