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-12491-25bcc53fd1b2850b79127980de4f9bcb41e545cb in repository https://gitbox.apache.org/repos/asf/seatunnel.git
commit b663491db900cb7198e7908e07b0efe1121f0497 Author: NganWave <[email protected]> AuthorDate: Wed Oct 14 10:53:06 2026 +0000 Improve/Support JSON params via CLI (#12491) --- docs/en/introduction/concepts/config.md | 13 + .../introduction/concepts/incompatible-changes.md | 25 ++ docs/zh/introduction/concepts/config.md | 15 +- .../introduction/concepts/incompatible-changes.md | 23 ++ .../seatunnel/common/utils/ConfigValueUtils.java | 233 ++++++++++++++++++ .../common/utils/ConfigValueUtilsTest.java | 158 ++++++++++++ .../apache/seatunnel/core/starter/SeaTunnel.java | 4 +- .../core/starter/command/ConfEncryptCommand.java | 29 ++- .../core/starter/command/ParameterSplitter.java | 37 ++- .../core/starter/utils/ConfigBuilder.java | 186 +++++++++----- .../starter/command/ConfEncryptCommandTest.java | 63 +++++ .../starter/command/ParameterSplitterTest.java | 181 ++++++++++++++ .../core/starter/utils/ConfigBuilderTest.java | 28 +++ .../core/starter/utils/ConfigShadeTest.java | 79 ++++++ .../src/test/resources/config_json_variables.conf | 41 ++++ .../src/test/resources/json_params.conf | 43 ++++ .../seatunnel/jdbc/JdbcMysqlJsonParamIT.java | 270 +++++++++++++++++++++ .../src/test/resources/jdbc_mysql_json_params.conf | 59 +++++ .../resources/jdbc_mysql_plugin_json_params.conf | 82 +++++++ 19 files changed, 1485 insertions(+), 84 deletions(-) diff --git a/docs/en/introduction/concepts/config.md b/docs/en/introduction/concepts/config.md index fdbf54003c..85e5db9d28 100644 --- a/docs/en/introduction/concepts/config.md +++ b/docs/en/introduction/concepts/config.md @@ -320,7 +320,20 @@ sink { ``` ### Important Notes: +- To avoid unexpected data synchronization errors, duplicate keys are not allowed in the `-i` parameters. +- If a value contains commas, it must be wrapped with `\"`. For example: `-i read_cols=\"id,name\"`. However, array types do not require wrapping with `\"`; for example: `-i include_fields=[id,name]`. - If a value contains special characters like `(`, enclose it in single quotes (`'`). +- If the value is a map type, you can pass it as a JSON string, which supports arrays and JSON objects nested to any depth. You can format the parameter value by enclosing in a pair of single quotes as follows: + + `-i mysql_properties='{"connectTimeout":"5000","connectionTimeZone":"UTC","serverTimezone":"UTC","useSSL":"false","allowPublicKeyRetrieval":"true"}'` +- JSON parameter values are compatible with HOCON syntax: an equals sign (`=`) can be used instead of a colon (`:`). For example, a='{"k1"="v1","k2"="v2"}' is equivalent to a='{"k1":"v1","k2":"v2"}', but using a colon is still recommended. +- If the value is an array with map inside, json parameters should be enclosed in single quotes (`\'`): + + `-i table_list=['{"table_path":"movie_lens.tags_test"}','{"table_path":"movie_lens.ml_*","use_regex":"true"}']` +- If the value is a map with array inside, the array parameter format inside the JSON must adhere to JSON format specifications. Keys must be enclosed in double quotes, while elements within the array can be either quoted or unquoted, as they will be parsed automatically: + + `-i table_filter='{"plugin_input":"mysql_source","plugin_output":"filter","include_fields":["movie_id","unix_time"]}'` +- Placeholders within the JSON content of parameter values will not be resolved or replaced; they will be preserved as-is as part of the content. - If the substitution variable contains double or single quotes (e.g., `"resName"` or `"nameVal"`), you need to include them with the value. - The value cannot contain spaces (`' '`). For example, `-i jobName='this is a job name'` will be replaced with `job.name = "this"`. You can use environment variables to pass values with spaces. - For dynamic parameters, you can use the following format: `-i date=$(date +"%Y%m%d")`. diff --git a/docs/en/introduction/concepts/incompatible-changes.md b/docs/en/introduction/concepts/incompatible-changes.md index 243f9b9a28..42c3cc4a2f 100644 --- a/docs/en/introduction/concepts/incompatible-changes.md +++ b/docs/en/introduction/concepts/incompatible-changes.md @@ -22,6 +22,31 @@ You need to check this document before you upgrade to related version. 4. If you customized `${SEATUNNEL_HOME}/config/jvm_options` (or the client, master and worker variants), check your additions for flags that Java 11 removed, such as `-XX:+UseConcMarkSweepGC` or `-XX:MaxPermSize`, because the JVM refuses to start on an unrecognized flag. The options shipped by default are already Java 11 compatible. 5. You do not have to copy the new JDK module flags into a preserved config directory. `seatunnel.sh` and `seatunnel-cluster.sh` append the mandatory `--add-opens`/`--add-exports` flags (`java.base/java.lang`, `java.net`, `java.nio`, `java.util`, `sun.nio.ch`, and `java.security.jgss/sun.security.krb5`) themselves and skip any your `jvm_*_options` already carries, so an in-place upgrade that keeps an old `config/` directory (a mounted Docker volume or a Kubernetes ConfigMap) still st [...] +### User Variables(`-i` variables) + +- **`-i` values are no longer visible via `System.getProperties()`** + + - Previously, `-i` parameters were injected into `System.getProperties()` and accessible via `System.getProperty()`. Starting from this version, `-i` parameters are resolved through `Config.resolveWith()` and are no longer visible via `System.getProperties()`. + +- **Duplicate keys via `-i` are no longer supported** + - Duplicate keys passed via `-i` are now rejected with an error. + Previously, duplicate keys were silently overridden, which could cause unexpected data sync errors due to mistakes such as copy-paste typo. + - If your code generation or CLI produces duplicated keys, please remove the unused ones. + +- **Double-quoted `-i` values are now unwrapped** + - For example, `-i k="v"` is now equivalent to `-i k=v` (quotes are stripped). Double quotes are mainly intended for string values that contain structural delimiters such as commas (`,`), curly braces (`{}`), and square brackets (`[]`). Plain values without such characters do not need double quotes. + +- **Unbalanced `{}` or `[]` are partially parsed into maps and lists** + + - In `-i k=[a,b],c]`, `[a,b]` is parsed as an array, and `c]` is ignored because it has no = (it is not kept as a string). + - Balanced values that start with `{` or `[` but are not valid HOCON/JSON now fail with a `BadValue` error at parse time. This includes previously working unquoted values such as `-i pattern={a,b}`. To keep them as plain strings, wrap the value in escaped double quotes: + `-i pattern=\"{a,b}\"` + +- **`-i` keys and values are now trimmed, and empty keys are rejected** + - `-i 'k1= , k2=v2'` now explicitly removes whitespace: `k1` becomes an empty string, and the key of the second pair becomes `k2` instead of ` k2`. + - `-i ' =v1 , k2=v2'` now reports an error, because the key for `v1` is empty after trimming. + - `-i`: Meaningful leading and trailing whitespace is silently removed. + ### SQL TINYINT array schema The Zeta SQL ARRAY function now declares TINYINT elements as `ARRAY<TINYINT>`, matching the Byte values it emits. The previous `ARRAY<STRING>` declaration could fail in schema-dependent row consumers. Update downstream declarations that assumed STRING elements; cast the SQL values to STRING explicitly when that schema is required. Restart affected jobs with the corrected schema rather than restoring state that relies on the old declaration. diff --git a/docs/zh/introduction/concepts/config.md b/docs/zh/introduction/concepts/config.md index 29d46d5356..627a42753f 100644 --- a/docs/zh/introduction/concepts/config.md +++ b/docs/zh/introduction/concepts/config.md @@ -322,8 +322,21 @@ sink { ``` 一些注意事项: - +- 为了避免非预期的数据同步错误，`-i` 传参不能出现重复的key。 +- 如果值包含逗号，需要用`\"`包裹起来，如 `-i read_cols=\"id,name\"`，但list和map类型不需要用`\"`包裹，如`-i include_fields=[id,name]`。 - 如果值包含特殊字符，如`(`，请使用`'`引号将其括起来。 +- 如果值是map类型，可以使用json传参，参数值支持任意深度的数组和json嵌套。参数值使用单引号包裹标准json，如下： + + `-i mysql_properties='{"connectTimeout":"5000","connectionTimeZone":"UTC","serverTimezone":"UTC","useSSL":"false","allowPublicKeyRetrieval":"true"}'` +- json参数值兼容hocon语法，可以用等号(`=`)代替冒号(`:`), 如 `a='{"k1"="v1","k2"="v2"}'` 与`a='{"k1":"v1","k2":"v2"}'` 等效，但仍然建议用冒号。 + +- 如果值是array中包含map类型，json参数需要用单引号(`\'`)包裹： + + `-i table_list=['{"table_path":"movie_lens.tags_test"}','{"table_path":"movie_lens.ml_*","use_regex":"true"}']` +- 如果值是map中包含array类型，json内部的array参数格式需要遵循json格式规范，key要带上双引号，array中的元素可以带引号也可以不带，会自动解析： + + `-i table_filter='{"plugin_input":"mysql_source","plugin_output":"filter","include_fields":["movie_id","unix_time"]}'` +- 参数值的json内容中的占位符不会被解析，会作为内容的一部分原样保留。 - 如果替换变量包含`"`或`'`(如`"resName"`和`"nameVal"`)，需要添加`"`。 - 值不能包含空格`' '`。例如, `-i jobName='this is a job name'`将被替换为`job.name = "this"`。 你可以使用环境变量传递带有空格的值。 - 如果要使用动态参数,可以使用以下格式: `-i date=$(date +"%Y%m%d")`。 diff --git a/docs/zh/introduction/concepts/incompatible-changes.md b/docs/zh/introduction/concepts/incompatible-changes.md index 861a1d4c5d..454b9848bd 100644 --- a/docs/zh/introduction/concepts/incompatible-changes.md +++ b/docs/zh/introduction/concepts/incompatible-changes.md @@ -21,6 +21,29 @@ 4. 如果您修改过 `${SEATUNNEL_HOME}/config/jvm_options`（以及 client、master、worker 对应的变体），请检查自己添加的参数中是否包含 Java 11 已移除的选项，例如 `-XX:+UseConcMarkSweepGC` 或 `-XX:MaxPermSize`，JVM 遇到无法识别的参数会直接拒绝启动。发行包默认提供的参数已经兼容 Java 11。 5. 无需把新增的 JDK 模块参数手工复制到保留下来的配置目录中。`seatunnel.sh` 和 `seatunnel-cluster.sh` 会自行追加必需的 `--add-opens`/`--add-exports` 参数（`java.base/java.lang`、`java.net`、`java.nio`、`java.util`、`sun.nio.ch`，以及 `java.security.jgss/sun.security.krb5`），并跳过您的 `jvm_*_options` 中已有的同名参数，因此原地升级并保留旧的 `config/` 目录（挂载的 Docker 卷或 Kubernetes ConfigMap）时，这些参数依然生效。当检测到的 JVM 版本低于 11 时，同样的脚本会直接以明确的 `SeaTunnel requires Java 11 or newer` 提示退出，而不是让 Java 8 启动器输出原始的 `Unrecognized option` 错误。 +### 用户变量(`-i` 参数) + +- **`-i` 参数值在 `System.getProperties()`中不可见** + + - 以前参数值通过 `-i` 注入到 `System.getProperties()` 并通过 `System.getProperty()`访问。从这个版本开始,`-i` 参数通过 `Config.resolveWith()` 解析，并且不能通过 `System.getProperties()` 访问。 + +- **`-i` 参数不再允许重复key** + - `-i` 参数如果有重复key会直接报错，之前重复的key对应的参数值会静默覆盖已有的key,会导致因复制粘贴等问题造成的重复key产生非预期的数据同步错误。 + - 如有代码生成或者cli输入产生的重复key,请检查后移除不需要的参数配置。 + +- **双引号包裹的 `-i` 值会被解包** + - 例如 `-i k="v"` 现在等价于 `-i k=v`（引号被剥掉）。双引号主要针对包含逗号(`,`),花括号(`{}`)和方括号(`[]`)等包含结构化定界符的字符串参数值使用，其他普通字符可不用双引号。 + +- **不平衡的`{}`或`[]`会被部分解析成map和list** + - `-i k=[a,b],c]` 中`[a,b]`会当成数组解析，`c]` 因为没有 `=` 会被忽略（不会作为字符串保留） + - 以 `{` 或 `[` 开头、括号平衡但不是合法 HOCON/JSON 的参数值，现在会在解析阶段抛出 BadValue 错误。之前能正常工作的未加引号的值（如 `-i pattern={a,b} `）会因此失败。如需保留为普通字符串，请用转义双引号包裹： + `-i pattern=\"{a,b}\"` + +- **`-i` key 和 value 现在会 trim，空 key 会被拒绝** + - `-i 'k1= , k2=v2'` 现在会明确将空格去掉，k1的会变为空字符串，第二个键值对的key会变为`k2`而不是带空格的` k2`。 + - `-i ' =v1 , k2=v2'` 现在会报错，因为v1的key在trim后为空。 + - 有意义的首尾空白会被静默去除。 + ### SQL TINYINT 数组模式 Zeta SQL ARRAY 函数现在将 TINYINT 元素声明为 `ARRAY<TINYINT>`，与实际输出的 Byte 值一致。此前错误的 `ARRAY<STRING>` 声明会导致依赖模式的行处理失败。请更新假定元素为 STRING 的下游声明；需要字符串模式时，在 SQL 中显式将值转换为 STRING。受影响的作业应使用修正后的模式重新启动，不要恢复依赖旧声明的状态。 diff --git a/seatunnel-common/src/main/java/org/apache/seatunnel/common/utils/ConfigValueUtils.java b/seatunnel-common/src/main/java/org/apache/seatunnel/common/utils/ConfigValueUtils.java new file mode 100644 index 0000000000..ee7bdaaeef --- /dev/null +++ b/seatunnel-common/src/main/java/org/apache/seatunnel/common/utils/ConfigValueUtils.java @@ -0,0 +1,233 @@ +/* + * 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.common.utils; + +import org.apache.seatunnel.shade.com.typesafe.config.Config; +import org.apache.seatunnel.shade.com.typesafe.config.ConfigException; +import org.apache.seatunnel.shade.com.typesafe.config.ConfigFactory; +import org.apache.seatunnel.shade.com.typesafe.config.ConfigOriginFactory; +import org.apache.seatunnel.shade.com.typesafe.config.ConfigValue; +import org.apache.seatunnel.shade.com.typesafe.config.ConfigValueFactory; +import org.apache.seatunnel.shade.org.apache.commons.lang3.StringUtils; + +import java.util.ArrayDeque; +import java.util.Arrays; +import java.util.Deque; +import java.util.HashSet; +import java.util.Set; + +public class ConfigValueUtils { + + // Characters that may appear right before a start quote. + private static final Set<Character> START_DELIMITERS = + new HashSet<>(Arrays.asList('=', ':', '{', '[', ',')); + + // Characters that may appear right after an end quote. + private static final Set<Character> END_DELIMITERS = + new HashSet<>(Arrays.asList(',', '}', ']', ':')); + + private ConfigValueUtils() {} + + /** + * Parses a raw string value into a {@link ConfigValue}. + * + * <p>The parsing rules are: + * + * <ul> + * <li>{@code null} is converted to a {@code ConfigValue} holding {@code null}. + * <li>A value wrapped in double quotes is unwrapped and treated as a plain string, e.g. + * {@code "123"} becomes the string {@code 123} rather than a number. + * <li>A value that starts with {@code "{"} and ends with {@code "}"}, or starts with {@code + * "["} and ends with {@code "]"}, is parsed as a JSON object or array. + * <li>Any other value is treated as a plain string. + * </ul> + * + * @param value the raw string value to parse + * @return the parsed {@link ConfigValue} + * @throws ConfigException.BadValue if the value looks like a JSON object or array but fails to + * parse + */ + public static ConfigValue parseValue(String value) { + if (value == null) { + return ConfigValueFactory.fromAnyRef(null); + } + + if (value.isEmpty()) { + return ConfigValueFactory.fromAnyRef(""); + } + + if (value.startsWith("\"") && value.endsWith("\"") && value.length() > 1) { + value = StringUtils.unwrap(value, "\""); + return ConfigValueFactory.fromAnyRef(value); + } + + if (isStructured(value)) { + try { + Config parsed = ConfigFactory.parseString("v = " + value); + return parsed.root().get("v"); + } catch (ConfigException e) { + throw new ConfigException.BadValue( + ConfigOriginFactory.newSimple(), + "", + String.format( + "Value '%s' looks like JSON or Array but failed to parse. " + + "If you intended to pass a Map/List, please check the syntax. " + + "If you intended to pass a plain string with comma, wrap the entire value in double quotes (e.g., \"your_value\"). ", + value), + e); + } + } + + return ConfigValueFactory.fromAnyRef(value); + } + + /** + * Returns {@code true} if the quote character at {@code quoteIndex} is escaped by a preceding + * backslash. + * + * <p>A quote is treated as escaped when it is preceded by an odd number of consecutive + * backslashes, e.g. {@code \"} is escaped, {@code \\"} is not. + * + * @param value the user input string via {@code -i} + * @param quoteIndex index of the current quote character in {@code value} + * @return {@code true} if the quote at {@code quoteIndex} is escaped + */ + public static boolean isEscapedQuote(String value, int quoteIndex) { + int backslashCount = 0; + int i = quoteIndex - 1; + while (i >= 0 && value.charAt(i) == '\\') { + backslashCount++; + i--; + } + return backslashCount % 2 == 1; + } + + /** + * Returns the updated "inside quotes" state after scanning the character at {@code quoteIndex} + * in {@code value}. + * + * @param value the user input string being scanned + * @param quoteIndex index of the current character in {@code value} + * @param insideQuotes the quote state before processing this character + * @return the quote state after processing this character + */ + public static boolean updateQuoteState(String value, int quoteIndex, boolean insideQuotes) { + + boolean isStartWrapper = isStartWrapper(insideQuotes, value, quoteIndex); + boolean isEndWrapper = isEndWrapper(insideQuotes, value, quoteIndex); + + if (isStartWrapper) { + insideQuotes = true; + } else if (isEndWrapper) { + insideQuotes = false; + } else { + insideQuotes = !insideQuotes; + } + return insideQuotes; + } + + /** + * Checks if the quote at {@code quoteIndex} is a start wrapper: not inside quotes, and preceded + * by a {@link #START_DELIMITERS} or the start of string. + * + * @param insideQuotes whether the caller is currently inside quotes + * @param value the string being scanned + * @param quoteIndex index of the quote character to check + * @return {@code true} if the quote starts a quoted region + */ + private static boolean isStartWrapper(boolean insideQuotes, String value, int quoteIndex) { + char prev = (quoteIndex > 0) ? value.charAt(quoteIndex - 1) : 0; + char beforePrev = (quoteIndex > 1) ? value.charAt(quoteIndex - 2) : 0; + + return !insideQuotes + && (quoteIndex == 0 + || START_DELIMITERS.contains(prev) + || (prev == ' ' && START_DELIMITERS.contains(beforePrev))); + } + + /** + * Checks if the quote at {@code quoteIndex} is an end wrapper: inside quotes, and followed by + * an {@link #END_DELIMITERS} or the end of string. + * + * <p>Checks are ordered: an already-closed quote state short-circuits first, then the + * end-of-string case, then the immediate next character, then the "space + end delimiter" form. + * A sentinel {@code 0} is used for missing following characters at the string boundary, so they + * never match a delimiter. + * + * @param insideQuotes whether the caller is currently inside quotes + * @param value the string being scanned + * @param quoteIndex index of the quote character to check + * @return {@code true} if the quote ends a quoted region + */ + private static boolean isEndWrapper(boolean insideQuotes, String value, int quoteIndex) { + char next = (quoteIndex + 1 < value.length()) ? value.charAt(quoteIndex + 1) : 0; + char afterNext = (quoteIndex + 2 < value.length()) ? value.charAt(quoteIndex + 2) : 0; + + return insideQuotes + && (quoteIndex == value.length() - 1 + || END_DELIMITERS.contains(next) + || (next == ' ' && END_DELIMITERS.contains(afterNext))); + } + + /** + * Checks if the value is a balanced structured string (e.g., JSON/HOCON). Returns false if + * brackets are unbalanced, empty, or unpaired '{' or '['. Characters inside quotes are ignored + * for bracket counting. + * + * @param value user input string value + * @return {@code true} if the value is a structured string + */ + private static boolean isStructured(String value) { + if (value.length() < 2) return false; + + char first = value.charAt(0); + char last = value.charAt(value.length() - 1); + if (!((first == '{' && last == '}') || (first == '[' && last == ']'))) { + return false; + } + + Deque<Character> stack = new ArrayDeque<>(); + boolean inQuote = false; + + for (int i = 0; i < value.length(); i++) { + char c = value.charAt(i); + + if (c == '"' && !isEscapedQuote(value, i)) { + inQuote = !inQuote; + continue; + } + if (inQuote) { + continue; + } + + if (c == '{' || c == '[') { + stack.push(c); + } else if (c == '}' || c == ']') { + if (stack.isEmpty()) { + return false; + } + char open = stack.pop(); + if ((c == '}' && open != '{') || (c == ']' && open != '[')) { + return false; + } + } + } + + return !inQuote && stack.isEmpty(); + } +} diff --git a/seatunnel-common/src/test/java/org/apache/seatunnel/common/utils/ConfigValueUtilsTest.java b/seatunnel-common/src/test/java/org/apache/seatunnel/common/utils/ConfigValueUtilsTest.java new file mode 100644 index 0000000000..bb4571e7bc --- /dev/null +++ b/seatunnel-common/src/test/java/org/apache/seatunnel/common/utils/ConfigValueUtilsTest.java @@ -0,0 +1,158 @@ +/* + * 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.common.utils; + +import org.apache.seatunnel.shade.com.typesafe.config.ConfigException; + +import org.junit.jupiter.api.Assertions; +import org.junit.jupiter.api.Test; + +import java.util.List; +import java.util.Map; + +import static org.junit.jupiter.api.Assertions.assertDoesNotThrow; +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertFalse; +import static org.junit.jupiter.api.Assertions.assertNotNull; +import static org.junit.jupiter.api.Assertions.assertTrue; + +public class ConfigValueUtilsTest { + + @Test + void testParseValueWithEmptyString() { + Object result = ConfigValueUtils.parseValue(""); + assertNotNull(result); + } + + @Test + void testParseValueWithNull() { + assertDoesNotThrow( + () -> { + ConfigValueUtils.parseValue(null); + }); + } + + @Test + void testParseValueWithBlankString() { + Object result = ConfigValueUtils.parseValue(" "); + assertNotNull(result); + } + + @Test + void testParseValueWithNormalString() { + Object result = ConfigValueUtils.parseValue("\"hello\"").unwrapped(); + assertEquals("hello", result); + } + + @Test + void testParseValueWithNumber() { + Object result = ConfigValueUtils.parseValue("235.60").unwrapped(); + assertEquals("235.60", result); + } + + @Test + void testParseValueWithNumber2() { + Object result = ConfigValueUtils.parseValue("007").unwrapped(); + assertEquals("007", result); + } + + @Test + void testParseValueWithNumber3() { + Object result = ConfigValueUtils.parseValue("1e3").unwrapped(); + assertEquals("1e3", result); + } + + @Test + void testParseValueWithBoolean() { + Object result = ConfigValueUtils.parseValue("true").unwrapped(); + assertEquals("true", result); + } + + @Test + void testParseValueWithArray() { + Object result = ConfigValueUtils.parseValue("[\"a\",\"b\"]"); + assertTrue(result instanceof List); + } + + @Test + void testParseValueWithJsonObject() { + Object result = ConfigValueUtils.parseValue("{\"k1\":\"v1\",\"k2\":\"v2\"}"); + assertTrue(result instanceof Map); + } + + @Test + void testParseValueWithWrongJsonFormat() { + // right json format should end with '}]',not ']}' + String value = + "[{\"table_path\":\"testdb.t_*\",\"use_regex\":\"true\"},{\"table_path\":\"testdb.tt\"]}"; + + Object result = ConfigValueUtils.parseValue(value).unwrapped(); + assertTrue(result instanceof String); + assertFalse(result instanceof Map); + } + + @Test + void testParseValueAsStringWithUnclosedJsonFormat() { + // right json format should end with '}' + String value = "{\"k1\":\"v1\",\"k2\":\"v2\""; + Object result = ConfigValueUtils.parseValue(value).unwrapped(); + Assertions.assertInstanceOf(String.class, result); + } + + @Test + void testParseValueAsMapWithCompatibleJsonFormat() { + // right json format should end with '}' + String value = "{\"k1\"=\"v1\",\"k2\":\"v2\"}"; + Object result = ConfigValueUtils.parseValue(value).unwrapped(); + + Assertions.assertEquals(((Map) result).size(), 2); + } + + @Test + void testParseValueThrowsForBalancedNonJson() { + String value = "{k1,k2,[k3,k4]}"; + Assertions.assertThrows(ConfigException.class, () -> ConfigValueUtils.parseValue(value)); + } + + @Test + void testParseValueAsStringWithJsonStringFormatLike() { + String value = "\"{k1,k2,[k3,k4]}\""; + Object result = ConfigValueUtils.parseValue(value).unwrapped(); + Assertions.assertEquals(result, "{k1,k2,[k3,k4]}"); + } + + @Test + void testParseValueWithNestedJson() { + String assert_filed_rules = + "[" + + "{" + + "\"field_name\":\"movie_id\"," + + "\"field_type\":\"bigint\"," + + "\"field_value\":[{\"rule_type\":\"NOT_NULL\"}]" + + "}," + + "{" + + "\"field_name\":\"unix_time\"," + + "\"field_type\":\"bigint\"," + + "\"field_value\":[{\"rule_type\":\"NOT_NULL\"}]" + + "}" + + "]"; + + Object result = ConfigValueUtils.parseValue(assert_filed_rules).unwrapped(); + Assertions.assertInstanceOf(List.class, result); + } +} diff --git a/seatunnel-core/seatunnel-core-starter/src/main/java/org/apache/seatunnel/core/starter/SeaTunnel.java b/seatunnel-core/seatunnel-core-starter/src/main/java/org/apache/seatunnel/core/starter/SeaTunnel.java index e52b6a81e9..363a409f2a 100644 --- a/seatunnel-core/seatunnel-core-starter/src/main/java/org/apache/seatunnel/core/starter/SeaTunnel.java +++ b/seatunnel-core/seatunnel-core-starter/src/main/java/org/apache/seatunnel/core/starter/SeaTunnel.java @@ -17,12 +17,14 @@ package org.apache.seatunnel.core.starter; +import org.apache.seatunnel.shade.com.typesafe.config.ConfigException; import org.apache.seatunnel.shade.org.apache.commons.lang3.exception.ExceptionUtils; import org.apache.seatunnel.common.config.ConfigRuntimeException; import org.apache.seatunnel.core.starter.command.Command; import org.apache.seatunnel.core.starter.command.CommandArgs; import org.apache.seatunnel.core.starter.exception.CommandException; +import org.apache.seatunnel.core.starter.exception.ConfigCheckException; import lombok.extern.slf4j.Slf4j; @@ -38,7 +40,7 @@ public class SeaTunnel { public static <T extends CommandArgs> void run(Command<T> command) throws CommandException { try { command.execute(); - } catch (ConfigRuntimeException e) { + } catch (ConfigRuntimeException | ConfigException | ConfigCheckException e) { showConfigError(e); throw e; } catch (Exception e) { diff --git a/seatunnel-core/seatunnel-core-starter/src/main/java/org/apache/seatunnel/core/starter/command/ConfEncryptCommand.java b/seatunnel-core/seatunnel-core-starter/src/main/java/org/apache/seatunnel/core/starter/command/ConfEncryptCommand.java index f94b235599..1a04e68d11 100644 --- a/seatunnel-core/seatunnel-core-starter/src/main/java/org/apache/seatunnel/core/starter/command/ConfEncryptCommand.java +++ b/seatunnel-core/seatunnel-core-starter/src/main/java/org/apache/seatunnel/core/starter/command/ConfEncryptCommand.java @@ -30,8 +30,11 @@ import lombok.extern.slf4j.Slf4j; import java.nio.file.Path; import java.nio.file.Paths; -import java.util.Objects; +import java.util.List; +import java.util.Map; +import static org.apache.seatunnel.core.starter.utils.ConfigBuilder.extractUserVariables; +import static org.apache.seatunnel.core.starter.utils.ConfigBuilder.substituteUserConfig; import static org.apache.seatunnel.core.starter.utils.FileUtils.checkConfigExist; @Slf4j @@ -39,6 +42,8 @@ public class ConfEncryptCommand implements Command<AbstractCommandArgs> { private final AbstractCommandArgs abstractCommandArgs; + private Config encryptConfig; + public ConfEncryptCommand(AbstractCommandArgs abstractCommandArgs) { this.abstractCommandArgs = abstractCommandArgs; } @@ -55,22 +60,20 @@ public class ConfEncryptCommand implements Command<AbstractCommandArgs> { Config config = ConfigFactory.parseFile(configPath.toFile()) .resolve(ConfigResolveOptions.defaults().setAllowUnresolved(true)); - if (abstractCommandArgs.getVariables() != null) { - abstractCommandArgs.getVariables().stream() - .filter(Objects::nonNull) - .map(variable -> variable.split("=", 2)) - .filter(pair -> pair.length == 2) - .forEach(pair -> System.setProperty(pair[0], pair[1])); - config = - config.resolveWith( - ConfigFactory.systemProperties(), - ConfigResolveOptions.defaults().setAllowUnresolved(true)); - } - Config encryptConfig = ConfigShadeUtils.encryptConfig(config); + List<String> variables = abstractCommandArgs.getVariables(); + + Map<String, String> userConfigMap = extractUserVariables(variables); + Config resolvedConfig = substituteUserConfig(config, userConfigMap); + + encryptConfig = ConfigShadeUtils.encryptConfig(resolvedConfig); log.info( "Encrypt config: \n{}", encryptConfig .root() .render(ConfigRenderOptions.defaults().setOriginComments(false))); } + + Config getEncryptedConfig() { + return this.encryptConfig; + } } diff --git a/seatunnel-core/seatunnel-core-starter/src/main/java/org/apache/seatunnel/core/starter/command/ParameterSplitter.java b/seatunnel-core/seatunnel-core-starter/src/main/java/org/apache/seatunnel/core/starter/command/ParameterSplitter.java index 57d50b0d4a..170e7a8fa2 100644 --- a/seatunnel-core/seatunnel-core-starter/src/main/java/org/apache/seatunnel/core/starter/command/ParameterSplitter.java +++ b/seatunnel-core/seatunnel-core-starter/src/main/java/org/apache/seatunnel/core/starter/command/ParameterSplitter.java @@ -16,6 +16,8 @@ */ package org.apache.seatunnel.core.starter.command; +import org.apache.seatunnel.common.utils.ConfigValueUtils; + import com.beust.jcommander.converters.IParameterSplitter; import java.util.ArrayList; @@ -25,23 +27,38 @@ public class ParameterSplitter implements IParameterSplitter { @Override public List<String> split(String value) { - List<String> result = new ArrayList<>(); StringBuilder currentToken = new StringBuilder(); - boolean insideBrackets = false; boolean insideQuotes = false; + int braceDepth = 0; + int bracketDepth = 0; + + for (int i = 0; i < value.length(); i++) { + char c = value.charAt(i); - for (char c : value.toCharArray()) { + if (c == '"') { + if (ConfigValueUtils.isEscapedQuote(value, i)) { + currentToken.append(c); + continue; + } + insideQuotes = ConfigValueUtils.updateQuoteState(value, i, insideQuotes); + currentToken.append(c); + continue; + } - if (c == '[') { - insideBrackets = true; - } else if (c == ']') { - insideBrackets = false; - } else if (c == '"') { - insideQuotes = !insideQuotes; + if (!insideQuotes) { + if (c == '{') { + braceDepth++; + } else if (c == '}' && braceDepth > 0) { + braceDepth--; + } else if (c == '[') { + bracketDepth++; + } else if (c == ']' && bracketDepth > 0) { + bracketDepth--; + } } - if (c == ',' && !insideQuotes && !insideBrackets) { + if (c == ',' && !insideQuotes && braceDepth == 0 && bracketDepth == 0) { result.add(currentToken.toString().trim()); currentToken = new StringBuilder(); } else { diff --git a/seatunnel-core/seatunnel-core-starter/src/main/java/org/apache/seatunnel/core/starter/utils/ConfigBuilder.java b/seatunnel-core/seatunnel-core-starter/src/main/java/org/apache/seatunnel/core/starter/utils/ConfigBuilder.java index 7d409c60e0..754695826d 100644 --- a/seatunnel-core/seatunnel-core-starter/src/main/java/org/apache/seatunnel/core/starter/utils/ConfigBuilder.java +++ b/seatunnel-core/seatunnel-core-starter/src/main/java/org/apache/seatunnel/core/starter/utils/ConfigBuilder.java @@ -27,6 +27,7 @@ import org.apache.seatunnel.shade.com.typesafe.config.impl.Parseable; import org.apache.seatunnel.api.configuration.ConfigAdapter; import org.apache.seatunnel.api.sink.TablePlaceholder; +import org.apache.seatunnel.common.utils.ConfigValueUtils; import org.apache.seatunnel.common.utils.JsonUtils; import org.apache.seatunnel.common.utils.ParserException; import org.apache.seatunnel.core.starter.exception.ConfigCheckException; @@ -250,66 +251,126 @@ public class ConfigBuilder { } private static Config backfillUserVariables(Config config, List<String> variables) { - if (variables != null) { - variables.stream() - .filter(Objects::nonNull) - .map(variable -> variable.split("=", 2)) - .filter(pair -> pair.length == 2) - .peek( - pair -> { - if (TablePlaceholder.isSystemPlaceholder(pair[0])) { - throw new ConfigCheckException( - "System placeholders cannot be used. Incorrect config parameter: " - + pair[0]); - } - }) - .forEach(pair -> System.setProperty(pair[0], pair[1])); - Config systemConfig = - Parseable.newProperties( - System.getProperties(), - ConfigParseOptions.defaults() - .setOriginDescription("system properties")) - .parse() - .toConfig(); - - Config resolvedConfig = - config.resolveWith( - systemConfig, ConfigResolveOptions.defaults().setAllowUnresolved(true)); - - Map<String, Object> configMap = resolvedConfig.root().unwrapped(); - - configMap.forEach( - (key, value) -> { - if (value instanceof Map) { - processVariablesMap((Map<String, Object>) value); - } else if (value instanceof List) { - ((List<Map<String, Object>>) value) - .forEach(map -> processVariablesMap(map)); - } - }); - - return ConfigFactory.parseString( - JsonUtils.toJsonString(configMap), - ConfigParseOptions.defaults().setSyntax(ConfigSyntax.JSON)) - .resolve(ConfigResolveOptions.defaults().setAllowUnresolved(true)); + Map<String, String> userConfigMap = extractUserVariables(variables); + Config resolvedConfig = substituteUserConfig(config, userConfigMap); + + Map<String, Object> resolvedConfigMap = resolvedConfig.root().unwrapped(); + + processVariablesMap(resolvedConfigMap, userConfigMap); + + return ConfigFactory.parseMap(resolvedConfigMap); + } + + public static Config substituteUserConfig( + Config originalConfig, Map<String, String> userConfigMap) { + Config userConfig = + ConfigFactory.parseMap( + userConfigMap.entrySet().stream() + .collect( + Collectors.toMap( + Map.Entry::getKey, + entry -> + ConfigValueUtils.parseValue( + entry.getValue()) + .unwrapped()))); + + Config systemConfig = + Parseable.newProperties( + System.getProperties(), + ConfigParseOptions.defaults() + .setOriginDescription("system properties")) + .parse() + .toConfig(); + + // Priority: userConfig > systemProperties + Map<String, Object> mergedMap = new LinkedHashMap<>(systemConfig.root().unwrapped()); + mergedMap.putAll(userConfig.root().unwrapped()); + Config sourceConfig = ConfigFactory.parseMap(mergedMap); + + return originalConfig.resolveWith( + sourceConfig, ConfigResolveOptions.defaults().setAllowUnresolved(true)); + } + + /** + * Extracts user variables into a map by splitting each entry on '='. + * + * <p>Inputs like {@code -i 'k1= , k2=v2'} or {@code -i ' =v1,k2=v2'} have ambiguous semantics, + * so both key and value are trimmed. + * + * <p>Empty values are allowed, but empty keys, duplicate keys, system placeholder keys are + * rejected. + * + * <p>Entries without '=' are silently skipped. + * + * @param variables the {@code -i} variables from CLI + * @return map stores trimmed key and value + */ + public static Map<String, String> extractUserVariables(List<String> variables) { + Map<String, String> userConfigMap = new LinkedHashMap<>(); + + if (variables == null || variables.isEmpty()) { + return userConfigMap; } - return config; + + for (String variable : variables) { + if (variable == null) { + continue; + } + + String[] pair = variable.split("=", 2); + + if (pair.length != 2) { + continue; + } + + String userValueString = pair[1].trim(); + + String userKey = getUserKey(pair, userConfigMap); + + userConfigMap.put(userKey, userValueString); + } + + return userConfigMap; + } + + private static String getUserKey(String[] pair, Map<String, String> userConfigMap) { + String userKey = pair[0].trim(); + + if (userKey.isEmpty()) { + throw new ConfigCheckException("Invalid -i variable: empty key: '" + pair[0] + "'"); + } + + if (TablePlaceholder.isSystemPlaceholder(userKey)) { + throw new ConfigCheckException( + "System placeholders cannot be used. Incorrect config parameter: " + userKey); + } + + if (userConfigMap.containsKey(userKey)) { + throw new ConfigCheckException( + "Duplicate -i variable key detected: '" + + userKey + + "'. Please remove duplicate keys."); + } + return userKey; } - private static void processVariablesMap(Map<String, Object> mapValue) { + private static void processVariablesMap( + Map<String, Object> mapValue, Map<String, String> userConfigMap) { mapValue.forEach( (innerKey, innerValue) -> { if (innerValue instanceof Map) { - processVariablesMap((Map<String, Object>) innerValue); + processVariablesMap((Map<String, Object>) innerValue, userConfigMap); } else if (innerValue instanceof List) { - mapValue.put(innerKey, processVariablesList((List<?>) innerValue)); + mapValue.put( + innerKey, + processVariablesList((List<?>) innerValue, userConfigMap)); } else { - processVariable(innerKey, innerValue, mapValue); + processVariable(innerKey, innerValue, mapValue, userConfigMap); } }); } - private static List<?> processVariablesList(List<?> list) { + private static List<?> processVariablesList(List<?> list, Map<String, String> userConfigMap) { return list.stream() .map( variable -> { @@ -319,17 +380,19 @@ public class ConfigBuilder { .reduce( variableString, (result, placeholder) -> { + String value = + userConfigMap.containsKey(placeholder) + ? userConfigMap.get(placeholder) + : System.getProperty( + placeholder); return replacePlaceholders( - result, - placeholder, - System.getProperty(placeholder), - null); + result, placeholder, value, null); }); } else if (variable instanceof Map) { - processVariablesMap((Map<String, Object>) variable); + processVariablesMap((Map<String, Object>) variable, userConfigMap); return variable; } else if (variable instanceof List) { - return processVariablesList((List<?>) variable); + return processVariablesList((List<?>) variable, userConfigMap); } return variable; }) @@ -337,7 +400,10 @@ public class ConfigBuilder { } private static void processVariable( - String variableKey, Object variableValue, Map<String, Object> parentMap) { + String variableKey, + Object variableValue, + Map<String, Object> parentMap, + Map<String, String> userConfigMap) { if (Objects.isNull(variableValue)) { return; } @@ -345,10 +411,12 @@ public class ConfigBuilder { List<String> placeholders = extractPlaceholder(variableString); for (String placeholder : placeholders) { - String replacedValue = - replacePlaceholders( - variableString, placeholder, System.getProperty(placeholder), null); - variableString = replacedValue; + String value = + userConfigMap.containsKey(placeholder) + ? userConfigMap.get(placeholder) + : System.getProperty(placeholder); + + variableString = replacePlaceholders(variableString, placeholder, value, null); } if (!placeholders.isEmpty()) { diff --git a/seatunnel-core/seatunnel-core-starter/src/test/java/org/apache/seatunnel/core/starter/command/ConfEncryptCommandTest.java b/seatunnel-core/seatunnel-core-starter/src/test/java/org/apache/seatunnel/core/starter/command/ConfEncryptCommandTest.java index d3b6837e25..a50c232339 100644 --- a/seatunnel-core/seatunnel-core-starter/src/test/java/org/apache/seatunnel/core/starter/command/ConfEncryptCommandTest.java +++ b/seatunnel-core/seatunnel-core-starter/src/test/java/org/apache/seatunnel/core/starter/command/ConfEncryptCommandTest.java @@ -17,6 +17,9 @@ package org.apache.seatunnel.core.starter.command; +import org.apache.seatunnel.shade.com.typesafe.config.Config; +import org.apache.seatunnel.shade.com.typesafe.config.ConfigObject; + import org.apache.seatunnel.common.config.DeployMode; import org.junit.jupiter.api.Assertions; @@ -26,6 +29,9 @@ import java.net.URISyntaxException; import java.net.URL; import java.nio.file.Path; import java.nio.file.Paths; +import java.util.ArrayList; +import java.util.List; +import java.util.Map; public class ConfEncryptCommandTest { @@ -45,6 +51,63 @@ public class ConfEncryptCommandTest { confEncryptCommand.execute(); } + @Test + public void testEncryptWithJsonUserVariables() throws Exception { + TestCommandArgs testCommandArgs = new TestCommandArgs(); + Path filePath = getFilePath("/json_params.conf"); + testCommandArgs.setEncrypt(true); + testCommandArgs.setConfigFile(filePath.toString()); + + List<String> variables = new ArrayList<>(); + variables.add("mysql_password=123456"); + variables.add("mysql_props={\"useSSL\":\"false\",\"allowPublicKeyRetrieval\":\"true\"}"); + variables.add( + "mysql_tables=[" + + "{\"table_path\":\"test.ml_*\",\"use_regex\":\"true\"}," + + "{\"table_path\":\"test.ratings\"}" + + "]"); + + testCommandArgs.setVariables(variables); + + ConfEncryptCommand command = new ConfEncryptCommand(testCommandArgs); + command.execute(); + Config encryptedConfig = command.getEncryptedConfig(); + + List<? extends ConfigObject> sourceConfigs = encryptedConfig.getObjectList("source"); + for (ConfigObject configObject : sourceConfigs) { + Config sourceConfig = configObject.toConfig(); + Assertions.assertTrue( + sourceConfig.hasPath("password"), + "password key should exist in encrypted config"); + + Assertions.assertTrue( + sourceConfig.hasPath("properties"), + "properties key should exist in encrypted config"); + + Assertions.assertTrue( + sourceConfig.hasPath("table_list"), + "table_list key should exist in encrypted config"); + + String mysql_password = sourceConfig.getString("password"); + Assertions.assertNotEquals(mysql_password, "123456"); + + Map<String, Object> mysqlProperties = sourceConfig.getObject("properties").unwrapped(); + Assertions.assertTrue( + mysqlProperties.containsKey("allowPublicKeyRetrieval"), + "properties should contain key: 'allowPublicKeyRetrieval'"); + + List<? extends ConfigObject> tableList = sourceConfig.getObjectList("table_list"); + boolean useRegex = + tableList.stream() + .map(tableObject -> tableObject.toConfig().getBoolean("use_regex")) + .findFirst() + .get(); + + Assertions.assertTrue( + useRegex, "useRegex should contain replaced placeholder value: " + useRegex); + } + } + public static class TestCommandArgs extends AbstractCommandArgs { @Override diff --git a/seatunnel-core/seatunnel-core-starter/src/test/java/org/apache/seatunnel/core/starter/command/ParameterSplitterTest.java b/seatunnel-core/seatunnel-core-starter/src/test/java/org/apache/seatunnel/core/starter/command/ParameterSplitterTest.java new file mode 100644 index 0000000000..c70dbaee88 --- /dev/null +++ b/seatunnel-core/seatunnel-core-starter/src/test/java/org/apache/seatunnel/core/starter/command/ParameterSplitterTest.java @@ -0,0 +1,181 @@ +/* + * 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.core.starter.command; + +import org.junit.jupiter.api.Test; + +import java.util.List; + +import static org.junit.jupiter.api.Assertions.assertArrayEquals; +import static org.junit.jupiter.api.Assertions.assertEquals; + +public class ParameterSplitterTest { + + private ParameterSplitter parameterSplitter = new ParameterSplitter(); + + @Test + void testQuotedBracesAndCommas() { + String input = "props={\"a\":\"}\",\"b\":\"x\"},other=1"; + String[] expected = {"props={\"a\":\"}\",\"b\":\"x\"}", "other=1"}; + assertArrayEquals(expected, parameterSplitter.split(input).toArray()); + } + + @Test + void testNestedBracesAndBrackets() { + String input = "{b={c=1}}"; + String[] expected = {"{b={c=1}}"}; + assertArrayEquals(expected, parameterSplitter.split(input).toArray()); + } + + @Test + void testEscapedQuotes() { + String input = + "json={\"key\":\"value with \\\"${var}\\\" inside\",\"path\":\"D:\\data\\file\\\\\"},next=2"; + String[] expected = { + "json={\"key\":\"value with \\\"${var}\\\" inside\",\"path\":\"D:\\data\\file\\\\\"}", + "next=2" + }; + assertArrayEquals(expected, parameterSplitter.split(input).toArray()); + } + + @Test + public void testEscapedQuoteFollowedByComma() { + String input = "{\"note\":\"he said,\\\"stop,\\\", then left\"}"; + + assertEquals(input, parameterSplitter.split(input).toArray()[0]); + } + + @Test + public void testEscapedQuoteFollowedByBrace() { + String input = "{\"note\":\"he said \\\"}stop{\\\"} then left\"}"; + + assertEquals(input, parameterSplitter.split(input).toArray()[0]); + } + + @Test + public void testEscapedQuoteFollowedByBracket() { + String input = "{\"note\":\"he said \\\"]stop]\\\"[ then left\"}"; + + assertEquals(input, parameterSplitter.split(input).toArray()[0]); + } + + @Test + public void testEscapedQuoteFollowedByEqual() { + String input = "{\"note\":\"he said \\\"=stop=\\\" then left\"}"; + + assertEquals(input, parameterSplitter.split(input).toArray()[0]); + } + + @Test + public void testEscapedQuoteFollowedByColon() { + String input = "{\"note\":\"he said: \\\":stop:\\\" then left\"}"; + + assertEquals(input, parameterSplitter.split(input).toArray()[0]); + } + + @Test + void testArrayWithCommas() { + String input = "arr=[1,2,3],other=4"; + String[] expected = {"arr=[1,2,3]", "other=4"}; + assertArrayEquals(expected, parameterSplitter.split(input).toArray()); + } + + @Test + void testArrayAndObjectMixed() { + String input = + "json_data={\"a\":\"}\",\"b\":\"{xyz}\",\"c\":[{\"k1\":\"v1\"},{\"k2\":\"v2\",\"k3\":\"v3\"}],\"d\":{\"list\":[{\"k1\":\"v1\"},{\"k2\":\"v2\",\"k3\":\"v3\"}]}},extra=5"; + String[] expected = { + "json_data={\"a\":\"}\",\"b\":\"{xyz}\",\"c\":[{\"k1\":\"v1\"},{\"k2\":\"v2\",\"k3\":\"v3\"}],\"d\":{\"list\":[{\"k1\":\"v1\"},{\"k2\":\"v2\",\"k3\":\"v3\"}]}}", + "extra=5" + }; + assertArrayEquals(expected, parameterSplitter.split(input).toArray()); + } + + @Test + void testPlain() { + String input = "a=1,b=2,c=3"; + String[] expected = {"a=1", "b=2", "c=3"}; + assertArrayEquals(expected, parameterSplitter.split(input).toArray()); + } + + @Test + void testSingleBalancedBracesNotSplit() { + String value = "{abcd,1234}"; + List<String> result = parameterSplitter.split(value); + assertEquals(result.toArray()[0], value); + } + + @Test + void testSplitWithMultiBalancedBraces() { + String value = "{abcd,1234},{21434,7786}"; + List<String> result = parameterSplitter.split(value); + + assertEquals(result.size(), 2); + } + + @Test + void testSplitWithQuotedMultiBalancedBracesWillBeReserved() { + String value = "\"{abcd,1234},{21434,7786}\""; + List<String> result = parameterSplitter.split(value); + + assertEquals(result.toArray()[0], value); + } + + @Test + void testSplitWithSingleBalancedBrackets() { + String value = "[a-z]+"; + List<String> result = parameterSplitter.split(value); + assertEquals(value, result.get(0).toString()); + } + + @Test + void testSplitWithMultiBalancedBrackets() { + String value = "[dbo].[orders]"; + List<String> result = parameterSplitter.split(value); + assertEquals(value, result.get(0).toString()); + } + + @Test + void testSplitWithUnmatchedVariablesWillBeReserved() { + String value = "a=1],b={2"; + List<String> result = parameterSplitter.split(value); + String[] splitted = value.split(","); + assertArrayEquals(splitted, result.toArray()); + } + + @Test + void testSplitWithUnmatchedBrackets() { + String value = "[a,\"b]\",c]"; + List<String> result = parameterSplitter.split(value); + assertEquals(value, result.get(0).toString()); + } + + @Test + void testQuotedUnmatchedBracketsWillBeReservedWithQuotes() { + String value = "\"[a,b],c]\""; + List<String> result = parameterSplitter.split(value); + assertEquals(value, result.get(0).toString()); + } + + @Test + void testSplitWithMultiUnmatchedBrackets() { + String value = "[[[a,b],c][]"; + List<String> result = parameterSplitter.split(value); + assertEquals(value, result.get(0).toString()); + } +} diff --git a/seatunnel-core/seatunnel-core-starter/src/test/java/org/apache/seatunnel/core/starter/utils/ConfigBuilderTest.java b/seatunnel-core/seatunnel-core-starter/src/test/java/org/apache/seatunnel/core/starter/utils/ConfigBuilderTest.java index 380e7c840a..9a86c43b98 100644 --- a/seatunnel-core/seatunnel-core-starter/src/test/java/org/apache/seatunnel/core/starter/utils/ConfigBuilderTest.java +++ b/seatunnel-core/seatunnel-core-starter/src/test/java/org/apache/seatunnel/core/starter/utils/ConfigBuilderTest.java @@ -17,6 +17,8 @@ package org.apache.seatunnel.core.starter.utils; +import org.apache.seatunnel.core.starter.exception.ConfigCheckException; + import org.junit.jupiter.api.Assertions; import org.junit.jupiter.api.Test; @@ -233,4 +235,30 @@ public class ConfigBuilderTest { Assertions.assertEquals("string", desensitizedFields.get("access_token")); Assertions.assertEquals("string", desensitizedFields.get("user-password")); } + + @Test + void testDuplicateKeyThrowsException() { + List<String> variables = Arrays.asList("jobName=value1", "jobName=value2"); + + Assertions.assertThrows( + ConfigCheckException.class, () -> ConfigBuilder.extractUserVariables(variables)); + } + + @Test + void testUserKeyValueTrimmed() { + List<String> variables = Arrays.asList("k1= , k2=v2".split(",")); + Map<String, String> userConfigMap = ConfigBuilder.extractUserVariables(variables); + + Assertions.assertTrue(userConfigMap.get("k1").isEmpty()); + Assertions.assertTrue(userConfigMap.containsKey("k2")); + Assertions.assertFalse(userConfigMap.containsKey(" k2")); + } + + @Test + void testUserEmptyKeyThrowsConfigCheckException() { + List<String> variables = Arrays.asList(" =v1 , k2=v2".split(",")); + + Assertions.assertThrows( + ConfigCheckException.class, () -> ConfigBuilder.extractUserVariables(variables)); + } } diff --git a/seatunnel-core/seatunnel-core-starter/src/test/java/org/apache/seatunnel/core/starter/utils/ConfigShadeTest.java b/seatunnel-core/seatunnel-core-starter/src/test/java/org/apache/seatunnel/core/starter/utils/ConfigShadeTest.java index 8e84186a84..a463ad50cc 100644 --- a/seatunnel-core/seatunnel-core-starter/src/test/java/org/apache/seatunnel/core/starter/utils/ConfigShadeTest.java +++ b/seatunnel-core/seatunnel-core-starter/src/test/java/org/apache/seatunnel/core/starter/utils/ConfigShadeTest.java @@ -19,6 +19,7 @@ package org.apache.seatunnel.core.starter.utils; import org.apache.seatunnel.shade.com.fasterxml.jackson.databind.node.ObjectNode; import org.apache.seatunnel.shade.com.typesafe.config.Config; +import org.apache.seatunnel.shade.com.typesafe.config.ConfigException; import org.apache.seatunnel.shade.com.typesafe.config.ConfigFactory; import org.apache.seatunnel.shade.com.typesafe.config.ConfigObject; import org.apache.seatunnel.shade.com.typesafe.config.ConfigResolveOptions; @@ -27,7 +28,9 @@ import org.apache.seatunnel.api.configuration.ConfigShade; import org.apache.seatunnel.common.utils.JsonUtils; import org.apache.seatunnel.core.starter.exception.ConfigCheckException; +import org.junit.jupiter.api.AfterEach; import org.junit.jupiter.api.Assertions; +import org.junit.jupiter.api.BeforeEach; import org.junit.jupiter.api.Test; import org.junit.jupiter.params.ParameterizedTest; import org.junit.jupiter.params.provider.CsvSource; @@ -59,6 +62,16 @@ public class ConfigShadeTest { private static final String ACCESS_KEY = "access_key"; private static final String SECRET_KEY = "secret_key"; + @BeforeEach + void setUp() { + ConfigFactory.invalidateCaches(); + } + + @AfterEach + void tearDown() { + ConfigFactory.invalidateCaches(); + } + @Test public void testParseConfig() throws URISyntaxException { URL resource = ConfigShadeTest.class.getResource("/config.shade.conf"); @@ -423,6 +436,72 @@ public class ConfigShadeTest { } } + @Test + public void testVariableReplacementWithJsonParams() throws URISyntaxException { + List<String> variables = new ArrayList<>(); + variables.add( + "mysql_properties={" + + "\"useSSL\":\"false\"," + + "\"allowPublicKeyRetrieval\":\"true\"}"); + variables.add( + "mysql_tables=[{\"table_path\":\"test.t_*\",\"use_regex\":\"true\"},{\"table_path\":\"test.tt\"}]"); + + URL resource = ConfigShadeTest.class.getResource("/config_json_variables.conf"); + Assertions.assertNotNull(resource); + Config config = ConfigBuilder.of(Paths.get(resource.toURI()), variables); + + List<? extends ConfigObject> sourceConfigs = config.getObjectList("source"); + for (ConfigObject configObject : sourceConfigs) { + Config sourceConfig = configObject.toConfig(); + + if (sourceConfig.hasPath("properties")) { + Map<String, Object> mysqlProperties = + sourceConfig.getObject("properties").unwrapped(); + Assertions.assertTrue( + mysqlProperties.containsKey("allowPublicKeyRetrieval"), + "properties should contain key: 'allowPublicKeyRetrieval'"); + + List<? extends ConfigObject> tableList = sourceConfig.getObjectList("table_list"); + boolean useRegex = + tableList.stream() + .map(tableObject -> tableObject.toConfig().getBoolean("use_regex")) + .findFirst() + .get(); + + Assertions.assertTrue( + useRegex, + "useRegex should contain replaced placeholder value: " + useRegex); + } + } + } + + @Test + public void testVariableReplacementWithWrongJsonFormatParams() throws URISyntaxException { + List<String> variables = new ArrayList<>(); + // missing closing brace,will be treated as String and throw ConfigException when calling + // getObject() as a map type + variables.add( + "mysql_properties={" + + "\"useSSL\":\"false\"," + + "\"allowPublicKeyRetrieval\":\"true\""); + variables.add( + "mysql_tables=[{\"table_path\":\"test.t_*\",\"use_regex\":\"true\"},{\"table_path\":\"test.tt\"}]"); + + URL resource = ConfigShadeTest.class.getResource("/config_json_variables.conf"); + Assertions.assertNotNull(resource); + Config config = ConfigBuilder.of(Paths.get(resource.toURI()), variables); + + List<? extends ConfigObject> sourceConfigs = config.getObjectList("source"); + for (ConfigObject configObject : sourceConfigs) { + Config sourceConfig = configObject.toConfig(); + + if (sourceConfig.hasPath("properties")) { + Assertions.assertThrows( + ConfigException.class, () -> sourceConfig.getObject("properties")); + } + } + } + @Test public void testDecryptAndEncrypt() { String encryptUsername = ConfigShadeUtils.encryptOption("base64", USERNAME); diff --git a/seatunnel-core/seatunnel-core-starter/src/test/resources/config_json_variables.conf b/seatunnel-core/seatunnel-core-starter/src/test/resources/config_json_variables.conf new file mode 100644 index 0000000000..b5585939f1 --- /dev/null +++ b/seatunnel-core/seatunnel-core-starter/src/test/resources/config_json_variables.conf @@ -0,0 +1,41 @@ +# +# 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. +# + +env { + job.mode = "BATCH" + job.name = "seatunnel json variable test job" + parallelism = 1 +} + +source { + Jdbc { + plugin_output = "source_result" + url = "jdbc:mysql://localhost:3306/test" + driver = "com.mysql.cj.jdbc.Driver" + properties=${mysql_properties} + connection_check_timeout_sec = 100 + user = root + password = "password" + table_list = ${mysql_tables} + } +} + +sink { + Console { + plugin_input = "source_result" + } +} diff --git a/seatunnel-core/seatunnel-core-starter/src/test/resources/json_params.conf b/seatunnel-core/seatunnel-core-starter/src/test/resources/json_params.conf new file mode 100644 index 0000000000..6a55549a5f --- /dev/null +++ b/seatunnel-core/seatunnel-core-starter/src/test/resources/json_params.conf @@ -0,0 +1,43 @@ +# +# 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. +# +env { + parallelism = 1 + job.mode = "BATCH" + shade.identifier = "base64" +} + +source { + Jdbc { + driver = "com.mysql.cj.jdbc.Driver" + url = "jdbc:mysql://localhost:3306/test" + properties =${mysql_props} + username = root + password = ${mysql_password} + table_list = ${mysql_tables} + plugin_output = "mysql_source" + } +} + +transform { + +} + +sink { + Console { + plugin_input=mysql_source + } +} \ No newline at end of file diff --git a/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-jdbc-e2e/connector-jdbc-e2e-part-1/src/test/java/org/apache/seatunnel/connectors/seatunnel/jdbc/JdbcMysqlJsonParamIT.java b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-jdbc-e2e/connector-jdbc-e2e-part-1/src/test/java/org/apache/seatunnel/connectors/seatunnel/jdbc/JdbcMysqlJsonParamIT.java new file mode 100644 index 0000000000..b47801ae7c --- /dev/null +++ b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-jdbc-e2e/connector-jdbc-e2e-part-1/src/test/java/org/apache/seatunnel/connectors/seatunnel/jdbc/JdbcMysqlJsonParamIT.java @@ -0,0 +1,270 @@ +/* + * 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.jdbc; + +import org.apache.seatunnel.e2e.common.TestResource; +import org.apache.seatunnel.e2e.common.TestSuiteBase; +import org.apache.seatunnel.e2e.common.container.ContainerExtendedFactory; +import org.apache.seatunnel.e2e.common.container.EngineType; +import org.apache.seatunnel.e2e.common.container.TestContainer; +import org.apache.seatunnel.e2e.common.junit.DisabledOnContainer; +import org.apache.seatunnel.e2e.common.junit.TestContainerExtension; +import org.apache.seatunnel.e2e.common.util.ContainerUtil; + +import org.junit.jupiter.api.AfterAll; +import org.junit.jupiter.api.Assertions; +import org.junit.jupiter.api.BeforeAll; +import org.junit.jupiter.api.TestTemplate; +import org.testcontainers.containers.Container; +import org.testcontainers.containers.MySQLContainer; +import org.testcontainers.containers.output.Slf4jLogConsumer; +import org.testcontainers.lifecycle.Startables; +import org.testcontainers.utility.DockerImageName; +import org.testcontainers.utility.DockerLoggerFactory; +import org.testcontainers.utility.MountableFile; + +import lombok.extern.slf4j.Slf4j; + +import java.io.IOException; +import java.sql.Connection; +import java.sql.DriverManager; +import java.sql.SQLException; +import java.sql.Statement; +import java.util.ArrayList; +import java.util.Arrays; +import java.util.List; +import java.util.concurrent.TimeUnit; +import java.util.stream.Stream; + +import static org.awaitility.Awaitility.given; + +/** + * E2E test verifying that json params via cli are correctly parsed by the ParameterSplitter and + * ConfigBuilder + */ +@Slf4j +@DisabledOnContainer( + value = {}, + type = {EngineType.FLINK, EngineType.SPARK}) +public class JdbcMysqlJsonParamIT extends TestSuiteBase implements TestResource { + + private static final String MYSQL_IMAGE = "mysql:8.0"; + private static final String MYSQL_HOST = "mysql_json_param_e2e"; + private static final String MYSQL_DATABASE = "json_test"; + private static final String MYSQL_USER = "root"; + private static final String MYSQL_PASSWORD = "Abc!@#135_seatunnel"; + + private static final List<String> MULTI_TABLE_DDL = + Arrays.asList( + "drop table if exists ml_tags;\n", + "create table ml_tags(" + + "user_id bigint," + + "movie_id bigint," + + "tag varchar(400)," + + "unix_time bigint" + + ");\n", + "INSERT INTO ml_tags(user_id, movie_id, tag, unix_time) VALUES" + + "(336, 1, 'pixar', 1139045764)," + + "(62, 2, 'fantasy', 1528843929)," + + "(289, 3, 'moldy', 1143424860);", + "drop table if exists ratings;\n", + "create table ratings(" + + "user_id bigint," + + "movie_id bigint," + + "rating float," + + "unix_time bigint" + + ");\n", + "INSERT INTO ratings(user_id, movie_id, rating, unix_time) VALUES" + + "(1, 1, 4.0, 964982703)," + + "(1, 3, 4.0, 964981247);"); + + private static final String MYSQL_DRIVER_URL = + "https://repo1.maven.org/maven2/com/mysql/mysql-connector-j/8.0.32/mysql-connector-j-8.0.32.jar";; + + private MySQLContainer<?> mysqlContainer; + + @TestContainerExtension + private final ContainerExtendedFactory extendedFactory = + container -> { + Container.ExecResult result = + container.execInContainer( + "bash", + "-c", + "mkdir -p /tmp/seatunnel/plugins/Jdbc/lib" + + " && cd /tmp/seatunnel/plugins/Jdbc/lib" + + " && wget -q " + + MYSQL_DRIVER_URL); + + ContainerUtil.copyConnectorJarToContainer( + container, + "/jdbc_mysql_json_params.conf", + "seatunnel-connectors-v2", + "connector-", + "seatunnel", + "/tmp/seatunnel"); + + container.copyFileToContainer( + MountableFile.forClasspathResource("jdbc_mysql_json_params.conf"), + "/tmp/jdbc_mysql_json_params.conf"); + + container.copyFileToContainer( + MountableFile.forClasspathResource("jdbc_mysql_plugin_json_params.conf"), + "/tmp/jdbc_mysql_plugin_json_params.conf"); + + Assertions.assertEquals( + 0, + result.getExitCode(), + "Failed to download MySQL driver: " + result.getStderr()); + }; + + @BeforeAll + @Override + public void startUp() throws Exception { + mysqlContainer = + new MySQLContainer<>(DockerImageName.parse(MYSQL_IMAGE)) + .withDatabaseName(MYSQL_DATABASE) + .withUsername(MYSQL_USER) + .withPassword(MYSQL_PASSWORD) + .withNetwork(NETWORK) + .withNetworkAliases(MYSQL_HOST) + .withLogConsumer( + new Slf4jLogConsumer(DockerLoggerFactory.getLogger(MYSQL_IMAGE))); + + Startables.deepStart(Stream.of(mysqlContainer)).join(); + + given().ignoreExceptions() + .await() + .atMost(60, TimeUnit.SECONDS) + .untilAsserted(() -> initMysqlData()); + log.info("MySQL container started and test data initialised."); + } + + @AfterAll + @Override + public void tearDown() { + if (mysqlContainer != null) { + mysqlContainer.close(); + } + } + + /** + * 1: Verify MySQL properties take effect via CLI JSON params. + * + * <p>2: Verify nested JSON params can be passed via cli and parsed as map or list. + * + * @param container + * @throws IOException + * @throws InterruptedException + */ + @TestTemplate + public void testJsonParams(TestContainer container) throws IOException, InterruptedException { + List<String> variables = new ArrayList<>(); + variables.add("-c /tmp/jdbc_mysql_json_params.conf"); + variables.add( + "-i 'mysql_host=" + + MYSQL_HOST + + ",mysql_port=3306" + + ",mysql_db=" + + MYSQL_DATABASE + + "'"); + variables.add( + "-i mysql_props='{" + + "\"useSSL\":\"false\"," + + "\"allowPublicKeyRetrieval\":\"true\"" + + "}'"); + + variables.add("-i mysql_password=" + MYSQL_PASSWORD); + + variables.add( + "-i table_list=[" + + "'{\"table_path\":\"json_test.ml_*\",\"use_regex\":\"true\"}'," + + "'{\"table_path\":\"json_test.ratings\"}'" + + "]"); + variables.add("-i read_cols=[\"movie_id\",\"unix_time\"]"); + + variables.add( + "-i assert_filed_rules=[" + + "'{" + + "\"field_name\":\"movie_id\"," + + "\"field_type\":\"bigint\"," + + "\"field_value\":['{\"rule_type\":\"NOT_NULL\"}']" + + "}'," + + "'{" + + "\"field_name\":\"unix_time\"," + + "\"field_type\":\"bigint\"," + + "\"field_value\":['{\"rule_type\":\"NOT_NULL\"}']" + + "}'" + + "]"); + + Container.ExecResult result = + container.executeBaseCommand(variables.toArray(new String[0])); + + Assertions.assertEquals( + 0, + result.getExitCode(), + "plain json or nested json or nested array value from -i variables assertion failed:\n" + + result.getStderr()); + } + + @TestTemplate + public void testJsonParamsAtPluginNodeNotSupported(TestContainer container) + throws IOException, InterruptedException { + List<String> variables = new ArrayList<>(); + variables.add("-c /tmp/jdbc_mysql_plugin_json_params.conf"); + variables.add( + "-i 'mysql_host=" + + MYSQL_HOST + + ",mysql_port=3306" + + ",mysql_db=" + + MYSQL_DATABASE + + "'"); + variables.add("-i mysql_password=" + MYSQL_PASSWORD); + variables.add( + "-i table_filter='{\"plugin_input\":\"mysql_source\",\"plugin_output\":\"table_filter\",\"include_fields\":[movie_id,unix_time]}'"); + + Container.ExecResult result = + container.executeBaseCommand(variables.toArray(new String[0])); + + Assertions.assertEquals( + 1, + result.getExitCode(), + "json param for plugin node from -i variables not supported:\n" + + result.getStderr()); + } + + private void initMysqlData() throws Exception { + String jdbcUrl = + String.format( + "jdbc:mysql://%s:%d/%s?useSSL=false&serverTimezone=UTC&allowPublicKeyRetrieval=true&allowMultiQueries=true", + mysqlContainer.getHost(), + mysqlContainer.getFirstMappedPort(), + MYSQL_DATABASE); + try (Connection conn = DriverManager.getConnection(jdbcUrl, MYSQL_USER, MYSQL_PASSWORD); + Statement stmt = conn.createStatement()) { + + MULTI_TABLE_DDL.forEach( + sql -> { + try { + stmt.execute(sql); + } catch (SQLException e) { + throw new RuntimeException(e); + } + }); + } + } +} diff --git a/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-jdbc-e2e/connector-jdbc-e2e-part-1/src/test/resources/jdbc_mysql_json_params.conf b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-jdbc-e2e/connector-jdbc-e2e-part-1/src/test/resources/jdbc_mysql_json_params.conf new file mode 100644 index 0000000000..f650516e75 --- /dev/null +++ b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-jdbc-e2e/connector-jdbc-e2e-part-1/src/test/resources/jdbc_mysql_json_params.conf @@ -0,0 +1,59 @@ +# +# 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. +# + +# Verifies that json params such as properties,table_list and filter config options taking effect via cli + +env { + parallelism = 1 + job.mode = "BATCH" +} + +source { + Jdbc { + driver = "com.mysql.cj.jdbc.Driver" + url = "jdbc:mysql://${mysql_host}:${mysql_port}/${mysql_db}" + properties = ${mysql_props} + username = "${mysql_user:root}" + password = ${mysql_password} + table_list = ${table_list} + plugin_output = "mysql_source" + } +} + +transform { + filter { + plugin_input = mysql_source + plugin_output = table_filter + include_fields = ${read_cols} + } +} + +sink { + Assert { + plugin_input = "table_filter" + rules = { + field_rules = ${assert_filed_rules} + row_rules = [ + { + rule_type = MAX_ROW + rule_value = 3 + } + ] + } + } +} + diff --git a/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-jdbc-e2e/connector-jdbc-e2e-part-1/src/test/resources/jdbc_mysql_plugin_json_params.conf b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-jdbc-e2e/connector-jdbc-e2e-part-1/src/test/resources/jdbc_mysql_plugin_json_params.conf new file mode 100644 index 0000000000..c1947e0bf4 --- /dev/null +++ b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-jdbc-e2e/connector-jdbc-e2e-part-1/src/test/resources/jdbc_mysql_plugin_json_params.conf @@ -0,0 +1,82 @@ +# +# 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. +# + +# Verifies that json params such as properties,table_list and filter config options taking effect via cli + +env { + parallelism = 1 + job.mode = "BATCH" +} + +source { + Jdbc { + driver = "com.mysql.cj.jdbc.Driver" + url = "jdbc:mysql://${mysql_host}:${mysql_port}/${mysql_db}" + properties { + useSSL=false + allowPublicKeyRetrieval=true + } + username = "${mysql_user:root}" + password = ${mysql_password} + table_list = [ + {"table_path":"json_test.ml_*" + "use_regex":"true" + }, + {"table_path":"json_test.ratings"} + ] + plugin_output = "mysql_source" + } +} + +transform { + # NOTE: json value is NOT supported at the plugin-level node via cli(e.g., -i table_filter = """${k:{...}}"""). + filter = ${table_filter} +} + +sink { + Assert { + plugin_input = "table_filter" + rules = { + field_rules = [ + { + field_name = "movie_id" + field_type = bigint + field_value = [ + { + rule_type = NOT_NULL + } + ] + }, + { + field_name = "unix_time" + field_type = bigint + field_value = [ + { + rule_type = NOT_NULL + } + ] + } + ] + row_rules = [ + { + rule_type = MAX_ROW + rule_value = 3 + } + ] + } + } +} \ No newline at end of file
