[feature](plugin) replace the always-true connector apiVersion() with a manifest-carried API version gate for all four plugin families

.licenserc.yaml

@@ -52,6 +52,13 @@ header:
     # Recorded method-surface baseline; ConnectorMetadataSurfaceTest reads every
     # non-blank line as a method signature, so a header would break the test.
     - "fe/fe-connector/fe-connector-api/src/test/resources/connector-metadata-methods.txt"
+    # Recorded plugin API surface baselines, one per plugin family. Same reason as
+    # above: each *PluginSurfaceTest.readBaseline() adds every non-blank line to the
+    # expected signature set, so a header would be read back as phantom signatures.
+    - "fe/fe-connector/fe-connector-spi/src/test/resources/connector-plugin-surface.txt"
+    - "fe/fe-filesystem/fe-filesystem-spi/src/test/resources/filesystem-plugin-surface.txt"
+    - "fe/fe-authentication/fe-authentication-spi/src/test/resources/authentication-plugin-surface.txt"
+    - "fe/fe-core/src/test/resources/lineage-plugin-surface.txt"
     - "fe/fe-core/src/main/java/software/amazon/awssdk/core/client/builder/SdkDefaultClientBuilder.java"
     - "fe/fe-core/src/main/antlr4/org/apache/doris/nereids/JavaLexer.g4"
     - "fe/fe-core/src/main/antlr4/org/apache/doris/nereids/JavaParser.g4"

fe/fe-authentication/fe-authentication-handler/src/main/java/org/apache/doris/authentication/handler/AuthenticationPluginManager.java

@@ -21,6 +21,7 @@
 import org.apache.doris.authentication.AuthenticationIntegration;
 import org.apache.doris.authentication.spi.AuthenticationPlugin;
 import org.apache.doris.authentication.spi.AuthenticationPluginFactory;
+import org.apache.doris.extension.loader.ApiVersionGate;
 import org.apache.doris.extension.loader.ClassLoadingPolicy;
 import org.apache.doris.extension.loader.DirectoryPluginRuntimeManager;
 import org.apache.doris.extension.loader.LoadFailure;
@@ -40,6 +41,7 @@
 import java.util.Optional;
 import java.util.ServiceLoader;
 import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.CopyOnWriteArrayList;
 
 /**
  * Manager for authentication plugins.
@@ -65,9 +67,28 @@ public class AuthenticationPluginManager {
     /** Family label in the process-wide {@link PluginRegistry}. */
     private static final String PLUGIN_FAMILY = "AUTHENTICATION";
 
+    /**
+     * The authentication plugin API contract this FE serves. Built from the version filtered into
+     * fe-authentication-spi at build time, anchored on {@link AuthenticationPluginFactory} so that it is read
+     * from the very artifact carrying the SPI. A missing or malformed resource is a build defect and fails
+     * class initialization loudly rather than degrading into a check that admits everything.
+     */
+    private static final ApiVersionGate API_VERSION_GATE =
+            ApiVersionGate.forFamily("authentication", AuthenticationPluginFactory.class);
+
     /** Factories by plugin name (e.g., "ldap", "oidc", "password") */
     private final Map<String, AuthenticationPluginFactory> factories = new ConcurrentHashMap<>();
 
+    /**
+     * Plugins the last {@link #loadAll} refused on their declared API version, newest run only.
+     *
+     * <p>Authentication is the one family that loads lazily and reports "no factory for this type" from a
+     * different place than the load itself. Without this, an operator whose plugin was refused on its version
+     * sees only "not found", with the actual reason buried in an FE log line — so the reason is kept here and
+     * appended to that exception (see {@link #apiVersionRejectionHint()}).
+     */
+    private final List<String> apiVersionRejections = new CopyOnWriteArrayList<>();
+
     /** Plugin instances by integration name */
     private final Map<String, AuthenticationPlugin> pluginByIntegration = new ConcurrentHashMap<>();
 
@@ -142,11 +163,16 @@ public void loadAll(List<Path> pluginRoots, ClassLoader parent) throws Authentic
                 pluginRoots,
                 parent,
                 AuthenticationPluginFactory.class,
