[server] Manage users for sasl/plain authentication via cluster properties.

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

@@ -287,6 +287,16 @@ public class ConfigOptions {
                                     + "and each super user should be specified in the format `principal_type:principal_name`, e.g., `User:admin;User:bob`. "
                                     + "This configuration is critical for defining administrative privileges in the system.");
 
+    public static final ConfigOption<Boolean> SECURITY_ACL_PRINCIPAL_IGNORE_CASE =
+            key("security.acl.principal.ignore-case")
+                    .booleanType()
+                    .defaultValue(false)
+                    .withDescription(
+                            "Whether to perform case-insensitive matching on principal name and type "
+                                    + "during ACL authorization checks. When set to true, principals "
+                                    + "such as 'User:Admin' and 'user:admin' will be treated as the same principal. "
+                                    + "Default is false for strict case-sensitive matching.");
+
     public static final ConfigOption<Integer> MAX_BUCKET_NUM =
             key("max.bucket.num")
                     .intType()
@@ -519,6 +529,26 @@ public class ConfigOptions {
                                     + "Each listener can be associated with a specific authentication protocol. "
                                     + "Listeners not included in the map will use PLAINTEXT by default, which does not require authentication.");
 
+    public static final ConfigOption<Map<String, String>> SERVER_SASL_CREDENTIALS =
+            key("security.sasl.plain.credentials")
+                    .mapType()
+                    .noDefaultValue()
+                    .withDescription(
+                            "Map of user credentials for SASL/PLAIN authentication in 'username:password' format. "
+                                    + "For example: 'admin:admin-secret,bob:bob-secret'. "
+                                    + "This is syntactic sugar that auto-generates the JAAS config string.");
+
+    public static final ConfigOption<String> SERVER_SASL_PLAIN_JAAS_CONFIG =
+            key("security.sasl.plain.jaas.config")
+                    .stringType()
+                    .noDefaultValue()
+                    .withDescription(
+                            "JAAS configuration string for server-side SASL/PLAIN authentication. "
+                                    + "The value should use PlainLoginModule and define users with "
+                                    + "'user_<username>=\"<password>\"' options. This option is generated "
+                                    + "from 'security.sasl.plain.credentials' when that credential map is set, "
+                                    + "and can also be configured directly for compatibility.");
+
     public static final ConfigOption<Integer> TABLET_SERVER_ID =
             key("tablet-server.id")
                     .intType()

fluss-common/src/main/java/org/apache/fluss/security/acl/FlussPrincipal.java

@@ -79,6 +79,34 @@ public int hashCode() {
         return Objects.hash(name, type);
     }
 
