[fix](catalog) safely publish Hadoop properties

fe/fe-core/src/main/java/org/apache/doris/datasource/CatalogProperty.java

@@ -33,6 +33,7 @@
 import org.apache.logging.log4j.LogManager;
 import org.apache.logging.log4j.Logger;
 
+import java.util.Collections;
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
@@ -281,10 +282,13 @@ public Map<String, String> getBackendStorageProperties() {
      * Get Hadoop properties with lazy loading, using double-check locking to ensure thread safety
      */
     public Map<String, String> getHadoopProperties() {
-        if (hadoopProperties == null) {
+        // Retain the observed snapshot because invalidation may clear the volatile cache concurrently.
+        Map<String, String> cachedProperties = hadoopProperties;
+        if (cachedProperties == null) {
             synchronized (this) {
-                if (hadoopProperties == null) {
-                    hadoopProperties = new HashMap<>();
+                cachedProperties = hadoopProperties;
+                if (cachedProperties == null) {
+                    Map<String, String> result = new HashMap<>();
                     Map<StorageProperties.Type, StorageProperties> storageMap = getStoragePropertiesMap();
 
                     for (StorageProperties sp : storageMap.values()) {
@@ -294,15 +298,19 @@ public Map<String, String> getHadoopProperties() {
                                 String key = entry.getKey();
                                 String value = entry.getValue();
                                 if (value != null) {
-                                    hadoopProperties.put(key, value);
+                                    result.put(key, value);
                                 }
                             });
                         }
                     }
-                    StorageProperties.setCombinedFsCacheKey(hadoopProperties, storageMap.values());
+                    StorageProperties.setCombinedFsCacheKey(result, storageMap.values());
+                    // Readers share this snapshot without locking, so publish it only when complete
+                    // and keep caller-specific mutations out of the catalog cache.
+                    cachedProperties = Collections.unmodifiableMap(result);
+                    hadoopProperties = cachedProperties;
                 }
             }
         }
-        return hadoopProperties;
+        return cachedProperties;
     }
 }

fe/fe-core/src/main/java/org/apache/doris/tablefunction/PaimonTableValuedFunction.java

@@ -48,6 +48,7 @@
 import org.apache.paimon.table.source.Split;
 import org.apache.paimon.table.system.SystemTableLoader;
 
+import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
 import java.util.stream.Collectors;
@@ -86,7 +87,8 @@ public PaimonTableValuedFunction(TableName paimonTableName, String queryType) th
         }
 
         PaimonExternalCatalog paimonExternalCatalog = (PaimonExternalCatalog) dorisCatalog;
-        this.hadoopProps = paimonExternalCatalog.getCatalogProperty().getHadoopProperties();
+        // Keep TVF-specific Kerberos entries isolated from the catalog's shared immutable snapshot.
+        this.hadoopProps = new HashMap<>(paimonExternalCatalog.getCatalogProperty().getHadoopProperties());
         appendHMSKerberosProps(hadoopProps, paimonExternalCatalog);
         this.hadoopAuthenticator = paimonExternalCatalog.getExecutionAuthenticator();
 

fe/fe-core/src/test/java/org/apache/doris/datasource/CatalogPropertyTest.java

