[server] Support plugin discovery for client authentication (#3474)

Author: swuferhong

fluss-common/src/main/java/org/apache/fluss/config/ConfigOptions.java

@@ -1291,6 +1291,18 @@ public class ConfigOptions {
                     .withDescription(
                             "The authentication protocol used to authenticate the client.");
 
+    public static final ConfigOption<Boolean> CLIENT_SECURITY_ENABLE_PLUGIN_DISCOVERY =
+            key("client.security.enable-plugin-discovery")
+                    .booleanType()
+                    .defaultValue(false)
+                    .withDescription(
+                            "When set to true, the client will additionally discover "
+                                    + "authentication plugins from the configured plugins/ folder "
+                                    + "via the PluginManager, in addition to the default classpath. "
+                                    + "This is useful for standalone tools or scripts that run outside "
+                                    + "the server JVM but still need to load authentication plugins "
+                                    + "shipped in plugins/.");
+
     public static final ConfigOption<MemorySize> CLIENT_SCANNER_LOG_FETCH_MAX_BYTES =
             key("client.scanner.log.fetch.max-bytes")
                     .memoryType()

fluss-common/src/main/java/org/apache/fluss/security/auth/AuthenticationFactory.java

@@ -68,8 +68,15 @@ public static Supplier<ClientAuthenticator> loadClientAuthenticatorSupplier(
             Configuration configuration) {
         String clientAuthenticateProtocol =
                 configuration.getString(ConfigOptions.CLIENT_SECURITY_PROTOCOL);
+        PluginManager pluginManager = null;
+        if (configuration.getBoolean(ConfigOptions.CLIENT_SECURITY_ENABLE_PLUGIN_DISCOVERY)) {
+            pluginManager = PluginUtils.createPluginManagerFromRootFolder(configuration);
+        }
         ClientAuthenticationPlugin authenticatorPlugin =
-                discoverPlugin(clientAuthenticateProtocol, ClientAuthenticationPlugin.class, null);
+                discoverPlugin(
+                        clientAuthenticateProtocol,
+                        ClientAuthenticationPlugin.class,
+                        pluginManager);
         return () -> authenticatorPlugin.createClientAuthenticator(configuration);
     }
 

fluss-server/src/main/java/org/apache/fluss/server/tools/ClusterHealthReadinessCheck.java

@@ -19,6 +19,7 @@
 
 import org.apache.fluss.cluster.ServerNode;
 import org.apache.fluss.cluster.ServerType;
+import org.apache.fluss.config.ConfigOptions;
 import org.apache.fluss.config.Configuration;
 import org.apache.fluss.exception.UnsupportedVersionException;
 import org.apache.fluss.metrics.registry.MetricRegistryImpl;
@@ -230,6 +231,9 @@ private static Configuration buildConfiguration(String authString) {
         if (authString == null || authString.trim().isEmpty()) {
             return conf;
         }
+        // Enable plugin discovery so the client authenticator can discover authentication
+        // plugins shipped in the server's plugins/ directory.
+        conf.setBoolean(ConfigOptions.CLIENT_SECURITY_ENABLE_PLUGIN_DISCOVERY, true);
         for (String pair : authString.split(";")) {
             String trimmed = pair.trim();
             if (trimmed.isEmpty()) {