-                classLoadingPolicy);
+                classLoadingPolicy,
+                API_VERSION_GATE);
 
+        apiVersionRejections.clear();
         for (LoadFailure failure : report.getFailures()) {
             LOG.warn("Skip plugin directory due to load failure: pluginDir={}, stage={}, message={}",
                     failure.getPluginDir(), failure.getStage(), failure.getMessage(), failure.getCause());
+            if (LoadFailure.STAGE_API_VERSION.equals(failure.getStage())) {
+                apiVersionRejections.add(failure.getMessage());
+            }
         }
 
         int loadedPlugins = 0;
@@ -177,6 +203,23 @@ public void loadAll(List<Path> pluginRoots, ClassLoader parent) throws Authentic
         }
     }
 
+    /**
+     * A clause naming any plugin the last {@link #loadAll} refused on its declared API version, or the empty
+     * string when there was none.
+     *
+     * <p>Callers append this to their "no factory for type X" exception. A plugin directory that loaded some
+     * other plugin successfully does not throw from {@code loadAll} at all, so without this the version
+     * rejection would never reach the user — exactly the undiagnosable case this exists to prevent.
+     */
+    public String apiVersionRejectionHint() {
+        if (apiVersionRejections.isEmpty()) {
+            return "";
+        }
+        return ". Note that " + apiVersionRejections.size()
+                + " plugin(s) were refused on their declared API version: "
+                + String.join("; ", apiVersionRejections);
+    }
+
     private static LoadFailure firstNonConflictFailure(List<LoadFailure> failures) {
         for (LoadFailure failure : failures) {
             if (!LoadFailure.STAGE_CONFLICT.equals(failure.getStage())) {

fe/fe-authentication/fe-authentication-handler/src/test/java/org/apache/doris/authentication/handler/AuthenticationPluginManagerTest.java

@@ -24,8 +24,10 @@
 import org.apache.doris.authentication.BasicPrincipal;
 import org.apache.doris.authentication.spi.AuthenticationPlugin;
 import org.apache.doris.authentication.spi.AuthenticationPluginFactory;
+import org.apache.doris.extension.loader.ApiVersionGate;
 import org.apache.doris.extension.loader.ClassLoadingPolicy;
 import org.apache.doris.extension.loader.DirectoryPluginRuntimeManager;
+import org.apache.doris.extension.loader.LoadFailure;
 import org.apache.doris.extension.loader.LoadReport;
 import org.apache.doris.extension.loader.PluginHandle;
 
@@ -36,6 +38,7 @@
 
 import java.io.Closeable;
 import java.io.IOException;
+import java.io.InputStream;
 import java.nio.charset.StandardCharsets;
 import java.nio.file.Files;
 import java.nio.file.Path;
@@ -45,15 +48,25 @@
 import java.util.Collections;
 import java.util.List;
 import java.util.Optional;
+import java.util.jar.Attributes;
 import java.util.jar.JarEntry;
 import java.util.jar.JarOutputStream;
+import java.util.jar.Manifest;
 
 /**
  * Unit tests for {@link AuthenticationPluginManager}.
  */
 @DisplayName("PluginManager Unit Tests")
 public class AuthenticationPluginManagerTest {
 
+    /**
+     * The same contract {@link AuthenticationPluginManager} loads plugins under. Building it here through the
+     * public {@code forFamily} path also asserts, by construction, that fe-authentication-spi really ships
+     * the filtered kernel resource — a build that dropped it fails every test in this class.
+     */
+    private static final ApiVersionGate AUTH_GATE =
+            ApiVersionGate.forFamily("authentication", AuthenticationPluginFactory.class);
+
     private AuthenticationPluginManager pluginManager;
 
     @BeforeEach
@@ -365,9 +378,114 @@ void testLoadAll_ConflictOnlyRepeatedInvocationIsIdempotent() throws Exception {
         Assertions.assertTrue(pluginManager.hasFactory("external-dir-test"));
     }
 
+    @Test
+    @DisplayName("UT-HANDLER-PM-021: Plugin declaring an incompatible API version is refused, with a reason")
+    void testLoadAll_IncompatibleApiVersionIsRefusedWithDiagnosableReason() throws Exception {
+        // Given: one plugin dir that loads fine, and one whose jar declares a different major. The healthy
+        // one matters: it keeps loadAll from throwing, which is exactly the case where the refusal would
+        // otherwise reach the operator as a bare "no factory found for type".
+        Path root = Files.createTempDirectory("plugin-root-api-version");
+        createServiceOnlyJar(
+                Files.createDirectories(root.resolve("external-dir-test")).resolve("external-dir-test.jar"),
+                DirectoryPluginFactory.class.getName());
+        createPluginJar(
+                Files.createDirectories(root.resolve("stale-plugin")).resolve("stale-plugin.jar"),
+                DirectoryPluginFactory.class.getName(),
+                (AUTH_GATE.getExpectedMajor() + 1) + ".0");
+
+        // When
+        pluginManager.loadAll(Arrays.asList(root), Thread.currentThread().getContextClassLoader());
+
+        // Then: the healthy plugin is in, and the refusal is retrievable rather than only in the FE log.
+        Assertions.assertTrue(pluginManager.hasFactory("external-dir-test"));
+        String hint = pluginManager.apiVersionRejectionHint();
+        Assertions.assertTrue(hint.contains("stale-plugin"), hint);
+        Assertions.assertTrue(hint.contains(AUTH_GATE.getManifestAttribute()), hint);
+        Assertions.assertTrue(hint.contains(AUTH_GATE.getExpectedVersion()), hint);
+    }
+
+    @Test
+    @DisplayName("UT-HANDLER-PM-022: Plugin declaring no API version is refused (fail-closed)")
+    void testLoadAll_UndeclaredApiVersionIsRefused() throws Exception {
+        // Given: a jar built with no awareness of this contract at all.
+        Path root = Files.createTempDirectory("plugin-root-no-api-version");
+        createPluginJar(
+                Files.createDirectories(root.resolve("external-dir-test")).resolve("external-dir-test.jar"),
+                DirectoryPluginFactory.class.getName(),
+                null);
+
+        // When: this is the only directory, so loadAll reports total failure...
+        AuthenticationException ex = Assertions.assertThrows(
+                AuthenticationException.class,
+                () -> pluginManager.loadAll(
+                        Arrays.asList(root),
+                        Thread.currentThread().getContextClassLoader()));
+
+        // Then: ...and the thrown message already carries the version reason, so the caller that wraps it
+        // does not have to.
+        Assertions.assertFalse(pluginManager.hasFactory("external-dir-test"));
+        Assertions.assertTrue(ex.getMessage().contains(LoadFailure.STAGE_API_VERSION), ex.getMessage());
+        Assertions.assertTrue(ex.getMessage().contains(AUTH_GATE.getManifestAttribute()), ex.getMessage());
+    }
+
+    @Test
+    @DisplayName("UT-HANDLER-PM-023: A clean load leaves no stale rejection hint behind")
+    void testLoadAll_RejectionHintIsResetPerRun() throws Exception {
+        Path bad = Files.createTempDirectory("plugin-root-hint-reset-bad");
+        createPluginJar(
+                Files.createDirectories(bad.resolve("stale-plugin")).resolve("stale-plugin.jar"),
+                DirectoryPluginFactory.class.getName(),
+                (AUTH_GATE.getExpectedMajor() + 1) + ".0");
+        Path good = Files.createTempDirectory("plugin-root-hint-reset-good");
+        createServiceOnlyJar(
+                Files.createDirectories(good.resolve("external-dir-test")).resolve("external-dir-test.jar"),
+                DirectoryPluginFactory.class.getName());
+
+        Assertions.assertThrows(AuthenticationException.class, () -> pluginManager.loadAll(
+                Arrays.asList(bad), Thread.currentThread().getContextClassLoader()));
+        Assertions.assertFalse(pluginManager.apiVersionRejectionHint().isEmpty());
+
+        // A later successful load must not keep advertising the earlier run's rejection: the hint is
+        // appended to "no factory found" messages, and a stale one would misdiagnose an unrelated failure.
+        pluginManager.loadAll(Arrays.asList(good), Thread.currentThread().getContextClassLoader());
+        Assertions.assertEquals("", pluginManager.apiVersionRejectionHint());
+    }
+
     private static void createServiceOnlyJar(Path jarPath, String providerClassName) throws IOException {
+        createPluginJar(jarPath, providerClassName, AUTH_GATE.getExpectedVersion());
+    }
+
+    /**
+     * Builds a plugin jar the way the build really produces one: a MANIFEST declaring the authentication
+     * plugin API version, the provider's class bytes, and the ServiceLoader registration.
+     *
+     * <p>The class bytes matter for the version check, not just for classloading: the loader reads the
+     * declared version from the jar that <em>defines</em> the factory class, so a jar carrying only a service
+     * descriptor (with the implementation coming from the FE's own classpath) declares nothing and is
+     * refused. Pass a null {@code declaredApiVersion} to reproduce exactly that.
+     */
+    private static void createPluginJar(Path jarPath, String providerClassName, String declaredApiVersion)
+            throws IOException {
         Files.createDirectories(jarPath.getParent());
-        try (JarOutputStream jar = new JarOutputStream(Files.newOutputStream(jarPath))) {
+        Manifest manifest = new Manifest();
+        manifest.getMainAttributes().put(Attributes.Name.MANIFEST_VERSION, "1.0");
+        if (declaredApiVersion != null) {
+            manifest.getMainAttributes().putValue(AUTH_GATE.getManifestAttribute(), declaredApiVersion);
+        }
+        try (JarOutputStream jar = new JarOutputStream(Files.newOutputStream(jarPath), manifest)) {
+            String classEntry = providerClassName.replace('.', '/') + ".class";
+            try (InputStream classBytes = AuthenticationPluginManagerTest.class.getClassLoader()
+                    .getResourceAsStream(classEntry)) {
+                if (classBytes != null) {
+                    jar.putNextEntry(new JarEntry(classEntry));
+                    byte[] buffer = new byte[8192];
+                    int read;
+                    while ((read = classBytes.read(buffer)) != -1) {
+                        jar.write(buffer, 0, read);
+                    }
+                    jar.closeEntry();
+                }
+            }
             JarEntry serviceEntry = new JarEntry(
                     "META-INF/services/org.apache.doris.authentication.spi.AuthenticationPluginFactory");
             jar.putNextEntry(serviceEntry);
@@ -507,7 +625,8 @@ private StaticRuntimeManager(LoadReport<AuthenticationPluginFactory> report) {
 
         @Override
         public LoadReport<AuthenticationPluginFactory> loadAll(List<Path> pluginRoots, ClassLoader parent,
-                Class<AuthenticationPluginFactory> factoryType, ClassLoadingPolicy policy) {
+                Class<AuthenticationPluginFactory> factoryType, ClassLoadingPolicy policy,
+                ApiVersionGate apiVersionGate) {
             return report;
         }
 

fe/fe-authentication/fe-authentication-spi/pom.xml

@@ -41,4 +41,21 @@ under the License.
             <version>${project.version}</version>
         </dependency>
     </dependencies>
+
+    <build>
+        <resources>
+            <!-- Default resource root, restated because declaring <resources> replaces the default. -->
+            <resource>
+                <directory>src/main/resources</directory>
+            </resource>
+            <!--
+              Kept apart from src/main/resources so that ${...} substitution applies ONLY to the
+              plugin API version file and can never rewrite a service descriptor by accident.
+            -->
+            <resource>
+                <directory>src/main/resources-filtered</directory>
+                <filtering>true</filtering>
+            </resource>
+        </resources>
+    </build>
 </project>

fe/fe-authentication/fe-authentication-spi/src/main/resources-filtered/META-INF/doris/authentication-plugin-api-version.properties

@@ -0,0 +1,23 @@
+# 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.
+
+# The AUTHENTICATION plugin API version this FE build serves, read by ApiVersionGate at startup.
+#
+# GENERATED BY MAVEN RESOURCE FILTERING - the value below comes from <authentication.plugin.api.version>
+# in fe/fe-authentication/pom.xml, the same property that stamps Doris-*-Plugin-Api-Version into plugin jars.
+# Never edit the version here; edit that property.
+api.version=${authentication.plugin.api.version}