@@ -0,0 +1,137 @@
+// 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.doris.datasource;
+
+import org.apache.doris.datasource.property.storage.StorageProperties;
+
+import org.apache.hadoop.conf.Configuration;
+import org.junit.Assert;
+import org.junit.Test;
+import org.mockito.Mockito;
+
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.Iterator;
+import java.util.Map;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicReference;
+
+public class CatalogPropertyTest {
+
+    @Test
+    public void testHadoopPropertiesArePublishedAfterInitialization() throws Exception {
+        CountDownLatch iterationStarted = new CountDownLatch(1);
+        CountDownLatch allowIteration = new CountDownLatch(1);
+        Configuration configuration = new BlockingConfiguration(iterationStarted, allowIteration);
+        configuration.set("fs.test.property", "complete");
+
+        StorageProperties storageProperties = Mockito.mock(StorageProperties.class);
+        Mockito.when(storageProperties.getHadoopStorageConfig()).thenReturn(configuration);
+        Mockito.when(storageProperties.getFsCacheFingerprint()).thenReturn("test-fingerprint");
+
+        CatalogProperty catalogProperty = new CatalogProperty(null, Collections.emptyMap()) {
+            @Override
+            public Map<StorageProperties.Type, StorageProperties> getStoragePropertiesMap() {
+                return Collections.singletonMap(StorageProperties.Type.HDFS, storageProperties);
+            }
+        };
+
+        ExecutorService executor = Executors.newSingleThreadExecutor();
+        AtomicReference<Map<String, String>> readerResult = new AtomicReference<>();
+        Thread concurrentReader = new Thread(
+                () -> readerResult.set(new HashMap<>(catalogProperty.getHadoopProperties())));
+        try {
+            Future<Map<String, String>> initializer = executor.submit(catalogProperty::getHadoopProperties);
+            Assert.assertTrue(iterationStarted.await(5, TimeUnit.SECONDS));
+
+            concurrentReader.start();
+            Assert.assertTrue(waitUntilBlockedOrTerminated(concurrentReader, 5, TimeUnit.SECONDS));
+            Assert.assertEquals("The reader must block until initialization publishes the completed map",
+                    Thread.State.BLOCKED, concurrentReader.getState());
+
+            allowIteration.countDown();
+            Assert.assertEquals("complete", initializer.get(5, TimeUnit.SECONDS).get("fs.test.property"));
+            concurrentReader.join(TimeUnit.SECONDS.toMillis(5));
+            Assert.assertFalse(concurrentReader.isAlive());
+            Assert.assertEquals("complete", readerResult.get().get("fs.test.property"));
+        } finally {
+            allowIteration.countDown();
+            concurrentReader.interrupt();
+            executor.shutdownNow();
+        }
+    }
+
+    @Test
+    public void testHadoopPropertiesCacheIsImmutable() {
+        Configuration configuration = new Configuration(false);
+        configuration.set("fs.test.property", "complete");
+
+        StorageProperties storageProperties = Mockito.mock(StorageProperties.class);
+        Mockito.when(storageProperties.getHadoopStorageConfig()).thenReturn(configuration);
+        Mockito.when(storageProperties.getFsCacheFingerprint()).thenReturn("test-fingerprint");
+
+        CatalogProperty catalogProperty = new CatalogProperty(null, Collections.emptyMap()) {
+            @Override
+            public Map<StorageProperties.Type, StorageProperties> getStoragePropertiesMap() {
+                return Collections.singletonMap(StorageProperties.Type.HDFS, storageProperties);
+            }
+        };
+
+        Map<String, String> hadoopProperties = catalogProperty.getHadoopProperties();
+        Assert.assertThrows(UnsupportedOperationException.class,
+                () -> hadoopProperties.put("fs.test.property", "modified"));
+    }
+
+    private static boolean waitUntilBlockedOrTerminated(Thread thread, long timeout, TimeUnit timeUnit) {
+        long deadline = System.nanoTime() + timeUnit.toNanos(timeout);
+        while (thread.isAlive() && thread.getState() != Thread.State.BLOCKED
+                && System.nanoTime() < deadline) {
+            Thread.yield();
+        }
+        return !thread.isAlive() || thread.getState() == Thread.State.BLOCKED;
+    }
+
+    private static class BlockingConfiguration extends Configuration {
+        private final CountDownLatch iterationStarted;
+        private final CountDownLatch allowIteration;
+
+        BlockingConfiguration(CountDownLatch iterationStarted, CountDownLatch allowIteration) {
+            super(false);
+            this.iterationStarted = iterationStarted;
+            this.allowIteration = allowIteration;
+        }
+
+        @Override
+        public Iterator<Map.Entry<String, String>> iterator() {
+            iterationStarted.countDown();
+            try {
+                if (!allowIteration.await(5, TimeUnit.SECONDS)) {
+                    throw new AssertionError("Timed out waiting to continue Hadoop configuration initialization");
+                }
+            } catch (InterruptedException e) {
+                Thread.currentThread().interrupt();
+                throw new AssertionError("Interrupted while initializing Hadoop configuration", e);
+            }
+            return super.iterator();
+        }
+    }
+}