From 97377194f119fb7dcb111fd024f6e5907fe4ce40 Mon Sep 17 00:00:00 2001 From: Gabriel Date: Mon, 3 Aug 2026 21:18:28 +0800 Subject: [PATCH 1/2] [fix](catalog) safely publish Hadoop properties --- .../doris/datasource/CatalogProperty.java | 9 +- .../doris/datasource/CatalogPropertyTest.java | 109 ++++++++++++++++++ 2 files changed, 115 insertions(+), 3 deletions(-) create mode 100644 fe/fe-core/src/test/java/org/apache/doris/datasource/CatalogPropertyTest.java 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..1e8c156d49b0fb 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 @@ -284,7 +284,9 @@ public Map getHadoopProperties() { if (hadoopProperties == null) { synchronized (this) { if (hadoopProperties == null) { - hadoopProperties = new HashMap<>(); + // Publish the volatile cache only after construction because readers skip this + // lock once it is non-null and must never observe a map still being mutated. + Map result = new HashMap<>(); Map storageMap = getStoragePropertiesMap(); for (StorageProperties sp : storageMap.values()) { @@ -294,12 +296,13 @@ 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()); + hadoopProperties = result; } } } 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..b17dfc008eb047 --- /dev/null +++ b/fe/fe-core/src/test/java/org/apache/doris/datasource/CatalogPropertyTest.java @@ -0,0 +1,109 @@ +// 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.TimeoutException; + +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.newFixedThreadPool(2); + CountDownLatch readerStarted = new CountDownLatch(1); + try { + Future> initializer = executor.submit(catalogProperty::getHadoopProperties); + Assert.assertTrue(iterationStarted.await(5, TimeUnit.SECONDS)); + + Future> concurrentReader = executor.submit(() -> { + readerStarted.countDown(); + return new HashMap<>(catalogProperty.getHadoopProperties()); + }); + Assert.assertTrue(readerStarted.await(5, TimeUnit.SECONDS)); + try { + concurrentReader.get(200, TimeUnit.MILLISECONDS); + Assert.fail("Concurrent readers must not observe a partially initialized cache"); + } catch (TimeoutException expected) { + // The reader must wait for the initializing thread to publish the completed map. + } + + allowIteration.countDown(); + Assert.assertEquals("complete", initializer.get(5, TimeUnit.SECONDS).get("fs.test.property")); + Assert.assertEquals("complete", concurrentReader.get(5, TimeUnit.SECONDS).get("fs.test.property")); + } finally { + allowIteration.countDown(); + executor.shutdownNow(); + } + } + + 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(); + } + } +} From edc0a8f21b5dd4415665a1786884a08278b4e7c4 Mon Sep 17 00:00:00 2001 From: Gabriel Date: Tue, 4 Aug 2026 08:25:42 +0800 Subject: [PATCH 2/2] [fix](catalog) address Hadoop property cache review --- .../doris/datasource/CatalogProperty.java | 17 ++++-- .../PaimonTableValuedFunction.java | 4 +- .../doris/datasource/CatalogPropertyTest.java | 58 ++++++++++++++----- 3 files changed, 57 insertions(+), 22 deletions(-) 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 1e8c156d49b0fb..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,11 +282,12 @@ 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) { - // Publish the volatile cache only after construction because readers skip this - // lock once it is non-null and must never observe a map still being mutated. + cachedProperties = hadoopProperties; + if (cachedProperties == null) { Map result = new HashMap<>(); Map storageMap = getStoragePropertiesMap(); @@ -302,10 +304,13 @@ public Map getHadoopProperties() { } } StorageProperties.setCombinedFsCacheKey(result, storageMap.values()); - hadoopProperties = result; + // 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 index b17dfc008eb047..071505565f816f 100644 --- 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 @@ -33,7 +33,7 @@ import java.util.concurrent.Executors; import java.util.concurrent.Future; import java.util.concurrent.TimeUnit; -import java.util.concurrent.TimeoutException; +import java.util.concurrent.atomic.AtomicReference; public class CatalogPropertyTest { @@ -55,33 +55,61 @@ public Map getStoragePropertiesMap() } }; - ExecutorService executor = Executors.newFixedThreadPool(2); - CountDownLatch readerStarted = new CountDownLatch(1); + 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)); - Future> concurrentReader = executor.submit(() -> { - readerStarted.countDown(); - return new HashMap<>(catalogProperty.getHadoopProperties()); - }); - Assert.assertTrue(readerStarted.await(5, TimeUnit.SECONDS)); - try { - concurrentReader.get(200, TimeUnit.MILLISECONDS); - Assert.fail("Concurrent readers must not observe a partially initialized cache"); - } catch (TimeoutException expected) { - // The reader must wait for the initializing thread to publish the completed map. - } + 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")); - Assert.assertEquals("complete", concurrentReader.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;