From 626111e65073e78e03441d14965e942c07029bfe Mon Sep 17 00:00:00 2001
From: kang <35803862+ghkang98@users.noreply.github.com>
Date: Fri, 9 May 2025 21:30:28 +0800
Subject: [PATCH] [fix](iceberg) fix iceberg hadoop type kerbros issue.
 (#50623)

Support Pre-Execution Authentication for Hadoop Type Iceberg Catalog
Operations Summary
This PR fix the hadoop type iceberg catalog kerbros use the
PreExecutionAuthenticator class, This is especially useful in
environments where secure access is required, such as Kerberos-based
Hadoop ecosystems. By integrating PreExecutionAuthenticator, each
relevant operation will undergo an authentication step prior to
execution, maintaining security compliance.


- Behavior changed:
     No.

- Does this need documentation?
     No.

Co-authored-by: lik40 <lik40@chinatelecom.cn>
---
 .../iceberg/IcebergHadoopExternalCatalog.java | 31 ++++++++++++++-----
 1 file changed, 24 insertions(+), 7 deletions(-)

diff --git a/fe/fe-core/src/main/java/org/apache/doris/datasource/iceberg/IcebergHadoopExternalCatalog.java b/fe/fe-core/src/main/java/org/apache/doris/datasource/iceberg/IcebergHadoopExternalCatalog.java
index bf9e8c2b3f06fb..cd84093c74793a 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/datasource/iceberg/IcebergHadoopExternalCatalog.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/datasource/iceberg/IcebergHadoopExternalCatalog.java
@@ -18,6 +18,8 @@
 package org.apache.doris.datasource.iceberg;
 
 import org.apache.doris.catalog.HdfsResource;
+import org.apache.doris.common.security.authentication.AuthenticationConfig;
+import org.apache.doris.common.security.authentication.HadoopAuthenticator;
 import org.apache.doris.datasource.CatalogProperty;
 import org.apache.doris.datasource.property.PropertyConverter;
 
@@ -52,15 +54,30 @@ public IcebergHadoopExternalCatalog(long catalogId, String name, String resource
     @Override
     protected void initCatalog() {
         icebergCatalogType = ICEBERG_HADOOP;
-        HadoopCatalog hadoopCatalog = new HadoopCatalog();
+
         Configuration conf = getConfiguration();
         initS3Param(conf);
+
+        //create the authenticator first
+        if (preExecutionAuthenticator.getHadoopAuthenticator() == null) {
+            AuthenticationConfig config = AuthenticationConfig.getKerberosConfig(conf);
+            HadoopAuthenticator authenticator = HadoopAuthenticator.getHadoopAuthenticator(config);
+            preExecutionAuthenticator.setHadoopAuthenticator(authenticator);
+        }
+
         // initialize hadoop catalog
-        Map<String, String> catalogProperties = catalogProperty.getProperties();
-        String warehouse = catalogProperty.getHadoopProperties().get(CatalogProperties.WAREHOUSE_LOCATION);
-        hadoopCatalog.setConf(conf);
-        catalogProperties.put(CatalogProperties.WAREHOUSE_LOCATION, warehouse);
-        hadoopCatalog.initialize(getName(), catalogProperties);
-        catalog = hadoopCatalog;
+        try {
+            this.catalog = preExecutionAuthenticator.execute(() -> {
+                Map<String, String> catalogProperties = catalogProperty.getProperties();
+                String warehouse = catalogProperty.getHadoopProperties().get(CatalogProperties.WAREHOUSE_LOCATION);
+                HadoopCatalog hadoopCatalog = new HadoopCatalog();
+                hadoopCatalog.setConf(conf);
+                catalogProperties.put(CatalogProperties.WAREHOUSE_LOCATION, warehouse);
+                hadoopCatalog.initialize(getName(), catalogProperties);
+                return hadoopCatalog;
+            });
+        } catch (Exception e) {
+            throw new RuntimeException("Hadoop catalog init error!", e);
+        }
     }
 }