+    /**
+     * Tests whether this principal matches another principal, with optional case-insensitive
+     * comparison for both name and type.
+     *
+     * @param other the other principal to match against
+     * @param ignoreCase if {@code true}, name and type are compared case-insensitively
+     * @return {@code true} if the principals match
+     */
+    public boolean matches(FlussPrincipal other, boolean ignoreCase) {
+        if (other == null) {
+            return false;
+        }
+        if (!ignoreCase) {
+            return this.equals(other);
+        }
+        return equalsIgnoreCase(name, other.name) && equalsIgnoreCase(type, other.type);
+    }
+
+    private static boolean equalsIgnoreCase(String a, String b) {
+        if (a == b) {
+            return true;
+        }
+        if (a == null || b == null) {
+            return false;
+        }
+        return a.equalsIgnoreCase(b);
+    }
+
     @Override
     public String toString() {
         return "FlussPrincipal{" + "name='" + name + '\'' + ", type='" + type + '\'' + '}';

fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/procedure/AppendClusterConfigsProcedure.java

@@ -0,0 +1,67 @@
+/*
+ * 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.flink.procedure;
+
+import org.apache.fluss.config.cluster.AlterConfigOpType;
+
+import org.apache.flink.table.annotation.ArgumentHint;
+import org.apache.flink.table.annotation.DataTypeHint;
+import org.apache.flink.table.annotation.ProcedureHint;
+import org.apache.flink.table.procedure.ProcedureContext;
+
+/**
+ * Procedure to append values to collection-type cluster configurations dynamically.
+ *
+ * <p>This procedure appends new values to existing list-type or map-type configurations. The APPEND
+ * operation only works on collection configurations (e.g., {@code
+ * security.sasl.plain.credentials}). The changes are:
+ *
+ * <ul>
+ *   <li>Validated by the CoordinatorServer before persistence
+ *   <li>Persisted in ZooKeeper for durability
+ *   <li>Applied to all relevant servers (Coordinator and TabletServers)
+ *   <li>Survives server restarts
+ * </ul>
+ *
+ * <p>Usage examples:
+ *
+ * <pre>
+ * -- Append a user to the SASL credentials map
+ * CALL sys.append_cluster_configs('security.sasl.plain.credentials', 'bob:bob-secret');
+ *
+ * -- Append multiple key-value pairs at one time
+ * CALL sys.append_cluster_configs(
+ *     'security.sasl.plain.credentials',
+ *     'bob:bob-secret',
+ *     'security.sasl.plain.credentials',
+ *     'alice:alice-secret');
+ * </pre>
+ *
+ * <p><b>Note:</b> APPEND operations are only supported for list-type or map-type configuration
+ * keys. The server will reject the change if the configuration key is not a collection type.
+ */
+public class AppendClusterConfigsProcedure extends CollectionClusterConfigsProcedureBase {
+
+    @ProcedureHint(
+            argument = {@ArgumentHint(name = "config_pairs", type = @DataTypeHint("STRING"))},
+            isVarArgs = true)
+    public String[] call(ProcedureContext context, String... configPairs) throws Exception {
+        return alterCollectionClusterConfigs(
+                configPairs, AlterConfigOpType.APPEND, "appended", "to", "append");
+    }
+}

fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/procedure/CollectionClusterConfigsProcedureBase.java

@@ -0,0 +1,79 @@
+/*
+ * 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.flink.procedure;
+
+import org.apache.fluss.config.cluster.AlterConfig;
+import org.apache.fluss.config.cluster.AlterConfigOpType;
+
+import java.util.ArrayList;
+import java.util.List;
+
+/** Base procedure for modifying collection-type cluster configurations. */
+abstract class CollectionClusterConfigsProcedureBase extends ProcedureBase {
+
+    protected String[] alterCollectionClusterConfigs(
+            String[] configPairs,
+            AlterConfigOpType opType,
+            String successVerb,
+            String successPreposition,
+            String failureVerb)
+            throws Exception {
+        try {
+            if (configPairs == null || configPairs.length == 0) {
+                throw new IllegalArgumentException(
+                        "config_pairs cannot be null or empty. "
+                                + "Please specify valid configuration pairs.");
+            }
+
+            if (configPairs.length % 2 != 0) {
+                throw new IllegalArgumentException(
+                        "config_pairs must be set in pairs. "
+                                + "Please specify valid configuration pairs.");
+            }
+
+            List<AlterConfig> alterConfigs = new ArrayList<>();
+            List<String> resultMessages = new ArrayList<>();
+
+            for (int i = 0; i < configPairs.length; i += 2) {
+                String configKey = configPairs[i].trim();
+                if (configKey.isEmpty()) {
+                    throw new IllegalArgumentException(
+                            "Config key cannot be null or empty. "
+                                    + "Please specify a valid configuration key.");
+                }
+                String configValue = configPairs[i + 1];
+
+                alterConfigs.add(new AlterConfig(configKey, configValue, opType));
+                resultMessages.add(
+                        String.format(
+                                "Successfully %s '%s' %s configuration '%s'. ",
+                                successVerb, configValue, successPreposition, configKey));
+            }
+
+            admin.alterClusterConfigs(alterConfigs).get();
+
+            return resultMessages.toArray(new String[0]);
+        } catch (IllegalArgumentException e) {
+            throw e;
+        } catch (Exception e) {
+            throw new RuntimeException(
+                    String.format("Failed to %s cluster config: %s", failureVerb, e.getMessage()),
+                    e);
+        }
+    }
+}

fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/procedure/ProcedureManager.java

@@ -73,6 +73,9 @@ private enum ProcedureEnum {
         SET_CLUSTER_CONFIGS("sys.set_cluster_configs", SetClusterConfigsProcedure.class),
         GET_CLUSTER_CONFIGS("sys.get_cluster_configs", GetClusterConfigsProcedure.class),
         RESET_CLUSTER_CONFIGS("sys.reset_cluster_configs", ResetClusterConfigsProcedure.class),
+        APPEND_CLUSTER_CONFIGS("sys.append_cluster_configs", AppendClusterConfigsProcedure.class),
+        SUBTRACT_CLUSTER_CONFIGS(
+                "sys.subtract_cluster_configs", SubtractClusterConfigsProcedure.class),
         ADD_SERVER_TAG("sys.add_server_tag", AddServerTagProcedure.class),
         REMOVE_SERVER_TAG("sys.remove_server_tag", RemoveServerTagProcedure.class),
         REBALANCE("sys.rebalance", RebalanceProcedure.class),

fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/procedure/SubtractClusterConfigsProcedure.java

@@ -0,0 +1,71 @@
+/*
+ * 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.flink.procedure;
+
+import org.apache.fluss.config.cluster.AlterConfigOpType;
+
+import org.apache.flink.table.annotation.ArgumentHint;
+import org.apache.flink.table.annotation.DataTypeHint;
+import org.apache.flink.table.annotation.ProcedureHint;
+import org.apache.flink.table.procedure.ProcedureContext;
+
+/**
+ * Procedure to subtract (remove) values from collection-type cluster configurations dynamically.
+ *
+ * <p>This procedure removes values from existing collection configurations. For list-type
+ * configurations, SUBTRACT removes the exact list value. For map-type configurations, SUBTRACT
+ * removes the entry by map key: the supplied value must still be a valid {@code key:value} pair,
+ * but only the key is used to find the entry to remove. The SUBTRACT operation only works on
+ * collection configurations (e.g., {@code security.sasl.plain.credentials}). If the collection
+ * becomes empty after subtraction, the configuration key is removed entirely. The changes are:
+ *
+ * <ul>
+ *   <li>Validated by the CoordinatorServer before persistence
+ *   <li>Persisted in ZooKeeper for durability
+ *   <li>Applied to all relevant servers (Coordinator and TabletServers)
+ *   <li>Survives server restarts
+ * </ul>
+ *
+ * <p>Usage examples:
+ *
+ * <pre>
+ * -- Remove user "bob" from the SASL credentials map. For map configs, only "bob" is matched.
+ * CALL sys.subtract_cluster_configs('security.sasl.plain.credentials', 'bob:any-secret');
+ *
+ * -- Remove multiple key-value pairs at one time
+ * CALL sys.subtract_cluster_configs(
+ *     'security.sasl.plain.credentials',
+ *     'bob:bob-secret',
+ *     'security.sasl.plain.credentials',
+ *     'alice:alice-secret');
+ * </pre>
+ *
+ * <p><b>Note:</b> SUBTRACT operations are only supported for list-type or map-type configuration
+ * keys. The server will reject the change if the configuration key is not a collection type.
+ * Subtracting a list value or map key that does not exist in the collection is a no-op.
+ */
+public class SubtractClusterConfigsProcedure extends CollectionClusterConfigsProcedureBase {
+
+    @ProcedureHint(
+            argument = {@ArgumentHint(name = "config_pairs", type = @DataTypeHint("STRING"))},
+            isVarArgs = true)
+    public String[] call(ProcedureContext context, String... configPairs) throws Exception {
+        return alterCollectionClusterConfigs(
+                configPairs, AlterConfigOpType.SUBTRACT, "subtracted", "from", "subtract");
+    }
+}