diff --git a/fe/fe-core/src/main/java/org/apache/doris/datasource/CatalogProperty.java b/fe/fe-core/src/main/java/org/apache/doris/datasource/CatalogProperty.java index 1ca2cad83c2191..b3081296effb91 100644 --- a/fe/fe-core/src/main/java/org/apache/doris/datasource/CatalogProperty.java +++ b/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 getBackendStorageProperties() { * Get Hadoop properties with lazy loading, using double-check locking to ensure thread safety */ public Map getHadoopProperties() { - if (hadoopProperties == null) { + // Retain the observed snapshot because invalidation may clear the volatile cache concurrently. + Map cachedProperties = hadoopProperties; + if (cachedProperties == null) { synchronized (this) { - if (hadoopProperties == null) { - hadoopProperties = new HashMap<>(); + cachedProperties = hadoopProperties; + if (cachedProperties == null) { + Map result = new HashMap<>(); Map storageMap = getStoragePropertiesMap(); for (StorageProperties sp : storageMap.values()) { @@ -294,15 +298,19 @@ public Map 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; } } diff --git a/fe/fe-core/src/main/java/org/apache/doris/tablefunction/PaimonTableValuedFunction.java b/fe/fe-core/src/main/java/org/apache/doris/tablefunction/PaimonTableValuedFunction.java index 3e7e247e865fa1..bd7da501ccdbdb 100644 --- a/fe/fe-core/src/main/java/org/apache/doris/tablefunction/PaimonTableValuedFunction.java +++ b/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(); diff --git a/fe/fe-core/src/test/java/org/apache/doris/datasource/CatalogPropertyTest.java b/fe/fe-core/src/test/java/org/apache/doris/datasource/CatalogPropertyTest.java new file mode 100644 index 00000000000000..071505565f816f --- /dev/null +++ b/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 getStoragePropertiesMap() { + return Collections.singletonMap(StorageProperties.Type.HDFS, storageProperties); + } + }; + + ExecutorService executor = Executors.newSingleThreadExecutor(); + AtomicReference> readerResult = new AtomicReference<>(); + Thread concurrentReader = new Thread( + () -> readerResult.set(new HashMap<>(catalogProperty.getHadoopProperties()))); + try { + Future> 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 getStoragePropertiesMap() { + return Collections.singletonMap(StorageProperties.Type.HDFS, storageProperties); + } + }; + + Map 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> 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(); + } + } +}