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);
+        }
     }
 }