[kafka] Add SASL/PLAIN authentication

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

@@ -2645,6 +2645,22 @@ public class ConfigOptions {
                     .withDescription(
                             "The database for fluss kafka. The default database is `kafka`.");
 
+    public static final ConfigOption<String> KAFKA_DEFAULT_KEY_FORMAT =
+            key("kafka.default.key.format")
+                    .stringType()
+                    .defaultValue("raw")
+                    .withDescription(
+                            "The default format for Kafka record keys when a CreateTopics request does not specify fluss.key.format. "
+                                    + "Supported formats are raw and string.");
+
+    public static final ConfigOption<String> KAFKA_DEFAULT_VALUE_FORMAT =
+            key("kafka.default.value.format")
+                    .stringType()
+                    .defaultValue("raw")
+                    .withDescription(
+                            "The default format for Kafka record values when a CreateTopics request does not specify fluss.value.format. "
+                                    + "Supported formats are raw and string.");
+
     public static final ConfigOption<Duration> KAFKA_CONNECTION_MAX_IDLE_TIME =
             key("kafka.connection.max-idle-time")
                     .durationType()

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

@@ -51,6 +51,7 @@ public class ConfigurationUtils {
         "token",
         "basic-auth",
         "jaas.config",
+        "security.sasl.plain.credentials",
         "http-headers",
         "private.key",
         "private-key",

fluss-common/src/main/java/org/apache/fluss/security/auth/sasl/authenticator/SaslServerAuthenticator.java

@@ -46,6 +46,7 @@ public class SaslServerAuthenticator implements ServerAuthenticator {
     private static final String SERVER_AUTHENTICATOR_PREFIX = "security.sasl.";
     private final List<String> enabledMechanisms;
     private SaslServer saslServer;
+    private LoginManager loginManager;
     private final Map<String, String> configs;
 
     public SaslServerAuthenticator(Configuration configuration) {
@@ -60,6 +61,7 @@ public SaslServerAuthenticator(Configuration configuration) {
 
     @Override
     public void initialize(AuthenticateContext context) {
+        close();
         String mechanism = context.protocol();
         String listenerName = context.listenerName();
         String address = context.ipAddress();
@@ -102,16 +104,22 @@ public void initialize(AuthenticateContext context) {
 
         JaasContext jaasContext = JaasContext.loadServerContext(listenerName, dynamicJaasConfig);
 
+        LoginManager acquiredLoginManager = null;
         try {
-            LoginManager loginManager = LoginManager.acquireLoginManager(jaasContext);
-            saslServer =
+            acquiredLoginManager = LoginManager.acquireLoginManager(jaasContext);
+            SaslServer newSaslServer =
                     createSaslServer(
                             mechanism,
                             address,
                             configs,
-                            loginManager,
+                            acquiredLoginManager,
                             jaasContext.configurationEntries());
+            loginManager = acquiredLoginManager;
+            saslServer = newSaslServer;
         } catch (Exception e) {
+            if (acquiredLoginManager != null) {
+                acquiredLoginManager.release();
+            }
             throw new RuntimeException(e);
         }
     }
@@ -150,4 +158,21 @@ public boolean isCompleted() {
     public FlussPrincipal createPrincipal() {
         return new FlussPrincipal(saslServer.getAuthorizationID(), "User");
     }
+
+    @Override
+    public void close() {
+        if (saslServer != null) {
+            try {
+                saslServer.dispose();
+            } catch (SaslException e) {
+                LOG.debug("Failed to dispose SASL server.", e);
+            } finally {
+                saslServer = null;
+            }
+        }
+        if (loginManager != null) {
+            loginManager.release();
+            loginManager = null;
+        }
+    }
 }

fluss-common/src/main/java/org/apache/fluss/security/auth/sasl/jaas/LoginManager.java

@@ -44,6 +44,7 @@ public class LoginManager {
 
     private int refCount;
     private final String loginKey;
+    private final boolean dynamic;
 
     /**
      * A global cache of LoginManager instances keyed by static JAAS configuration names (e.g.,
@@ -68,11 +69,13 @@ public class LoginManager {
      * @throws LoginException if the login operation fails due to invalid credentials, missing
      *     modules, or misconfigured JAAS settings
      */
-    private LoginManager(JaasContext jaasContext, String loginKey) throws LoginException {
+    private LoginManager(JaasContext jaasContext, String loginKey, boolean dynamic)
+            throws LoginException {
         this.login = new DefaultLogin();
         login.configure(jaasContext.name(), jaasContext.getConfiguration());
         login.login();
         this.loginKey = loginKey;
+        this.dynamic = dynamic;
     }
 
     public Subject subject() {
@@ -90,14 +93,14 @@ public static LoginManager acquireLoginManager(JaasContext jaasContext) throws L
             if (jaasConfigValue != null) {
                 loginManager = DYNAMIC_INSTANCES.get(jaasConfigValue);
                 if (loginManager == null) {
-                    loginManager = new LoginManager(jaasContext, jaasConfigValue);
+                    loginManager = new LoginManager(jaasContext, jaasConfigValue, true);
                     DYNAMIC_INSTANCES.put(jaasConfigValue, loginManager);
                 }
             } else {
                 String jaasContextName = jaasContext.name();
                 loginManager = STATIC_INSTANCES.get(jaasContextName);
                 if (loginManager == null) {
-                    loginManager = new LoginManager(jaasContext, jaasContextName);
+                    loginManager = new LoginManager(jaasContext, jaasContextName, false);
                     STATIC_INSTANCES.put(jaasContextName, loginManager);
                 }
             }
@@ -117,6 +120,11 @@ public void release() {
             if (refCount == 0) {
                 throw new IllegalStateException("release() called on disposed " + this);
             } else if (refCount == 1) {
+                if (dynamic) {
+                    DYNAMIC_INSTANCES.remove(loginKey, this);
+                } else {
+                    STATIC_INSTANCES.remove(loginKey, this);
+                }
                 login.close();
             }
             --refCount;

fluss-common/src/main/java/org/apache/fluss/security/auth/sasl/jaas/SaslServerFactory.java

@@ -17,6 +17,7 @@
 
 package org.apache.fluss.security.auth.sasl.jaas;
 
+import org.apache.fluss.security.auth.sasl.plain.PlainSaslServer;
 import org.apache.fluss.security.auth.sasl.plain.PlainServerCallbackHandler;
 
 import org.slf4j.Logger;
@@ -61,6 +62,13 @@ public static SaslServer createSaslServer(
             }
 
             callbackHandler.configure(mechanism, configurationEntries);
+            // Construct Fluss's PLAIN server directly. Kafka clients register a provider with the
+            // same JVM provider name as Fluss's PLAIN provider. Delegating this server-side path
+            // to Sasl.createSaslServer would therefore make the selected callback type depend on
+            // class-loading order when both implementations share a process.
+            if (PlainSaslServer.PLAIN_MECHANISM.equals(mechanism)) {
+                return new PlainSaslServer(callbackHandler);
+            }
             SaslServer saslServer =
                     Subject.doAs(
                             loginManager.subject(),

fluss-common/src/main/java/org/apache/fluss/security/auth/sasl/plain/PlainSaslServerConfigManager.java

@@ -0,0 +1,206 @@
+/*
+ * 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.security.auth.sasl.plain;
+
+import org.apache.fluss.annotation.Internal;
+import org.apache.fluss.config.ConfigOptions;
+import org.apache.fluss.config.Configuration;
+import org.apache.fluss.config.cluster.ServerReconfigurable;
+import org.apache.fluss.exception.ConfigException;
+
+import java.util.LinkedHashMap;
+import java.util.Map;
+import java.util.Objects;
+import java.util.regex.Matcher;
+import java.util.regex.Pattern;
+
+import static org.apache.fluss.utils.Preconditions.checkNotNull;
+
+/**
+ * Manages the effective server configuration for SASL/PLAIN authentication.
+ *
+ * <p>{@link ConfigOptions#SERVER_SASL_CREDENTIALS} is a convenient credential map, while the SASL
+ * implementation consumes {@link ConfigOptions#SERVER_SASL_PLAIN_JAAS_CONFIG}. This manager
+ * validates the credential map and converts it to a JAAS configuration. Credentials from the map
+ * are merged with credentials in the initial JAAS configuration and take precedence when the same
+ * username is present in both sources.
+ *
+ * <p>The managed {@link Configuration} has a stable identity so authenticator suppliers that
+ * capture it during server startup see later credential updates. Callers must treat the returned
+ * configuration as read-only and perform updates through this manager.
+ */
+@Internal
+public final class PlainSaslServerConfigManager implements ServerReconfigurable {
+
+    private static final String PLAIN_CREDENTIALS_CONFIG =
+            ConfigOptions.SERVER_SASL_CREDENTIALS.key();
+
+    /** Pattern to match {@code user_<username>="<password>"} entries in a JAAS config. */
+    private static final Pattern JAAS_USER_PATTERN = Pattern.compile("user_(\\w+)=\"([^\"]*)\"");
+
+    /** Usernames become JAAS option keys, so only word characters are accepted. */
+    private static final Pattern VALID_USERNAME_PATTERN = Pattern.compile("\\w+");
+
+    /** Characters that would break the credential-map syntax or generated JAAS statement. */
+    private static final Pattern INVALID_PASSWORD_PATTERN =
+            Pattern.compile("[,:\"\\\\;]|[\\x00-\\x1F\\x7F]");
+
+    private final Map<String, String> initialPlainCredentialsFromJaasConfig;
+
+    private final Configuration configuration;
+
+    // Access is guarded by synchronized validate/reconfigure calls.
+    private Map<String, String> currentPlainCredentials;
+
+    /**
+     * Creates a manager from the initial server configuration.
+     *
+     * @param configuration initial server configuration
+     * @throws ConfigException if the configured credential map is invalid
+     */
+    public PlainSaslServerConfigManager(Configuration configuration) throws ConfigException {
+        checkNotNull(configuration, "configuration must not be null");
+        this.configuration = new Configuration(configuration);
+        this.initialPlainCredentialsFromJaasConfig = parseCredentialsFromJaasConfig(configuration);
+        validate(configuration);
+        reconfigure(configuration);
+    }
+
+    /**
+     * Returns the managed configuration containing the effective generated JAAS configuration.
+     *
+     * <p>The returned object has a stable identity and must be treated as read-only by callers.
+     *
+     * @return the managed effective configuration
+     */
+    public Configuration getConfiguration() {
+        return configuration;
+    }
+
+    @Override
+    public synchronized void validate(Configuration newConfiguration) throws ConfigException {
+        Map<String, String> newCredentials = readPlainCredentials(newConfiguration);
+        if (Objects.equals(newCredentials, currentPlainCredentials)) {
+            return;
+        }
+
+        if (newCredentials != null && !newCredentials.isEmpty()) {
+            int index = 0;
+            for (Map.Entry<String, String> credential : newCredentials.entrySet()) {
+                validateUsername(credential.getKey());
+                validatePassword(index, credential.getKey(), credential.getValue());
+                index++;
+            }
+        }
+
+        // Build the value during validation so reconfigure cannot fail after validation succeeds.
+        generateMergedJaasConfig(newCredentials);
+    }
+
+    @Override
+    public synchronized void reconfigure(Configuration newConfiguration) throws ConfigException {
+        // DynamicServerConfig may continue to reconfigure other components after a validation
+        // failure when it is applying a best-effort update. Defensively validate here as well so
+        // malformed credentials can never be rendered into an effective JAAS statement.
+        validate(newConfiguration);
+        Map<String, String> newCredentials = readPlainCredentials(newConfiguration);
+        if (Objects.equals(newCredentials, currentPlainCredentials)) {
+            return;
+        }
+
+        configuration.setString(
+                ConfigOptions.SERVER_SASL_PLAIN_JAAS_CONFIG,
+                generateMergedJaasConfig(newCredentials));
+        currentPlainCredentials = copyCredentials(newCredentials);
+    }
+
+    private static Map<String, String> readPlainCredentials(Configuration configuration)
+            throws ConfigException {
+        try {
+            return copyCredentials(configuration.get(ConfigOptions.SERVER_SASL_CREDENTIALS));
+        } catch (IllegalArgumentException | IllegalStateException e) {
+            throw new ConfigException(
+                    String.format(
+                            "Failed to parse %s: %s", PLAIN_CREDENTIALS_CONFIG, e.getMessage()),
+                    e);
+        }
+    }
+
+    private static Map<String, String> copyCredentials(Map<String, String> credentials) {
+        return credentials == null ? null : new LinkedHashMap<>(credentials);
+    }
+
+    private static void validateUsername(String username) throws ConfigException {
+        if (username == null || !VALID_USERNAME_PATTERN.matcher(username).matches()) {
+            throw new ConfigException(
+                    String.format(
+                            "%s: username '%s' contains invalid characters. "
+                                    + "Only letters, digits, and underscores are allowed.",
+                            PLAIN_CREDENTIALS_CONFIG, username));
+        }
+    }
+
+    private static void validatePassword(int index, String username, String password)
+            throws ConfigException {
+        if (password == null || password.isEmpty()) {
+            throw new ConfigException(
+                    String.format(
+                            "%s[%d]: password for user '%s' must not be empty.",
+                            PLAIN_CREDENTIALS_CONFIG, index, username));
+        }
+        if (INVALID_PASSWORD_PATTERN.matcher(password).find()) {
+            throw new ConfigException(
+                    String.format(
+                            "%s[%d]: password for user '%s' contains invalid characters. "
+                                    + "Commas, colons, quotes, semicolons, backslashes, and control characters are not allowed.",
+                            PLAIN_CREDENTIALS_CONFIG, index, username));
+        }
+    }
+
+    private String generateMergedJaasConfig(Map<String, String> newCredentials) {
+        Map<String, String> mergedCredentials =
+                new LinkedHashMap<>(initialPlainCredentialsFromJaasConfig);
+        if (newCredentials != null) {
+            mergedCredentials.putAll(newCredentials);
+        }
+
+        StringBuilder jaasConfig =
+                new StringBuilder(PlainLoginModule.class.getName()).append(" required");
+        for (Map.Entry<String, String> entry : mergedCredentials.entrySet()) {
+            jaasConfig
+                    .append(" user_")
+                    .append(entry.getKey())
+                    .append("=\"")
+                    .append(entry.getValue())
+                    .append('"');
+        }
+        return jaasConfig.append(';').toString();
+    }
+
+    private static Map<String, String> parseCredentialsFromJaasConfig(Configuration configuration) {
+        Map<String, String> credentials = new LinkedHashMap<>();
+        String existingJaas = configuration.getString(ConfigOptions.SERVER_SASL_PLAIN_JAAS_CONFIG);
+        if (existingJaas != null) {
+            Matcher matcher = JAAS_USER_PATTERN.matcher(existingJaas);
+            while (matcher.find()) {
+                credentials.put(matcher.group(1), matcher.group(2));
+            }
+        }
+        return credentials;
+    }
+}

fluss-common/src/test/java/org/apache/fluss/config/ConfigurationTest.java

@@ -512,6 +512,10 @@ void testHideSensitiveValue() {
                 .isEqualTo(Password.HIDDEN_CONTENT);
         assertThat(ConfigurationUtils.hideSensitiveValue("client.security.sasl.password", "pwd"))
                 .isEqualTo(Password.HIDDEN_CONTENT);
+        assertThat(
+                        ConfigurationUtils.hideSensitiveValue(
+                                ConfigOptions.SERVER_SASL_CREDENTIALS.key(), "admin:admin-secret"))
+                .isEqualTo(Password.HIDDEN_CONTENT);
         assertThat(ConfigurationUtils.hideSensitiveValue("plain.key", new Password("pwd")))
                 .isEqualTo(Password.HIDDEN_CONTENT);
         assertThat(ConfigurationUtils.hideSensitiveValue("plain.key", "value")).isEqualTo("value");