feat: add user in spark

java/openmldb-spark-connector/src/main/java/com/_4paradigm/openmldb/spark/OpenmldbSource.java

@@ -62,6 +62,14 @@ public StructType inferSchema(CaseInsensitiveStringMap options) {
         if (timeout != null) {
             option.setRequestTimeout(Integer.parseInt(timeout));
         }
+        String user = options.get("user");
+        if (user != null) {
+            option.setUser(user);
+        }
+        String password = options.get("password");
+        if (password != null) {
+            option.setPassword(password);
+        }
         String debug = options.get("debug");
         if (debug != null) {
             option.setEnableDebug(Boolean.valueOf(debug));

java/openmldb-spark-connector/src/main/java/com/_4paradigm/openmldb/spark/read/OpenmldbReadConfig.java

@@ -22,13 +22,15 @@
 
 // Must serializable
 public class OpenmldbReadConfig implements Serializable {
-    public final String dbName, tableName, zkCluster, zkPath;
+    public final String dbName, tableName, zkCluster, zkPath, user, password;
 
     public OpenmldbReadConfig(String dbName, String tableName, SdkOption option) {
         this.dbName = dbName;
         this.tableName = tableName;
         this.zkCluster = option.getZkCluster();
         this.zkPath = option.getZkPath();
+        this.user = option.getUser();
+        this.password = option.getPassword();
         // TODO(hw): other configs in SdkOption
     }
 }

java/openmldb-spark-connector/src/main/java/com/_4paradigm/openmldb/spark/write/OpenmldbDataSingleWriter.java

@@ -45,6 +45,8 @@ public OpenmldbDataSingleWriter(OpenmldbWriteConfig config, int partitionId, lon
             option.setZkCluster(config.zkCluster);
             option.setZkPath(config.zkPath);
             option.setLight(true);
+            option.setUser(config.user);
+            option.setPassword(config.password);
             SqlClusterExecutor executor = new SqlClusterExecutor(option);
             String dbName = config.dbName;
             String tableName = config.tableName;

java/openmldb-spark-connector/src/main/java/com/_4paradigm/openmldb/spark/write/OpenmldbDataWriter.java

@@ -45,6 +45,8 @@ public OpenmldbDataWriter(OpenmldbWriteConfig config, int partitionId, long task
             option.setZkCluster(config.zkCluster);
             option.setZkPath(config.zkPath);
             option.setLight(true);
+            option.setUser(config.user);
+            option.setPassword(config.password);
             SqlClusterExecutor executor = new SqlClusterExecutor(option);
             String dbName = config.dbName;
             String tableName = config.tableName;

java/openmldb-spark-connector/src/main/java/com/_4paradigm/openmldb/spark/write/OpenmldbWriteConfig.java

@@ -23,14 +23,16 @@
 
 // Must serializable
 public class OpenmldbWriteConfig implements Serializable {
-    public final String dbName, tableName, zkCluster, zkPath, writerType;
+    public final String dbName, tableName, zkCluster, zkPath, writerType, user, password;
 
     public OpenmldbWriteConfig(String dbName, String tableName, SdkOption option, String writerType) {
         this.dbName = dbName;
         this.tableName = tableName;
         this.zkCluster = option.getZkCluster();
         this.zkPath = option.getZkPath();
         this.writerType = writerType;
+        this.user = option.getUser();
+        this.password = option.getPassword();
         // TODO(hw): other configs in SdkOption
     }
 }

java/openmldb-spark-connector/src/main/scala/com/_4paradigm/openmldb/spark/read/OpenmldbPartitionReader.scala

@@ -14,6 +14,8 @@ class OpenmldbPartitionReader(config: OpenmldbReadConfig) extends PartitionReade
   option.setZkCluster(config.zkCluster)
   option.setZkPath(config.zkPath)
   option.setLight(true)
+  option.setUser(config.user)
+  option.setPassword(config.password)
   val executor = new SqlClusterExecutor(option)
   val dbName: String = config.dbName
   val tableName: String = config.tableName

java/openmldb-synctool/src/main/java/com/_4paradigm/openmldb/synctool/SyncToolConfig.java

@@ -37,6 +37,9 @@ public class SyncToolConfig {
     // public static int CHANNEL_KEEP_ALIVE_TIME;
     public static String ZK_CLUSTER;
     public static String ZK_ROOT_PATH;
+
+    public static String USER;
+    public static String PASSWORD;
     public static String ZK_CERT;
     public static String SYNC_TASK_PROGRESS_PATH;
 
@@ -87,6 +90,8 @@ private static void parseFromProperties(Properties prop) {
         if (ZK_ROOT_PATH.isEmpty()) {
             throw new RuntimeException("zookeeper.root_path should not be empty");
         }
+        USER = prop.getProperty("user", "root");
+        PASSWORD = prop.getProperty("password", "");
         ZK_CERT = prop.getProperty("zookeeper.cert", "");
 
         HADOOP_CONF_DIR = prop.getProperty("hadoop.conf.dir", "");

java/openmldb-synctool/src/main/java/com/_4paradigm/openmldb/synctool/SyncToolImpl.java

@@ -92,6 +92,8 @@ public SyncToolImpl(String endpoint) throws SqlException, InterruptedException {
         option.setZkCluster(SyncToolConfig.ZK_CLUSTER);
         option.setZkPath(SyncToolConfig.ZK_ROOT_PATH);
         option.setZkCert(SyncToolConfig.ZK_CERT);
+        option.setUser(SyncToolConfig.USER);
+        option.setPassword(SyncToolConfig.PASSWORD);
         this.router = new SqlClusterExecutor(option);
         this.zkCollectorPath = SyncToolConfig.ZK_ROOT_PATH + "/sync_tool/collector";
 

java/openmldb-taskmanager/src/main/java/com/_4paradigm/openmldb/taskmanager/config/TaskManagerConfig.java

@@ -121,6 +121,10 @@ public static int getZkMaxConnectWaitTime() {
         return getInt("zookeeper.max_connect_waitTime");
     }
 
+    public static String getUser() { return getString("user"); }
+
+    public static String getPassword() { return getString("password"); }
+
     public static String getSparkMaster() {
         return getString("spark.master");
     }
@@ -283,6 +287,14 @@ private void init() throws ConfigException {
             props.setProperty("zookeeper.session_timeout", "5000");
         }
 
+        if (props.getProperty("user") == null) {
+            props.setProperty("user", "root");
+        }
+
+        if (props.getProperty("password") == null) {
+            props.setProperty("password", "");
+        }
+
         if (getZkSessionTimeout() <= 0) {
             throw new ConfigException("zookeeper.session_timeout", "should be larger than 0");
         }

java/openmldb-taskmanager/src/main/scala/com/_4paradigm/openmldb/taskmanager/JobInfoManager.scala

@@ -45,6 +45,10 @@ object JobInfoManager {
   private val option = new SdkOption
   option.setZkCluster(TaskManagerConfig.getZkCluster)
   option.setZkPath(TaskManagerConfig.getZkRootPath)
+  option.setUser(TaskManagerConfig.getUser)
+  if (!TaskManagerConfig.getPassword.isEmpty) {
+    option.setPassword(TaskManagerConfig.getPassword)
+  }
   val sqlExecutor = new SqlClusterExecutor(option)
   sqlExecutor.executeSQL("", "set @@execute_mode='online';")