This is an automated email from the ASF dual-hosted git repository.
Gabriel39 pushed a commit to branch branch-4.1
in repository https://gitbox.apache.org/repos/asf/doris.git
The following commit(s) were added to refs/heads/branch-4.1 by this push:
new e0115cac0cd [fix](catalog) safely publish Hadoop properties (#66392)
e0115cac0cd is described below
commit e0115cac0cd7fcef867801469a05ceb2b3ca1616
Author: Gabriel <[email protected]>
AuthorDate: Tue Aug 4 12:04:16 2026 +0800
[fix](catalog) safely publish Hadoop properties (#66392)
### What problem does this PR solve?
Concurrent first consumers of a catalog's Hadoop properties could
observe the cache before its initialization completed. One thread could
copy the `HashMap` while the initializer was still populating it,
causing a `ConcurrentModificationException` during external table sink
binding.
### What is changed?
- Build the Hadoop property map locally and publish it through the
volatile cache only after all entries and the filesystem cache key are
ready.
- Add a deterministic concurrency regression test that verifies readers
cannot observe a partially initialized cache.
### Validation
- `CatalogPropertyTest`
- `PaimonWriteBindingTest`
- `StoragePropertiesFsCacheFingerprintTest`
- FE Checkstyle
---
.../apache/doris/datasource/CatalogProperty.java | 20 ++-
.../tablefunction/PaimonTableValuedFunction.java | 4 +-
.../doris/datasource/CatalogPropertyTest.java | 137 +++++++++++++++++++++
3 files changed, 154 insertions(+), 7 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 1ca2cad83c2..b3081296eff 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.hadoop.conf.Configuration;
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 class CatalogProperty {
* 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 class CatalogProperty {
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 3e7e247e865..bd7da501ccd 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.Table;
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 class PaimonTableValuedFunction extends
MetadataTableValuedFunction {
}
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 00000000000..071505565f8
--- /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<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();
+ }
+ }
+}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
