Merge branch 'apache:dev' into dev

config/plugin_config

@@ -36,7 +36,7 @@ connector-file-ftp
 connector-file-hadoop
 connector-file-local
 connector-file-oss
-connector-file-oss-jindo
+connector-file-jindo-oss
 connector-file-s3
 connector-file-sftp
 connector-google-sheets

seatunnel-config/seatunnel-config-shade/src/main/java/org/apache/seatunnel/shade/com/typesafe/config/impl/PropertiesParser.java

@@ -7,8 +7,7 @@
 import java.io.IOException;
 import java.io.Reader;
 import java.util.ArrayList;
-import java.util.Collections;
-import java.util.Comparator;
+import java.util.Arrays;
 import java.util.LinkedHashMap;
 import java.util.LinkedHashSet;
 import java.util.List;
@@ -58,7 +57,15 @@ private static <K, V> AbstractConfigObject fromEntrySet(
     }
 
     private static <K, V> Map<Path, Object> getPathMap(Set<Map.Entry<K, V>> entries) {
-        Map<Path, Object> pathMap = new LinkedHashMap<Path, Object>();
+        Map<Path, Object> pathMap = new LinkedHashMap<>();
+        System.getProperties()
+                .forEach(
+                        (key, value) -> {
+                            if (key instanceof String) {
+                                Path path = pathFromPropertyKey((String) key);
+                                pathMap.put(path, value);
+                            }
+                        });
         for (Map.Entry<K, V> entry : entries) {
             Object key = entry.getKey();
             if (key instanceof String) {
@@ -74,7 +81,7 @@ static AbstractConfigObject fromStringMap(ConfigOrigin origin, Map<String, Strin
     }
 
     static AbstractConfigObject fromPathMap(ConfigOrigin origin, Map<?, ?> pathExpressionMap) {
-        Map<Path, Object> pathMap = new LinkedHashMap<Path, Object>();
+        Map<Path, Object> pathMap = new LinkedHashMap<>();
         for (Map.Entry<?, ?> entry : pathExpressionMap.entrySet()) {
             Object keyObj = entry.getKey();
             if (!(keyObj instanceof String)) {
@@ -93,8 +100,8 @@ private static AbstractConfigObject fromPathMap(
          * First, build a list of paths that will have values, either string or
          * object values.
          */
-        Set<Path> scopePaths = new LinkedHashSet<Path>();
-        Set<Path> valuePaths = new LinkedHashSet<Path>();
+        Set<Path> scopePaths = new LinkedHashSet<>();
+        Set<Path> valuePaths = new LinkedHashSet<>();
         for (Path path : pathMap.keySet()) {
             // add value's path
             valuePaths.add(path);
@@ -129,13 +136,11 @@ private static AbstractConfigObject fromPathMap(
         /*
          * Create maps for the object-valued values.
          */
-        Map<String, AbstractConfigValue> root = new LinkedHashMap<String, AbstractConfigValue>();
-        Map<Path, Map<String, AbstractConfigValue>> scopes =
-                new LinkedHashMap<Path, Map<String, AbstractConfigValue>>();
+        Map<String, AbstractConfigValue> root = new LinkedHashMap<>();
+        Map<Path, Map<String, AbstractConfigValue>> scopes = new LinkedHashMap<>();
 
         for (Path path : scopePaths) {
-            Map<String, AbstractConfigValue> scope =
-                    new LinkedHashMap<String, AbstractConfigValue>();
+            Map<String, AbstractConfigValue> scope = new LinkedHashMap<>();
             scopes.put(path, scope);
         }
 
@@ -150,7 +155,17 @@ private static AbstractConfigObject fromPathMap(
             AbstractConfigValue value;
             if (convertedFromProperties) {
                 if (rawValue instanceof String) {
-                    value = new ConfigString.Quoted(origin, (String) rawValue);
+                    if (((String) rawValue).startsWith("[") && ((String) rawValue).endsWith("]")) {
+                        List<String> list =
+                                Arrays.asList(
+                                        ((String) rawValue)
+                                                .substring(1, ((String) rawValue).length() - 1)
+                                                .split(","));
+                        value = ConfigImpl.fromAnyRef(list, origin, FromMapMode.KEYS_ARE_PATHS);
+                    } else {
+                        value = new ConfigString.Quoted(origin, (String) rawValue);
+                    }
+
                 } else {
                     // silently ignore non-string values in Properties
                     value = null;
@@ -167,19 +182,14 @@ private static AbstractConfigObject fromPathMap(
          * Make a list of scope paths from longest to shortest, so children go
          * before parents.
          */
-        List<Path> sortedScopePaths = new ArrayList<Path>();
-        sortedScopePaths.addAll(scopePaths);
+        List<Path> sortedScopePaths = new ArrayList<>(scopePaths);
         // sort descending by length
-        Collections.sort(
-                sortedScopePaths,
-                new Comparator<Path>() {
-                    @Override
-                    public int compare(Path a, Path b) {
-                        // Path.length() is O(n) so in theory this sucks
-                        // but in practice we can make Path precompute length
-                        // if it ever matters.
-                        return b.length() - a.length();
-                    }
+        sortedScopePaths.sort(
+                (a, b) -> {
+                    // Path.length() is O(n) so in theory this sucks
+                    // but in practice we can make Path precompute length
+                    // if it ever matters.
+                    return b.length() - a.length();
                 });
 
         /*

seatunnel-core/seatunnel-core-starter/src/main/java/org/apache/seatunnel/core/starter/command/AbstractCommandArgs.java

@@ -41,6 +41,7 @@ public abstract class AbstractCommandArgs extends CommandArgs {
     /** user-defined parameters */
     @Parameter(
             names = {"-i", "--variable"},
+            splitter = ParameterSplitter.class,
             description = "Variable substitution, such as -i city=beijing, or -i date=20190318")
     protected List<String> variables = Collections.emptyList();
 

seatunnel-core/seatunnel-core-starter/src/main/java/org/apache/seatunnel/core/starter/command/ParameterSplitter.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.core.starter.command;
+
+import com.beust.jcommander.converters.IParameterSplitter;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+
+public class ParameterSplitter implements IParameterSplitter {
+
+    @Override
+    public List<String> split(String value) {
+        if (!value.contains(",")) {
+            return Collections.singletonList(value);
+        }
+
+        List<String> result = new ArrayList<>();
+        StringBuilder currentToken = new StringBuilder();
+        boolean insideBrackets = false;
+
+        for (char c : value.toCharArray()) {
+            if (c == '[') {
+                insideBrackets = true;
+            } else if (c == ']') {
+                insideBrackets = false;
+            }
+
+            if (c == ',' && !insideBrackets) {
+                result.add(currentToken.toString().trim());
+                currentToken = new StringBuilder();
+            } else {
+                currentToken.append(c);
+            }
+        }
+
+        if (currentToken.length() > 0) {
+            result.add(currentToken.toString().trim());
+        }
+
+        return result;
+    }
+}

seatunnel-core/seatunnel-starter/src/test/java/org/apache/seatunnel/core/starter/seatunnel/args/ClientCommandArgsTest.java

@@ -40,6 +40,7 @@ public void testUserDefinedParamsCommand() throws URISyntaxException {
         String password = "dsjr42=4wfskahdsd=w1chh";
         String fakeSourceTable = "fake";
         String fakeSinkTable = "sink";
+        String list = "[par1=20230829,par2=20230829]";
         String[] args = {
             "-c",
             "/args/user_defined_params.conf",
@@ -54,7 +55,9 @@ public void testUserDefinedParamsCommand() throws URISyntaxException {
             "-i",
             "password=" + password,
             "-i",
-            "username=" + username
+            "username=" + username,
+            "-i",
+            "list=" + list,
         };
         ClientCommandArgs clientCommandArgs =
                 CommandLineUtils.parse(args, new ClientCommandArgs(), "seatunnel-zeta", true);
@@ -88,6 +91,9 @@ public void testUserDefinedParamsCommand() throws URISyntaxException {
 
             Assertions.assertEquals(sinkConfig.getString("username"), username);
             Assertions.assertEquals(sinkConfig.getString("password"), password);
+            List<String> list1 = sinkConfig.getStringList("list");
+            Assertions.assertEquals(list1.get(0), "par1=20230829");
+            Assertions.assertEquals(list1.get(1), "par2=20230829");
         }
     }
 }

seatunnel-core/seatunnel-starter/src/test/resources/args/user_defined_params.conf

@@ -47,5 +47,6 @@ sink {
      result_table_name = ${fake_sink_table}
      username = ${username}
      password = ${password}
+     list = ${list}
   }
-}
+}