[lake] Fail fast when a lake plugin is shadowed by a jar in lib

fluss-common/src/main/java/org/apache/fluss/lake/lakestorage/LakeStoragePluginSetUp.java

@@ -17,11 +17,12 @@
 
 package org.apache.fluss.lake.lakestorage;
 
+import org.apache.fluss.exception.FlussRuntimeException;
 import org.apache.fluss.plugin.PluginManager;
-import org.apache.fluss.shaded.guava32.com.google.common.collect.Iterators;
 
 import javax.annotation.Nullable;
 
+import java.util.Collections;
 import java.util.Iterator;
 import java.util.Objects;
 import java.util.ServiceLoader;
@@ -32,35 +33,65 @@
  */
 public class LakeStoragePluginSetUp {
 
+    /**
+     * Finds the {@link LakeStoragePlugin} for the given datalake format, from the plugins directory
+     * or from the main classpath.
+     *
+     * <p>Providing it from both cannot work: {@code org.apache.fluss.} is parent-first, so the copy
+     * on the main classpath shadows the plugin and is loaded without the dependencies bundled next
+     * to it. That is rejected here rather than failing later with a {@link NoClassDefFoundError}.
+     */
     public static LakeStoragePlugin fromDataLakeFormat(
             final String dataLakeFormat, @Nullable final PluginManager pluginManager) {
-        // now, load lake storage plugin
-        Iterator<LakeStoragePlugin> lakeStoragePluginIterator =
-                getAllLakeStoragePlugins(pluginManager);
+        LakeStoragePlugin fromPluginsDir =
+                findByIdentifier(loadFromPluginManager(pluginManager), dataLakeFormat);
+        LakeStoragePlugin fromClasspath = findByIdentifier(loadFromClasspath(), dataLakeFormat);
 
-        while (lakeStoragePluginIterator.hasNext()) {
-            LakeStoragePlugin lakeStoragePlugin = lakeStoragePluginIterator.next();
+        if (fromPluginsDir != null && fromClasspath != null) {
+            throw new FlussRuntimeException(
+                    String.format(
+                            "Found two LakeStoragePlugin for datalake format '%s': one in the plugins "
+                                    + "directory and one on the main classpath (usually <FLUSS_HOME>/lib). "
+                                    + "The copy on the main classpath shadows the one in the plugins "
+                                    + "directory, but is loaded without the dependencies bundled next to "
+                                    + "the plugin, such as Hadoop. This later fails with errors like "
+                                    + "'NoClassDefFoundError: org/apache/hadoop/hdfs/HdfsConfiguration'. "
+                                    + "Remove the fluss-lake-%s jar from <FLUSS_HOME>/lib and keep it only "
+                                    + "in <FLUSS_HOME>/plugins/%s/.",
+                            dataLakeFormat, dataLakeFormat, dataLakeFormat));
+        }
+
+        LakeStoragePlugin lakeStoragePlugin =
+                fromPluginsDir != null ? fromPluginsDir : fromClasspath;
+        if (lakeStoragePlugin == null) {
+            throw new UnsupportedOperationException(
+                    "No LakeStoragePlugin can be found for datalake format: " + dataLakeFormat);
+        }
+        return PluginLakeStorageWrapper.of(lakeStoragePlugin);
+    }
+
+    @Nullable
+    private static LakeStoragePlugin findByIdentifier(
+            Iterator<LakeStoragePlugin> lakeStoragePlugins, String dataLakeFormat) {
+        while (lakeStoragePlugins.hasNext()) {
+            LakeStoragePlugin lakeStoragePlugin = lakeStoragePlugins.next();
             if (Objects.equals(lakeStoragePlugin.identifier(), dataLakeFormat)) {
-                return PluginLakeStorageWrapper.of(lakeStoragePlugin);
+                return lakeStoragePlugin;
             }
         }
-
-        // if come here, means we haven't found LakeStoragePlugin match the configured
-        // datalake, throw exception
-        throw new UnsupportedOperationException(
-                "No LakeStoragePlugin can be found for datalake format: " + dataLakeFormat);
+        return null;
     }
 
-    private static Iterator<LakeStoragePlugin> getAllLakeStoragePlugins(
+    private static Iterator<LakeStoragePlugin> loadFromPluginManager(
             @Nullable PluginManager pluginManager) {
-        final Iterator<LakeStoragePlugin> pluginIteratorSPI =
-                ServiceLoader.load(
-                                LakeStoragePlugin.class, LakeStoragePlugin.class.getClassLoader())
-                        .iterator();
         if (pluginManager == null) {
-            return pluginIteratorSPI;
-        } else {
-            return Iterators.concat(pluginManager.load(LakeStoragePlugin.class), pluginIteratorSPI);
+            return Collections.emptyIterator();
         }
+        return pluginManager.load(LakeStoragePlugin.class);
+    }
+
+    private static Iterator<LakeStoragePlugin> loadFromClasspath() {
+        return ServiceLoader.load(LakeStoragePlugin.class, LakeStoragePlugin.class.getClassLoader())
+                .iterator();
     }
 }

fluss-common/src/test/java/org/apache/fluss/lake/lakestorage/LakeStorageTest.java

@@ -18,6 +18,7 @@
 package org.apache.fluss.lake.lakestorage;
 
 import org.apache.fluss.config.Configuration;
+import org.apache.fluss.exception.FlussRuntimeException;
 import org.apache.fluss.exception.TableAlreadyExistException;
 import org.apache.fluss.exception.TableNotExistException;
 import org.apache.fluss.lake.source.LakeSource;
@@ -98,6 +99,36 @@ void testWithPluginManager() throws Exception {
                 .isInstanceOf(TestPaimonLakeCatalog.class);
     }
 
+    @Test
+    void testPluginFoundInBothPluginsDirAndClasspath() {
+        // same identifier from the plugin manager and from SPI, as when the jar is also in lib
+        final Map<Class<?>, Iterator<?>> lakeStoragePlugins = new HashMap<>();
+        lakeStoragePlugins.put(
+                LakeStoragePlugin.class,
+                Collections.singletonList(new TestingClasspathLakeStoragePlugin()).iterator());
+
+        assertThatThrownBy(
+                        () ->
+                                LakeStoragePluginSetUp.fromDataLakeFormat(
+                                        TestingClasspathLakeStoragePlugin.IDENTIFIER,
+                                        new TestingPluginManager(lakeStoragePlugins)))
+                .isInstanceOf(FlussRuntimeException.class)
+                .hasMessageContaining("Found two LakeStoragePlugin for datalake format")
+                .hasMessageContaining(TestingClasspathLakeStoragePlugin.IDENTIFIER)
+                .hasMessageContaining("<FLUSS_HOME>/lib");
+    }
+
+    @Test
+    void testPluginFoundOnClasspathOnly() {
+        LakeStoragePlugin lakeStoragePlugin =
+                LakeStoragePluginSetUp.fromDataLakeFormat(
+                        TestingClasspathLakeStoragePlugin.IDENTIFIER, null);
+
+        assertThat(lakeStoragePlugin).isInstanceOf(PluginLakeStorageWrapper.class);
+        assertThat(lakeStoragePlugin.identifier())
+                .isEqualTo(TestingClasspathLakeStoragePlugin.IDENTIFIER);
+    }
+
     private static class TestingPluginManager implements PluginManager {
 
         private final Map<Class<?>, Iterator<?>> plugins;

fluss-common/src/test/java/org/apache/fluss/lake/lakestorage/TestingClasspathLakeStoragePlugin.java

@@ -0,0 +1,36 @@
+/*
+ * 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.fluss.lake.lakestorage;
+
+import org.apache.fluss.config.Configuration;
+
+/** A {@link LakeStoragePlugin} registered via SPI on the main classpath. */
+public class TestingClasspathLakeStoragePlugin implements LakeStoragePlugin {
+
+    public static final String IDENTIFIER = "test-classpath-plugin";
+
+    @Override
+    public String identifier() {
+        return IDENTIFIER;
+    }
+
+    @Override
+    public LakeStorage createLakeStorage(Configuration configuration) {
+        throw new UnsupportedOperationException("Not implemented");
+    }
+}

fluss-common/src/test/java/org/apache/fluss/plugin/PluginDirectoryLayoutTest.java

@@ -0,0 +1,159 @@
+/*
+ * 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.fluss.plugin;
+
+import org.apache.fluss.plugin.jar.plugina.DynamicClassA;
+import org.apache.fluss.plugin.jar.plugina.TestServiceA;
+import org.apache.fluss.shaded.guava32.com.google.common.collect.Lists;
+
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.io.TempDir;
+
+import java.io.File;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.nio.file.Paths;
+import java.util.Collection;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Set;
+
+import static org.apache.fluss.utils.Preconditions.checkState;
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+/**
+ * Tests for the plugin directory layout of a deployed cluster, where each plugin lives in its own
+ * sub directory of {@code <FLUSS_HOME>/plugins} and the main classpath is {@code <FLUSS_HOME>/lib}.
+ */
+class PluginDirectoryLayoutTest extends PluginTestBase {
+
+    @TempDir private Path tmp;
+
+    @Test
+    void testMultiplePluginsLoadedSimultaneously() throws Exception {
+        PluginManager pluginManager =
+                createPluginManager(
+                        new String[] {TestSpi.class.getName(), OtherTestSpi.class.getName()},
+                        "multi");
+
+        List<TestSpi> plugins = Lists.newArrayList(pluginManager.load(TestSpi.class));
+
+        assertThat(plugins).hasSize(2);
+        Set<ClassLoader> classLoaders = new HashSet<>();
+        for (TestSpi plugin : plugins) {
+            assertThat(plugin.testMethod()).isNotNull();
+            assertThat(plugin.getClassLoader()).isNotSameAs(PARENT_CLASS_LOADER);
+            classLoaders.add(plugin.getClassLoader());
+        }
+        assertThat(classLoaders).hasSize(2);
+    }
+
+    @Test
+    void testPluginsDoNotSeeEachOthersClasses() throws Exception {
+        PluginManager pluginManager =
+                createPluginManager(
+                        new String[] {TestSpi.class.getName(), OtherTestSpi.class.getName()},
+                        "isolation");
+
+        List<TestSpi> plugins = Lists.newArrayList(pluginManager.load(TestSpi.class));
+        assertThat(plugins).hasSize(2);
+
+        // DynamicClassA is bundled in plugin A's jar only
+        ClassLoader loaderOfA = classLoaderOf(plugins, TestServiceA.class.getName());
+        ClassLoader loaderOfB = otherClassLoader(plugins, loaderOfA);
+
+        assertThat(loaderOfA.loadClass(DynamicClassA.class.getName())).isNotNull();
+        assertThatThrownBy(() -> loaderOfB.loadClass(DynamicClassA.class.getName()))
+                .isInstanceOf(ClassNotFoundException.class);
+    }
+
+    /** The main classpath copy wins, leaving the dependencies bundled next to the plugin unused. */
+    @Test
+    void testClassOnMainClasspathShadowsPluginCopy() throws Exception {
+        // TestServiceA is on the main classpath, standing in for a jar also copied into lib
+        PluginManager pluginManager =
+                createPluginManager(
+                        new String[] {
+                            TestSpi.class.getName(),
+                            OtherTestSpi.class.getName(),
+                            TestServiceA.class.getName()
+                        },
+                        "shadowed");
+
+        List<TestSpi> plugins = Lists.newArrayList(pluginManager.load(TestSpi.class));
+        TestSpi fromPluginA = pluginNamed(plugins, TestServiceA.class.getName());
+
+        assertThat(fromPluginA).isInstanceOf(TestServiceA.class);
+        assertThat(fromPluginA.getClassLoader()).isSameAs(PARENT_CLASS_LOADER);
+    }
+
+    @Test
+    void testPluginCopyIsUsedWhenNotOnMainClasspath() throws Exception {
+        PluginManager pluginManager =
+                createPluginManager(
+                        new String[] {TestSpi.class.getName(), OtherTestSpi.class.getName()},
+                        "not-shadowed");
+
+        List<TestSpi> plugins = Lists.newArrayList(pluginManager.load(TestSpi.class));
+        TestSpi fromPluginA = pluginNamed(plugins, TestServiceA.class.getName());
+
+        // same class name, but a distinct class loaded from the plugin jar
+        assertThat(fromPluginA).isNotInstanceOf(TestServiceA.class);
+        assertThat(fromPluginA.getClassLoader()).isNotSameAs(PARENT_CLASS_LOADER);
+    }
+
+    /** Builds a {@code plugins/} root holding plugin A and plugin B in separate sub directories. */
+    private PluginManager createPluginManager(String[] parentPatterns, String name)
+            throws Exception {
+        File pluginRootFolder = new File(tmp.toFile(), name);
+        File pluginAFolder = new File(pluginRootFolder, "A");
+        File pluginBFolder = new File(pluginRootFolder, "B");
+        checkState(pluginAFolder.mkdirs());
+        checkState(pluginBFolder.mkdirs());
+        Files.copy(locateJarFile(PLUGIN_A).toPath(), Paths.get(pluginAFolder.toString(), PLUGIN_A));
+        Files.copy(locateJarFile(PLUGIN_B).toPath(), Paths.get(pluginBFolder.toString(), PLUGIN_B));
+
+        Collection<PluginDescriptor> descriptors =
+                new DirectoryBasedPluginFinder(pluginRootFolder.toPath()).findPlugins();
+        checkState(descriptors.size() == 2);
+        return new DefaultPluginManager(descriptors, PARENT_CLASS_LOADER, parentPatterns);
+    }
+
+    private static TestSpi pluginNamed(List<TestSpi> plugins, String className) {
+        for (TestSpi plugin : plugins) {
+            if (plugin.getClass().getName().equals(className)) {
+                return plugin;
+            }
+        }
+        throw new AssertionError("No plugin found with class name " + className);
+    }
+
+    private static ClassLoader classLoaderOf(List<TestSpi> plugins, String className) {
+        return pluginNamed(plugins, className).getClassLoader();
+    }
+
+    private static ClassLoader otherClassLoader(List<TestSpi> plugins, ClassLoader classLoader) {
+        for (TestSpi plugin : plugins) {
+            if (plugin.getClassLoader() != classLoader) {
+                return plugin.getClassLoader();
+            }
+        }
+        throw new AssertionError("Expected a second, distinct plugin class loader");
+    }
+}

fluss-common/src/test/resources/META-INF/services/org.apache.fluss.lake.lakestorage.LakeStoragePlugin

@@ -0,0 +1,17 @@
+# 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.
+
+org.apache.fluss.lake.lakestorage.TestingClasspathLakeStoragePlugin

website/docs/install-deploy/deploying-streaming-lakehouse.md

@@ -124,6 +124,14 @@ See [set_cluster_configs](../engine-flink/procedures.md#set_cluster_configs) for
 
 Add JARs to `${FLUSS_HOME}/plugins/<format>/` based on your configuration:
 
+:::warning
+These JARs go in `${FLUSS_HOME}/plugins/<format>/`, together with the `fluss-lake-<format>` JAR and
+the dependencies it needs. Do not also place them in `${FLUSS_HOME}/lib`: a copy on the main
+classpath shadows the plugin and is loaded without its bundled dependencies, so the CoordinatorServer
+fails to start with errors such as `NoClassDefFoundError: org/apache/hadoop/hdfs/HdfsConfiguration`.
+See [Troubleshooting Plugins](../maintenance/troubleshooting-plugins.md).
+:::
+
 <Tabs groupId="datalake-format">
 <TabItem value="paimon" label="Paimon" default>
 
@@ -169,7 +177,8 @@ The Tiering Service is a Flink job that continuously tiers data from Fluss to th
 
 ### Flink JARs
 
-Add the following to `${FLINK_HOME}/lib`:
+Add the following to `${FLINK_HOME}/lib` — this is the **Flink** installation, not
+`${FLUSS_HOME}/lib`:
 
 <Tabs groupId="datalake-format">
 <TabItem value="paimon" label="Paimon" default>