[Feature][Paimon] Customize the hadoop user (#8888)

Author: hawk9821

Diff for: docs/en/connector-v2/sink/Paimon.md

@@ -246,6 +246,49 @@ sink {
 }
 ```
 
+### Single table(Specify hadoop HA config with hadoop_user_name) 
+
+```hocon
+env {
+  parallelism = 1
+  job.mode = "STREAMING"
+  checkpoint.interval = 5000
+}
+
+source {
+  Mysql-CDC {
+    base-url = "jdbc:mysql://127.0.0.1:3306/seatunnel"
+    username = "root"
+    password = "******"
+    table-names = ["seatunnel.role"]
+  }
+}
+
+transform {
+}
+
+sink {
+  Paimon {
+    catalog_name="seatunnel_test"
+    warehouse="hdfs:///tmp/seatunnel/paimon/hadoop-sink/"
+    database="seatunnel"
+    table="role"
+    paimon.hadoop.conf = {
+      hadoop_user_name = "hdfs"
+      fs.defaultFS = "hdfs://nameservice1"
+      dfs.nameservices = "nameservice1"
+      dfs.ha.namenodes.nameservice1 = "nn1,nn2"
+      dfs.namenode.rpc-address.nameservice1.nn1 = "hadoop03:8020"
+      dfs.namenode.rpc-address.nameservice1.nn2 = "hadoop04:8020"
+      dfs.client.failover.proxy.provider.nameservice1 = "org.apache.hadoop.hdfs.server.namenode.ha.ConfiguredFailoverProxyProvider"
+      dfs.client.use.datanode.hostname = "true"
+      security.kerberos.login.principal = "your-kerberos-principal"
+      security.kerberos.login.keytab = "your-kerberos-keytab-path"
+    }
+  }
+}
+```
+
 ### Single table(Hive catalog)
 
 ```hocon

Diff for: docs/en/connector-v2/source/Paimon.md

@@ -152,6 +152,7 @@ source {
     table="st_test"
     query = "select * from st_test where pk_id is not null and pk_id < 3"
     paimon.hadoop.conf = {
+      hadoop_user_name = "hdfs"
       fs.defaultFS = "hdfs://nameservice1"
       dfs.nameservices = "nameservice1"
       dfs.ha.namenodes.nameservice1 = "nn1,nn2"

Diff for: docs/zh/connector-v2/sink/Paimon.md

@@ -244,6 +244,49 @@ sink {
 }
 ```
 
+### 单表(指定hadoop HA配置和指定hadoop用户名)
+
+```hocon
+env {
+  parallelism = 1
+  job.mode = "STREAMING"
+  checkpoint.interval = 5000
+}
+
+source {
+  Mysql-CDC {
+    base-url = "jdbc:mysql://127.0.0.1:3306/seatunnel"
+    username = "root"
+    password = "******"
+    table-names = ["seatunnel.role"]
+  }
+}
+
+transform {
+}
+
+sink {
+  Paimon {
+    catalog_name="seatunnel_test"
+    warehouse="hdfs:///tmp/seatunnel/paimon/hadoop-sink/"
+    database="seatunnel"
+    table="role"
+    paimon.hadoop.conf = {
+      hadoop_user_name = "hdfs"
+      fs.defaultFS = "hdfs://nameservice1"
+      dfs.nameservices = "nameservice1"
+      dfs.ha.namenodes.nameservice1 = "nn1,nn2"
+      dfs.namenode.rpc-address.nameservice1.nn1 = "hadoop03:8020"
+      dfs.namenode.rpc-address.nameservice1.nn2 = "hadoop04:8020"
+      dfs.client.failover.proxy.provider.nameservice1 = "org.apache.hadoop.hdfs.server.namenode.ha.ConfiguredFailoverProxyProvider"
+      dfs.client.use.datanode.hostname = "true"
+      security.kerberos.login.principal = "your-kerberos-principal"
+      security.kerberos.login.keytab = "your-kerberos-keytab-path"
+    }
+  }
+}
+```
+
 ### 单表(使用Hive catalog)
 
 ```hocon

Diff for: docs/zh/connector-v2/source/Paimon.md

@@ -154,6 +154,7 @@ source {
     table="st_test"
     query = "select * from st_test where pk_id is not null and pk_id < 3"
     paimon.hadoop.conf = {
+      hadoop_user_name = "hdfs"
       fs.defaultFS = "hdfs://nameservice1"
       dfs.nameservices = "nameservice1"
       dfs.ha.namenodes.nameservice1 = "nn1,nn2"

Diff for: seatunnel-connectors-v2/connector-paimon/src/main/java/org/apache/seatunnel/connectors/seatunnel/paimon/catalog/PaimonCatalogLoader.java

@@ -25,6 +25,7 @@
 
 import org.apache.commons.lang3.StringUtils;
 import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.security.UserGroupInformation;
 import org.apache.paimon.catalog.Catalog;
 import org.apache.paimon.catalog.CatalogContext;
 import org.apache.paimon.catalog.CatalogFactory;
@@ -50,6 +51,8 @@ public class PaimonCatalogLoader implements Serializable {
 
     private static final String HDFS_IMPL_KEY = "fs.hdfs.impl";
 
+    private static final String HADOOP_USER_NAME = "hadoop_user_name";
+
     private String warehouse;
     private PaimonCatalogEnum catalogType;
     private String catalogUri;
@@ -72,6 +75,10 @@ public Catalog loadCatalog() {
         if (warehouse.startsWith(HDFS_PREFIX)) {
             checkConfiguration(paimonHadoopConfiguration, HDFS_DEF_FS_NAME);
             paimonHadoopConfiguration.set(HDFS_IMPL_KEY, HDFS_IMPL);
+            String username = paimonHadoopConfiguration.get(HADOOP_USER_NAME);
+            if (StringUtils.isNotBlank(username)) {
+                UserGroupInformation.setLoginUser(UserGroupInformation.createRemoteUser(username));
+            }
         } else if (warehouse.startsWith(S3A_PREFIX)) {
             optionsMap.putAll(paimonHadoopConfiguration.getPropsWithPrefix(StringUtils.EMPTY));
         }

Diff for: seatunnel-e2e/seatunnel-connector-v2-e2e/connector-paimon-e2e/src/test/resources/fake_sink_paimon_truncate_with_hdfs_case1.conf

@@ -68,6 +68,7 @@ sink {
     database = "seatunnel_namespace11"
     table = "st_test"
     paimon.hadoop.conf = {
+      hadoop_user_name = "hdfs"
       fs.defaultFS = "hdfs://nameservice1"
       dfs.nameservices = "nameservice1"
       dfs.ha.namenodes.nameservice1 = "nn1,nn2"

Diff for: seatunnel-e2e/seatunnel-connector-v2-e2e/connector-paimon-e2e/src/test/resources/fake_sink_paimon_truncate_with_hdfs_case2.conf

@@ -53,6 +53,7 @@ sink {
     table = "st_test"
     data_save_mode=DROP_DATA
     paimon.hadoop.conf = {
+      hadoop_user_name = "hdfs"
       fs.defaultFS = "hdfs://nameservice1"
       dfs.nameservices = "nameservice1"
       dfs.ha.namenodes.nameservice1 = "nn1,nn2"