From 68eee8c7b7f92bd659b16a590a93c8254a2325a4 Mon Sep 17 00:00:00 2001 From: Sun Dapeng Date: Thu, 5 Nov 2026 17:00:24 +0800 Subject: [PATCH] [common] Load index factories across classloader boundaries Discover built-in index factories with the interface classloader so a blind JobMaster context classloader cannot initialize an empty registry. On a lookup miss, discover optional providers with the current context classloader without retaining them in the static registry. Cover child-only provider discovery after blind initialization, sibling classloader isolation, registration checks, and global file selection. Align the CSV dynamic-option test with deferred runtime validation; the test-contract adjustment follows apache/paimon#10989. --- .../fileindex/FileIndexerFactoryUtils.java | 29 +- .../GlobalIndexerFactoryUtils.java | 29 +- .../IndexerFactoryUtilsClassLoaderTest.java | 281 ++++++++++++++++++ .../table/CsvSkipEmptyLinesTableTest.java | 10 +- 4 files changed, 338 insertions(+), 11 deletions(-) create mode 100644 paimon-common/src/test/java/org/apache/paimon/fileindex/IndexerFactoryUtilsClassLoaderTest.java diff --git a/paimon-common/src/main/java/org/apache/paimon/fileindex/FileIndexerFactoryUtils.java b/paimon-common/src/main/java/org/apache/paimon/fileindex/FileIndexerFactoryUtils.java index 944305178cde..7245fc87547c 100644 --- a/paimon-common/src/main/java/org/apache/paimon/fileindex/FileIndexerFactoryUtils.java +++ b/paimon-common/src/main/java/org/apache/paimon/fileindex/FileIndexerFactoryUtils.java @@ -33,8 +33,10 @@ public class FileIndexerFactoryUtils { private static final Map factories = new HashMap<>(); static { + // Not the context classloader: it may not see Paimon, e.g. on a Flink JobMaster RPC thread. ServiceLoader serviceLoader = - ServiceLoader.load(FileIndexerFactory.class); + ServiceLoader.load( + FileIndexerFactory.class, FileIndexerFactory.class.getClassLoader()); for (FileIndexerFactory indexerFactory : serviceLoader) { if (factories.put(indexerFactory.identifier(), indexerFactory) != null) { @@ -46,12 +48,33 @@ public class FileIndexerFactoryUtils { } } + private static FileIndexerFactory find(String type) { + FileIndexerFactory factory = factories.get(type); + if (factory != null) { + return factory; + } + + ClassLoader contextClassLoader = Thread.currentThread().getContextClassLoader(); + if (contextClassLoader == null + || contextClassLoader == FileIndexerFactory.class.getClassLoader()) { + return null; + } + // Keep job-scoped providers local so their classloaders are not retained or shared. + for (FileIndexerFactory contextFactory : + ServiceLoader.load(FileIndexerFactory.class, contextClassLoader)) { + if (contextFactory.identifier().equals(type)) { + return contextFactory; + } + } + return null; + } + public static boolean isRegistered(String type) { - return factories.containsKey(type); + return find(type) != null; } public static FileIndexerFactory load(String type) { - FileIndexerFactory fileIndexerFactory = factories.get(type); + FileIndexerFactory fileIndexerFactory = find(type); if (fileIndexerFactory == null) { throw new RuntimeException("Can't find file index for type: " + type); } diff --git a/paimon-common/src/main/java/org/apache/paimon/globalindex/GlobalIndexerFactoryUtils.java b/paimon-common/src/main/java/org/apache/paimon/globalindex/GlobalIndexerFactoryUtils.java index 6e045ec79c6e..5d8cfc213442 100644 --- a/paimon-common/src/main/java/org/apache/paimon/globalindex/GlobalIndexerFactoryUtils.java +++ b/paimon-common/src/main/java/org/apache/paimon/globalindex/GlobalIndexerFactoryUtils.java @@ -37,8 +37,10 @@ public class GlobalIndexerFactoryUtils { private static final Map factories = new HashMap<>(); static { + // Not the context classloader: it may not see Paimon, e.g. on a Flink JobMaster RPC thread. ServiceLoader serviceLoader = - ServiceLoader.load(GlobalIndexerFactory.class); + ServiceLoader.load( + GlobalIndexerFactory.class, GlobalIndexerFactory.class.getClassLoader()); for (GlobalIndexerFactory indexerFactory : serviceLoader) { if (factories.put(indexerFactory.identifier(), indexerFactory) != null) { @@ -50,8 +52,29 @@ public class GlobalIndexerFactoryUtils { } } + private static GlobalIndexerFactory find(String type) { + GlobalIndexerFactory factory = factories.get(type); + if (factory != null) { + return factory; + } + + ClassLoader contextClassLoader = Thread.currentThread().getContextClassLoader(); + if (contextClassLoader == null + || contextClassLoader == GlobalIndexerFactory.class.getClassLoader()) { + return null; + } + // Keep job-scoped providers local so their classloaders are not retained or shared. + for (GlobalIndexerFactory contextFactory : + ServiceLoader.load(GlobalIndexerFactory.class, contextClassLoader)) { + if (contextFactory.identifier().equals(type)) { + return contextFactory; + } + } + return null; + } + public static GlobalIndexerFactory load(String type) { - GlobalIndexerFactory globalIndexerFactory = factories.get(type); + GlobalIndexerFactory globalIndexerFactory = find(type); if (globalIndexerFactory == null) { throw new RuntimeException("Can't find global index for type: " + type); } @@ -64,7 +87,7 @@ public static List selectFiles( List indexFields, Predicate predicate, List files) { - GlobalIndexerFactory factory = factories.get(type); + GlobalIndexerFactory factory = find(type); return factory == null ? files : factory.selectFiles(indexFields, predicate, files); } } diff --git a/paimon-common/src/test/java/org/apache/paimon/fileindex/IndexerFactoryUtilsClassLoaderTest.java b/paimon-common/src/test/java/org/apache/paimon/fileindex/IndexerFactoryUtilsClassLoaderTest.java new file mode 100644 index 000000000000..02fd312484a9 --- /dev/null +++ b/paimon-common/src/test/java/org/apache/paimon/fileindex/IndexerFactoryUtilsClassLoaderTest.java @@ -0,0 +1,281 @@ +/* + * 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.paimon.fileindex; + +import org.apache.paimon.globalindex.GlobalIndexIOMeta; +import org.apache.paimon.globalindex.GlobalIndexer; +import org.apache.paimon.globalindex.GlobalIndexerFactory; +import org.apache.paimon.globalindex.GlobalIndexerFactoryUtils; +import org.apache.paimon.options.Options; +import org.apache.paimon.predicate.Predicate; +import org.apache.paimon.types.DataField; +import org.apache.paimon.types.DataType; + +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.io.TempDir; + +import java.io.File; +import java.io.InputStream; +import java.net.URL; +import java.net.URLClassLoader; +import java.nio.file.Files; +import java.nio.file.Path; +import java.util.ArrayList; +import java.util.Collections; +import java.util.List; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatThrownBy; + +/** Tests index discovery with blind and child context classloaders. */ +public class IndexerFactoryUtilsClassLoaderTest { + + @Test + public void testFileIndexFactoriesIgnoreContextClassLoader() throws Exception { + Object factory = + loadWithBlindContextClassLoader( + FileIndexerFactoryUtils.class.getName(), "bloom-filter"); + assertThat(factory.getClass().getName()) + .isEqualTo("org.apache.paimon.fileindex.bloomfilter.BloomFilterFileIndexFactory"); + } + + @Test + public void testGlobalIndexFactoriesIgnoreContextClassLoader() throws Exception { + Object factory = + loadWithBlindContextClassLoader(GlobalIndexerFactoryUtils.class.getName(), "btree"); + assertThat(factory.getClass().getName()) + .isEqualTo("org.apache.paimon.globalindex.btree.BTreeGlobalIndexerFactory"); + } + + @Test + public void testFileIndexFactoriesFromChildContextClassLoader(@TempDir Path tempDir) + throws Exception { + Object factory = + loadWithChildContextClassLoader( + FileIndexerFactoryUtils.class.getName(), + ChildFileIndexerFactory.IDENTIFIER, + FileIndexerFactory.class, + ChildFileIndexerFactory.class, + tempDir); + assertThat(factory.getClass().getName()).isEqualTo(ChildFileIndexerFactory.class.getName()); + } + + @Test + public void testGlobalIndexFactoriesFromChildContextClassLoader(@TempDir Path tempDir) + throws Exception { + Object factory = + loadWithChildContextClassLoader( + GlobalIndexerFactoryUtils.class.getName(), + ChildGlobalIndexerFactory.IDENTIFIER, + GlobalIndexerFactory.class, + ChildGlobalIndexerFactory.class, + tempDir); + assertThat(factory.getClass().getName()) + .isEqualTo(ChildGlobalIndexerFactory.class.getName()); + } + + @Test + public void testChildFileIndexRegistration(@TempDir Path tempDir) throws Exception { + writeService(FileIndexerFactory.class, ChildFileIndexerFactory.class, tempDir); + Thread thread = Thread.currentThread(); + ClassLoader original = thread.getContextClassLoader(); + try (URLClassLoader child = + new URLClassLoader( + new URL[] {tempDir.toUri().toURL()}, getClass().getClassLoader())) { + thread.setContextClassLoader(child); + assertThat(FileIndexerFactoryUtils.isRegistered(ChildFileIndexerFactory.IDENTIFIER)) + .isTrue(); + assertThat(FileIndexerFactoryUtils.isRegistered("missing-index")).isFalse(); + thread.setContextClassLoader(FileIndexerFactory.class.getClassLoader()); + assertThat(FileIndexerFactoryUtils.isRegistered(ChildFileIndexerFactory.IDENTIFIER)) + .isFalse(); + } finally { + thread.setContextClassLoader(original); + } + } + + @Test + public void testChildGlobalIndexFileSelection(@TempDir Path tempDir) throws Exception { + writeService(GlobalIndexerFactory.class, ChildGlobalIndexerFactory.class, tempDir); + List files = Collections.singletonList(null); + Thread thread = Thread.currentThread(); + ClassLoader original = thread.getContextClassLoader(); + try (URLClassLoader child = + new URLClassLoader( + new URL[] {tempDir.toUri().toURL()}, getClass().getClassLoader())) { + thread.setContextClassLoader(child); + assertThat( + GlobalIndexerFactoryUtils.selectFiles( + ChildGlobalIndexerFactory.IDENTIFIER, + Collections.emptyList(), + null, + files)) + .isEmpty(); + assertThat( + GlobalIndexerFactoryUtils.selectFiles( + "missing-index", Collections.emptyList(), null, files)) + .isSameAs(files); + thread.setContextClassLoader(GlobalIndexerFactory.class.getClassLoader()); + assertThat( + GlobalIndexerFactoryUtils.selectFiles( + ChildGlobalIndexerFactory.IDENTIFIER, + Collections.emptyList(), + null, + files)) + .isSameAs(files); + } finally { + thread.setContextClassLoader(original); + } + } + + // Initializes the utils class in a fresh classloader, as a Flink JobMaster RPC thread would. + private static Object loadWithBlindContextClassLoader(String utilsClass, String type) + throws Exception { + Thread thread = Thread.currentThread(); + ClassLoader original = thread.getContextClassLoader(); + try (URLClassLoader isolated = + new URLClassLoader( + classPath(), ClassLoader.getSystemClassLoader().getParent()); + URLClassLoader blind = new URLClassLoader(new URL[0], null)) { + thread.setContextClassLoader(blind); + Class utils = Class.forName(utilsClass, true, isolated); + return utils.getMethod("load", String.class).invoke(null, type); + } finally { + thread.setContextClassLoader(original); + } + } + + private static Object loadWithChildContextClassLoader( + String utilsClass, String type, Class service, Class provider, Path tempDir) + throws Exception { + copyProviderClass(provider, tempDir); + writeService(service, provider, tempDir); + + Thread thread = Thread.currentThread(); + ClassLoader original = thread.getContextClassLoader(); + try (URLClassLoader isolated = + new URLClassLoader( + classPath(), ClassLoader.getSystemClassLoader().getParent()); + URLClassLoader blind = new URLClassLoader(new URL[0], null); + URLClassLoader child = + new URLClassLoader(new URL[] {tempDir.toUri().toURL()}, isolated); + URLClassLoader sibling = + new URLClassLoader(new URL[] {tempDir.toUri().toURL()}, isolated)) { + assertThatThrownBy(() -> isolated.loadClass(provider.getName())) + .isInstanceOf(ClassNotFoundException.class); + thread.setContextClassLoader(blind); + Class utils = Class.forName(utilsClass, true, isolated); + assertThatThrownBy(() -> utils.getMethod("load", String.class).invoke(null, type)) + .hasRootCauseMessage( + "Can't find " + + (service == FileIndexerFactory.class ? "file" : "global") + + " index for type: " + + type); + thread.setContextClassLoader(child); + Object factory = utils.getMethod("load", String.class).invoke(null, type); + assertThat(factory.getClass().getClassLoader()).isSameAs(child); + thread.setContextClassLoader(sibling); + Object siblingFactory = utils.getMethod("load", String.class).invoke(null, type); + assertThat(siblingFactory.getClass().getClassLoader()).isSameAs(sibling); + thread.setContextClassLoader(null); + assertThatThrownBy(() -> utils.getMethod("load", String.class).invoke(null, type)) + .hasRootCauseMessage( + "Can't find " + + (service == FileIndexerFactory.class ? "file" : "global") + + " index for type: " + + type); + return factory; + } finally { + thread.setContextClassLoader(original); + } + } + + private static void writeService(Class service, Class provider, Path tempDir) + throws Exception { + Path services = tempDir.resolve("META-INF/services"); + Files.createDirectories(services); + Files.write(services.resolve(service.getName()), Collections.singleton(provider.getName())); + } + + private static void copyProviderClass(Class provider, Path tempDir) throws Exception { + String resourceName = provider.getName().replace('.', '/') + ".class"; + Path classFile = tempDir.resolve(resourceName); + Files.createDirectories(classFile.getParent()); + try (InputStream input = provider.getClassLoader().getResourceAsStream(resourceName)) { + if (input == null) { + throw new IllegalStateException("Cannot find provider class: " + resourceName); + } + Files.copy(input, classFile); + } + } + + /** Test factory loaded from a child classloader. */ + public static class ChildFileIndexerFactory implements FileIndexerFactory { + + private static final String IDENTIFIER = "child-file-index"; + + @Override + public String identifier() { + return IDENTIFIER; + } + + @Override + public FileIndexer create(DataType type, Options options) { + return null; + } + } + + /** Test factory loaded from a child classloader. */ + public static class ChildGlobalIndexerFactory implements GlobalIndexerFactory { + + private static final String IDENTIFIER = "child-global-index"; + + @Override + public String identifier() { + return IDENTIFIER; + } + + @Override + public GlobalIndexer create(List indexFields, Options options) { + return null; + } + + @Override + public List selectFiles( + List indexFields, Predicate predicate, List files) { + return Collections.emptyList(); + } + } + + private static URL[] classPath() throws Exception { + List urls = new ArrayList<>(); + URL testClasses = + IndexerFactoryUtilsClassLoaderTest.class + .getProtectionDomain() + .getCodeSource() + .getLocation(); + for (String entry : System.getProperty("java.class.path").split(File.pathSeparator)) { + URL url = new File(entry).toURI().toURL(); + if (!url.equals(testClasses)) { + urls.add(url); + } + } + return urls.toArray(new URL[0]); + } +} diff --git a/paimon-core/src/test/java/org/apache/paimon/table/CsvSkipEmptyLinesTableTest.java b/paimon-core/src/test/java/org/apache/paimon/table/CsvSkipEmptyLinesTableTest.java index 0a11cb8e5b06..51b8fb80e76d 100644 --- a/paimon-core/src/test/java/org/apache/paimon/table/CsvSkipEmptyLinesTableTest.java +++ b/paimon-core/src/test/java/org/apache/paimon/table/CsvSkipEmptyLinesTableTest.java @@ -92,11 +92,11 @@ void testRejectedAsDynamicOptionOfFileStoreTable() throws Exception { // skipped while the file metadata still counts it. write(table, GenericRow.of((Object) null), GenericRow.of(1)); - assertThatThrownBy( - () -> - table.copy( - Collections.singletonMap( - CsvOptions.SKIP_EMPTY_LINES.key(), "true"))) + // File-store validation is deferred until runtime initialization, so the dynamic option + // is rejected when the copied table is read rather than when it is copied. + Table copied = + table.copy(Collections.singletonMap(CsvOptions.SKIP_EMPTY_LINES.key(), "true")); + assertThatThrownBy(() -> read(copied)) .satisfies(anyCauseMatches(IllegalArgumentException.class, ERROR)); // Without the option, the scan and the row count in the metadata agree.