This is an automated email from the ASF dual-hosted git repository.
wuchong pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/fluss.git
The following commit(s) were added to refs/heads/main by this push:
new 3acc1c14a [server] Manage multiple users for sasl/plain authentication
via cluster properties (#3047)
3acc1c14a is described below
commit 3acc1c14acf72351bff13f171ea287a0f074076a
Author: Hongshun Wang <[email protected]>
AuthorDate: Fri Jul 10 14:07:35 2026 +0800
[server] Manage multiple users for sasl/plain authentication via cluster
properties (#3047)
---
.../org/apache/fluss/config/ConfigOptions.java | 30 ++
.../apache/fluss/security/acl/FlussPrincipal.java | 28 ++
.../procedure/AppendClusterConfigsProcedure.java | 67 +++++
.../CollectionClusterConfigsProcedureBase.java | 79 ++++++
.../fluss/flink/procedure/ProcedureManager.java | 3 +
.../procedure/SubtractClusterConfigsProcedure.java | 71 +++++
.../flink/procedure/FlinkProcedureITCase.java | 211 +++++++++++++-
.../main/java/org/apache/fluss/rpc/RpcServer.java | 7 +
.../rpc/netty/server/FlussProtocolPlugin.java | 164 ++++++++++-
.../apache/fluss/rpc/netty/server/NettyServer.java | 9 +
.../rpc/netty/authenticate/AuthenticationTest.java | 6 +-
.../authenticate/SaslAuthenticationITCase.java | 311 +++++++++++++++++++++
.../apache/fluss/server/DynamicConfigManager.java | 191 ++++++++++++-
.../apache/fluss/server/DynamicServerConfig.java | 32 ++-
.../fluss/server/authorizer/DefaultAuthorizer.java | 21 +-
.../apache/fluss/server/config/ConfigRedactor.java | 34 +++
.../fluss/server/config/ConfigRedactors.java | 39 +++
.../fluss/server/config/MapConfigRedactor.java | 61 ++++
.../fluss/server/config/ValueConfigRedactor.java | 42 +++
.../server/coordinator/CoordinatorServer.java | 18 +-
.../apache/fluss/server/tablet/TabletServer.java | 28 +-
.../fluss/server/DynamicConfigChangeTest.java | 243 ++++++++++++++++
.../server/authorizer/DefaultAuthorizerTest.java | 35 ++-
website/docs/engine-flink/procedures.md | 74 ++++-
.../maintenance/operations/updating-configs.md | 2 +-
25 files changed, 1752 insertions(+), 54 deletions(-)
diff --git
a/fluss-common/src/main/java/org/apache/fluss/config/ConfigOptions.java
b/fluss-common/src/main/java/org/apache/fluss/config/ConfigOptions.java
index 2f6617b20..73eccf1b7 100644
--- a/fluss-common/src/main/java/org/apache/fluss/config/ConfigOptions.java
+++ b/fluss-common/src/main/java/org/apache/fluss/config/ConfigOptions.java
@@ -298,6 +298,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()
@@ -533,6 +543,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()
diff --git
a/fluss-common/src/main/java/org/apache/fluss/security/acl/FlussPrincipal.java
b/fluss-common/src/main/java/org/apache/fluss/security/acl/FlussPrincipal.java
index 6084ec1a2..7a60e9c7b 100644
---
a/fluss-common/src/main/java/org/apache/fluss/security/acl/FlussPrincipal.java
+++
b/fluss-common/src/main/java/org/apache/fluss/security/acl/FlussPrincipal.java
@@ -79,6 +79,34 @@ public class FlussPrincipal implements Principal {
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
+ '\'' + '}';
diff --git
a/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/procedure/AppendClusterConfigsProcedure.java
b/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/procedure/AppendClusterConfigsProcedure.java
new file mode 100644
index 000000000..2ccdc820f
--- /dev/null
+++
b/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");
+ }
+}
diff --git
a/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/procedure/CollectionClusterConfigsProcedureBase.java
b/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/procedure/CollectionClusterConfigsProcedureBase.java
new file mode 100644
index 000000000..8479a2847
--- /dev/null
+++
b/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);
+ }
+ }
+}
diff --git
a/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/procedure/ProcedureManager.java
b/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/procedure/ProcedureManager.java
index fc2632e85..654f707c2 100644
---
a/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/procedure/ProcedureManager.java
+++
b/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/procedure/ProcedureManager.java
@@ -73,6 +73,9 @@ public class ProcedureManager {
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),
diff --git
a/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/procedure/SubtractClusterConfigsProcedure.java
b/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/procedure/SubtractClusterConfigsProcedure.java
new file mode 100644
index 000000000..9e6de7fb3
--- /dev/null
+++
b/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");
+ }
+}
diff --git
a/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/procedure/FlinkProcedureITCase.java
b/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/procedure/FlinkProcedureITCase.java
index 8b2817e32..5dfaf3f7f 100644
---
a/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/procedure/FlinkProcedureITCase.java
+++
b/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/procedure/FlinkProcedureITCase.java
@@ -63,6 +63,7 @@ import static
org.apache.fluss.server.testutils.FlussClusterExtension.BUILTIN_DA
import static org.apache.fluss.testutils.DataTestUtils.row;
import static org.apache.fluss.testutils.common.CommonTestUtils.retry;
import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatNoException;
import static org.assertj.core.api.Assertions.assertThatThrownBy;
/** ITCase for Flink Procedure. */
@@ -139,6 +140,8 @@ public abstract class FlinkProcedureITCase {
"+I[sys.list_acl]",
"+I[sys.set_cluster_configs]",
"+I[sys.reset_cluster_configs]",
+ "+I[sys.append_cluster_configs]",
+ "+I[sys.subtract_cluster_configs]",
"+I[sys.add_server_tag]",
"+I[sys.remove_server_tag]",
"+I[sys.rebalance]",
@@ -783,6 +786,209 @@ public abstract class FlinkProcedureITCase {
});
}
+ @Test
+ void testAddAndDeleteUser() throws Exception {
+ String bobBootstrapServers =
+ String.join(
+ ",",
+ FLUSS_CLUSTER_EXTENSION
+ .getClientConfig("CLIENT")
+ .get(ConfigOptions.BOOTSTRAP_SERVERS));
+ String bobCatalog = "bob_catalog";
+ String createCatalogDDL =
+ String.format(
+ "create catalog %s with ("
+ + "'type' = 'fluss', "
+ + "'bootstrap.servers' = '%s', "
+ + "'client.security.protocol' = 'sasl', "
+ + "'client.security.sasl.mechanism' = 'PLAIN',
"
+ + "'client.security.sasl.username' = 'bob', "
+ + "'client.security.sasl.password' =
'bob_pass'"
+ + ")",
+ bobCatalog, bobBootstrapServers);
+
+ assertThatThrownBy(() -> tEnv.executeSql(createCatalogDDL).await())
+ .hasMessageContaining("Invalid username or password");
+
+ // Step 1: Add user "bob" via append_cluster_configs
+ try (CloseableIterator<Row> resultIterator =
+ tEnv.executeSql(
+ String.format(
+ "Call
%s.sys.append_cluster_configs('%s', 'bob:bob_pass')",
+ CATALOG_NAME,
ConfigOptions.SERVER_SASL_CREDENTIALS.key()))
+ .collect()) {
+ List<Row> results = CollectionUtil.iteratorToList(resultIterator);
+ assertThat(results).hasSize(1);
+ assertThat(results.get(0).getField(0))
+ .asString()
+ .contains("Successfully appended")
+ .contains(ConfigOptions.SERVER_SASL_CREDENTIALS.key());
+ }
+
+ // Verify user "bob" was added
+ try (CloseableIterator<Row> resultIterator =
+ tEnv.executeSql(
+ String.format(
+ "Call
%s.sys.get_cluster_configs('%s')",
+ CATALOG_NAME,
ConfigOptions.SERVER_SASL_CREDENTIALS.key()))
+ .collect()) {
+ List<Row> results = CollectionUtil.iteratorToList(resultIterator);
+ assertThat(results).hasSize(1);
+
assertThat(results.stream().map(Row::toString).collect(Collectors.toList()))
+ .containsExactly(
+ "+I[security.sasl.plain.credentials,
root:******,guest:******,bob:******, DYNAMIC_SERVER_CONFIG]");
+ }
+
+ // Verify "bob" can authenticate by creating a catalog with bob's
credentials.
+ // Use retry to wait for ZK config notification to propagate to all
TabletServers.
+ retry(
+ Duration.ofSeconds(30),
+ () ->
+ assertThatNoException()
+ .isThrownBy(() ->
tEnv.executeSql(createCatalogDDL).await()));
+
+ // Grant bob DESCRIBE permission on cluster so bob can query configs
+ tEnv.executeSql(
+ String.format(
+ "Call %s.sys.add_acl('CLUSTER', 'ALLOW',
'User:bob', 'DESCRIBE', '*')",
+ CATALOG_NAME))
+ .await();
+
+ // Bob should be able to get cluster configs
+ try (CloseableIterator<Row> resultIterator =
+ tEnv.executeSql(
+ String.format(
+ "Call
%s.sys.get_cluster_configs('%s')",
+ bobCatalog,
ConfigOptions.SERVER_SASL_CREDENTIALS.key()))
+ .collect()) {
+ List<Row> results = CollectionUtil.iteratorToList(resultIterator);
+
assertThat(results.stream().map(Row::toString).collect(Collectors.toList()))
+ .containsExactly(
+ "+I[security.sasl.plain.credentials,
root:******,guest:******,bob:******, DYNAMIC_SERVER_CONFIG]");
+ }
+ tEnv.executeSql("drop catalog " + bobCatalog);
+
+ // Step 2: Delete user "bob" via subtract_cluster_configs
+ tEnv.executeSql(
+ String.format(
+ "Call %s.sys.subtract_cluster_configs('%s',
'bob:bob_pass')",
+ CATALOG_NAME,
ConfigOptions.SERVER_SASL_CREDENTIALS.key()))
+ .await();
+
+ // Verify "bob" was deleted from config
+ try (CloseableIterator<Row> resultIterator =
+ tEnv.executeSql(
+ String.format(
+ "Call
%s.sys.get_cluster_configs('%s')",
+ CATALOG_NAME,
ConfigOptions.SERVER_SASL_CREDENTIALS.key()))
+ .collect()) {
+ List<Row> results = CollectionUtil.iteratorToList(resultIterator);
+ // After subtracting the only dynamically-added entry, the config
may be empty
+
assertThat(results.stream().map(Row::toString).collect(Collectors.toList()))
+ .containsExactly(
+ "+I[security.sasl.plain.credentials,
root:******,guest:******, DYNAMIC_SERVER_CONFIG]");
+ }
+
+ // Verify "bob" can no longer authenticate.
+ // Use retry to wait for ZK config notification to propagate to all
TabletServers.
+ retry(
+ Duration.ofSeconds(30),
+ () ->
+ assertThatThrownBy(() ->
tEnv.executeSql(createCatalogDDL).await())
+ .hasMessageContaining("Invalid username or
password"));
+ // Cleanup: remove bob's ACL
+ tEnv.executeSql(
+ String.format(
+ "Call %s.sys.drop_acl('CLUSTER', 'ALLOW',
'User:bob', 'DESCRIBE', '*')",
+ CATALOG_NAME))
+ .await();
+ // Try to append a map entry with the same key as the existing "root"
entry
+ assertThatThrownBy(
+ () ->
+ tEnv.executeSql(
+ String.format(
+ "Call
%s.sys.append_cluster_configs('%s', 'root:another-pass')",
+ CATALOG_NAME,
+
ConfigOptions.SERVER_SASL_CREDENTIALS
+ .key()))
+ .await())
+ .hasMessageContaining("duplicate map entry keys")
+ .hasMessageContaining("root");
+
+ // Try to append a user with no colon (invalid format)
+ assertThatThrownBy(
+ () ->
+ tEnv.executeSql(
+ String.format(
+ "Call
%s.sys.append_cluster_configs('%s', 'usernameonly')",
+ CATALOG_NAME,
+
ConfigOptions.SERVER_SASL_CREDENTIALS
+ .key()))
+ .await())
+ .hasMessageContaining("Invalid map entry");
+
+ // Try to append a user with invalid username characters (@)
+ assertThatThrownBy(
+ () ->
+ tEnv.executeSql(
+ String.format(
+ "Call
%s.sys.append_cluster_configs('%s', 'user@domain:pass')",
+ CATALOG_NAME,
+
ConfigOptions.SERVER_SASL_CREDENTIALS
+ .key()))
+ .await())
+ .hasMessageContaining("contains invalid characters");
+
+ // Try to append a user with invalid username characters (-)
+ assertThatThrownBy(
+ () ->
+ tEnv.executeSql(
+ String.format(
+ "Call
%s.sys.append_cluster_configs('%s', 'user-name:pass')",
+ CATALOG_NAME,
+
ConfigOptions.SERVER_SASL_CREDENTIALS
+ .key()))
+ .await())
+ .hasMessageContaining("contains invalid characters");
+
+ // Try to append a user with double-quote in password (breaks JAAS
value)
+ assertThatThrownBy(
+ () ->
+ tEnv.executeSql(
+ String.format(
+ "Call
%s.sys.append_cluster_configs('%s', 'eve:pass\"word')",
+ CATALOG_NAME,
+
ConfigOptions.SERVER_SASL_CREDENTIALS
+ .key()))
+ .await())
+ .hasMessageContaining("contains invalid characters");
+
+ // Try to append a user with semicolon in password (breaks JAAS
statement)
+ assertThatThrownBy(
+ () ->
+ tEnv.executeSql(
+ String.format(
+ "Call
%s.sys.append_cluster_configs('%s', 'eve:pass;word')",
+ CATALOG_NAME,
+
ConfigOptions.SERVER_SASL_CREDENTIALS
+ .key()))
+ .await())
+ .hasMessageContaining("contains invalid characters");
+
+ // Subtract a non-existent user - should be a no-op (no error)
+ tEnv.executeSql(
+ String.format(
+ "Call %s.sys.subtract_cluster_configs('%s',
'nonexistent:whatever')",
+ CATALOG_NAME,
ConfigOptions.SERVER_SASL_CREDENTIALS.key()))
+ .await();
+
+ tEnv.executeSql(
+ String.format(
+ "Call %s.sys.subtract_cluster_configs('%s',
'special_user:P@ss!w0rd#$&*()')",
+ CATALOG_NAME,
ConfigOptions.SERVER_SASL_CREDENTIALS.key()))
+ .await();
+ }
+
@Test
void testDropKvSnapshotLeaseProcedure() throws Exception {
tEnv.executeSql(
@@ -838,10 +1044,7 @@ public abstract class FlinkProcedureITCase {
conf.setString(ConfigOptions.SERVER_SECURITY_PROTOCOL_MAP.key(),
"CLIENT:sasl");
conf.setString("security.sasl.enabled.mechanisms", "plain");
conf.setString(
- "security.sasl.plain.jaas.config",
- "org.apache.fluss.security.auth.sasl.plain.PlainLoginModule
required "
- + " user_root=\"password\" "
- + " user_guest=\"password2\";");
+ ConfigOptions.SERVER_SASL_CREDENTIALS.key(),
"root:password,guest:passwords");
conf.set(ConfigOptions.SUPER_USERS, "User:root");
conf.set(ConfigOptions.AUTHORIZER_ENABLED, true);
return conf;
diff --git a/fluss-rpc/src/main/java/org/apache/fluss/rpc/RpcServer.java
b/fluss-rpc/src/main/java/org/apache/fluss/rpc/RpcServer.java
index bae71acaa..317a12380 100644
--- a/fluss-rpc/src/main/java/org/apache/fluss/rpc/RpcServer.java
+++ b/fluss-rpc/src/main/java/org/apache/fluss/rpc/RpcServer.java
@@ -19,12 +19,14 @@ package org.apache.fluss.rpc;
import org.apache.fluss.cluster.Endpoint;
import org.apache.fluss.config.Configuration;
+import org.apache.fluss.config.cluster.ServerReconfigurable;
import org.apache.fluss.metrics.groups.MetricGroup;
import org.apache.fluss.rpc.netty.server.NettyServer;
import org.apache.fluss.rpc.netty.server.RequestsMetrics;
import org.apache.fluss.utils.AutoCloseableAsync;
import java.io.IOException;
+import java.util.Collections;
import java.util.List;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ScheduledExecutorService;
@@ -79,4 +81,9 @@ public interface RpcServer extends AutoCloseableAsync {
* @return The RPC server provided scheduled executor
*/
ScheduledExecutorService getScheduledExecutor();
+
+ /** Returns server components that should receive dynamic config updates.
*/
+ default List<ServerReconfigurable> getServerReconfigurables() {
+ return Collections.emptyList();
+ }
}
diff --git
a/fluss-rpc/src/main/java/org/apache/fluss/rpc/netty/server/FlussProtocolPlugin.java
b/fluss-rpc/src/main/java/org/apache/fluss/rpc/netty/server/FlussProtocolPlugin.java
index 54bf34ecf..25ae18941 100644
---
a/fluss-rpc/src/main/java/org/apache/fluss/rpc/netty/server/FlussProtocolPlugin.java
+++
b/fluss-rpc/src/main/java/org/apache/fluss/rpc/netty/server/FlussProtocolPlugin.java
@@ -17,10 +17,11 @@
package org.apache.fluss.rpc.netty.server;
-import org.apache.fluss.annotation.VisibleForTesting;
import org.apache.fluss.cluster.ServerType;
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 org.apache.fluss.rpc.RpcGatewayService;
import org.apache.fluss.rpc.protocol.ApiManager;
import org.apache.fluss.rpc.protocol.NetworkProtocolPlugin;
@@ -28,15 +29,47 @@ import org.apache.fluss.security.auth.AuthenticationFactory;
import org.apache.fluss.security.auth.PlainTextAuthenticationPlugin;
import org.apache.fluss.shaded.netty4.io.netty.channel.ChannelHandler;
+import java.util.LinkedHashMap;
import java.util.List;
+import java.util.Map;
+import java.util.Objects;
import java.util.Optional;
+import java.util.regex.Matcher;
+import java.util.regex.Pattern;
/** Build-in protocol plugin for Fluss. */
-public class FlussProtocolPlugin implements NetworkProtocolPlugin {
+public class FlussProtocolPlugin implements NetworkProtocolPlugin,
ServerReconfigurable {
+
+ private static final String PLAIN_CREDENTIALS_CONFIG =
+ ConfigOptions.SERVER_SASL_CREDENTIALS.key();
+
+ /** Pattern to match {@code user_<username>="<password>"} entries in JAAS
config strings. */
+ private static final Pattern JAAS_USER_PATTERN =
Pattern.compile("user_(\\w+)=\"([^\"]*)\"");
+
+ /**
+ * Valid username pattern. Only letters, digits, and underscores are
allowed because the
+ * username is used as part of the JAAS option key {@code user_<username>}.
+ */
+ private static final Pattern VALID_USERNAME_PATTERN =
Pattern.compile("\\w+");
+
+ /**
+ * Characters forbidden in passwords. These would break the map format or
the generated JAAS
+ * config string: comma (entry separator), colon (key-value separator),
double-quote (JAAS value
+ * delimiter), semicolon (JAAS statement terminator), backslash (escape
char), and control
+ * characters.
+ */
+ private static final Pattern INVALID_PASSWORD_PATTERN =
+ Pattern.compile("[,:\"\\\\;]|[\\x00-\\x1F\\x7F]");
+
private final ApiManager apiManager;
private final List<String> listeners;
private final RequestsMetrics requestsMetrics;
private Configuration conf;
+ /** Initial credentials from `security.sasl.plain.jaas.config`. */
+ private Map<String, String> initialPlainCredentialsFromJaasConfig;
+
+ /** Current config `security.sasl.plain.credentials`. */
+ private Map<String, String> currentPlainCredentials;
public FlussProtocolPlugin(
ServerType serverType, List<String> listeners, RequestsMetrics
requestsMetrics) {
@@ -52,7 +85,9 @@ public class FlussProtocolPlugin implements
NetworkProtocolPlugin {
@Override
public void setup(Configuration conf) {
- this.conf = conf;
+ this.conf = new Configuration(conf);
+ this.initialPlainCredentialsFromJaasConfig =
parseCredentialsFromJaasConfig(conf);
+ enrichWithJaasConfig(conf);
}
@Override
@@ -72,7 +107,7 @@ public class FlussProtocolPlugin implements
NetworkProtocolPlugin {
conf.get(ConfigOptions.NETTY_CONNECTION_MAX_IDLE_TIME).getSeconds(),
(int)
conf.get(ConfigOptions.NETTY_SERVER_MAX_REQUEST_SIZE).getBytes(),
Optional.ofNullable(
-
AuthenticationFactory.loadServerAuthenticatorSuppliers(conf)
+
AuthenticationFactory.loadServerAuthenticatorSuppliers(this.conf)
.get(listenerName))
.orElse(PlainTextAuthenticationPlugin.PlainTextServerAuthenticator::new));
}
@@ -82,8 +117,123 @@ public class FlussProtocolPlugin implements
NetworkProtocolPlugin {
return new FlussRequestHandler(service);
}
- @VisibleForTesting
- ApiManager getApiManager() {
- return apiManager;
+ // --- ServerReconfigurable ---
+
+ @Override
+ public void validate(Configuration newConfig) throws ConfigException {
+ Map<String, String> newCredentials = readPlainCredentials(newConfig);
+ 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++;
+ }
+ }
+
+ // Generate the merged JAAS config value to ensure it is valid.
+ generateMergedJaasConfig(newCredentials);
+ }
+
+ @Override
+ public void reconfigure(Configuration newConfig) throws ConfigException {
+ enrichWithJaasConfig(newConfig);
+ }
+
+ /**
+ * Enriches the plugin's configuration with a generated JAAS config string
by merging:
+ *
+ * <ol>
+ * <li>Existing credentials parsed from the current {@code
security.sasl.plain.jaas.config}
+ * <li>New credentials from the {@code security.sasl.plain.credentials}
map in {@code
+ * newConfig}
+ * </ol>
+ *
+ * <p>New credentials take priority when a username exists in both
sources. If the credentials
+ * map is not present in {@code newConfig}, the configuration is returned
unchanged.
+ */
+ private void enrichWithJaasConfig(Configuration newConfig) throws
ConfigException {
+ Map<String, String> newCredentials = readPlainCredentials(newConfig);
+ if (Objects.equals(newCredentials, currentPlainCredentials)) {
+ return;
+ }
+
+ conf.setString(
+ ConfigOptions.SERVER_SASL_PLAIN_JAAS_CONFIG,
+ generateMergedJaasConfig(newCredentials));
+ currentPlainCredentials = newCredentials;
+ }
+
+ private static Map<String, String> readPlainCredentials(Configuration
config)
+ throws ConfigException {
+ try {
+ return config.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 void validateUsername(String username) throws
ConfigException {
+ if (!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 ||
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));
+ }
+ }
+
+ /**
+ * Generates the merged JAAS config string by combining existing
credentials from the current
+ * {@code security.sasl.plain.jaas.config} with the given new credentials
map. New credentials
+ * take priority on username conflict.
+ *
+ * @param newCredentials map of username to password from
SERVER_SASL_CREDENTIALS
+ * @return the generated JAAS config string
+ */
+ private String generateMergedJaasConfig(Map<String, String>
newCredentials) {
+ Map<String, String> mergedCredentials =
+ new LinkedHashMap<>(initialPlainCredentialsFromJaasConfig);
+ if (newCredentials != null) {
+ mergedCredentials.putAll(newCredentials);
+ }
+
+ StringBuilder sb =
+ new StringBuilder(
+
"org.apache.fluss.security.auth.sasl.plain.PlainLoginModule required");
+ for (Map.Entry<String, String> entry : mergedCredentials.entrySet()) {
+ sb.append(String.format(" user_%s=\"%s\"", entry.getKey(),
entry.getValue()));
+ }
+ sb.append(";");
+ return sb.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;
}
}
diff --git
a/fluss-rpc/src/main/java/org/apache/fluss/rpc/netty/server/NettyServer.java
b/fluss-rpc/src/main/java/org/apache/fluss/rpc/netty/server/NettyServer.java
index 7f2198566..03d798fb3 100644
--- a/fluss-rpc/src/main/java/org/apache/fluss/rpc/netty/server/NettyServer.java
+++ b/fluss-rpc/src/main/java/org/apache/fluss/rpc/netty/server/NettyServer.java
@@ -21,6 +21,7 @@ import org.apache.fluss.cluster.Endpoint;
import org.apache.fluss.cluster.ServerType;
import org.apache.fluss.config.ConfigOptions;
import org.apache.fluss.config.Configuration;
+import org.apache.fluss.config.cluster.ServerReconfigurable;
import org.apache.fluss.metrics.groups.MetricGroup;
import org.apache.fluss.rpc.RpcGateway;
import org.apache.fluss.rpc.RpcGatewayService;
@@ -272,6 +273,14 @@ public final class NettyServer implements RpcServer {
protocolName, protocols.keySet()));
}
+ @Override
+ public List<ServerReconfigurable> getServerReconfigurables() {
+ return protocols.stream()
+ .filter(protocol -> protocol instanceof ServerReconfigurable)
+ .map(protocol -> (ServerReconfigurable) protocol)
+ .collect(Collectors.toList());
+ }
+
@Override
public ScheduledExecutorService getScheduledExecutor() {
checkState(isRunning, "Netty server has not been started yet.");
diff --git
a/fluss-rpc/src/test/java/org/apache/fluss/rpc/netty/authenticate/AuthenticationTest.java
b/fluss-rpc/src/test/java/org/apache/fluss/rpc/netty/authenticate/AuthenticationTest.java
index 6f4977f8b..9b98d735a 100644
---
a/fluss-rpc/src/test/java/org/apache/fluss/rpc/netty/authenticate/AuthenticationTest.java
+++
b/fluss-rpc/src/test/java/org/apache/fluss/rpc/netty/authenticate/AuthenticationTest.java
@@ -242,11 +242,7 @@ public class AuthenticationTest {
configuration.setString(
ConfigOptions.SERVER_SECURITY_PROTOCOL_MAP.key(),
"CLIENT1:mutual,CLIENT2:sasl");
configuration.setString("security.sasl.enabled.mechanisms", "plain");
- configuration.setString(
- "security.sasl.plain.jaas.config",
- "org.apache.fluss.security.auth.sasl.plain.PlainLoginModule
required "
- + " user_root=\"password\" "
- + " user_guest=\"password2\";");
+ configuration.setString("security.sasl.plain.credentials",
"root:password,guest:password2");
configuration.set(ConfigOptions.SUPER_USERS, "User:root");
configuration.set(ConfigOptions.AUTHORIZER_ENABLED, true);
// 3 worker threads is enough for this test
diff --git
a/fluss-rpc/src/test/java/org/apache/fluss/rpc/netty/authenticate/SaslAuthenticationITCase.java
b/fluss-rpc/src/test/java/org/apache/fluss/rpc/netty/authenticate/SaslAuthenticationITCase.java
index b02e02d63..b8a2a155b 100644
---
a/fluss-rpc/src/test/java/org/apache/fluss/rpc/netty/authenticate/SaslAuthenticationITCase.java
+++
b/fluss-rpc/src/test/java/org/apache/fluss/rpc/netty/authenticate/SaslAuthenticationITCase.java
@@ -22,7 +22,9 @@ 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.config.cluster.ServerReconfigurable;
import org.apache.fluss.exception.AuthenticationException;
+import org.apache.fluss.exception.ConfigException;
import org.apache.fluss.metrics.groups.MetricGroup;
import org.apache.fluss.metrics.util.NOPMetricsGroup;
import org.apache.fluss.rpc.TestingTabletGatewayService;
@@ -180,6 +182,315 @@ public class SaslAuthenticationITCase {
testAuthentication(clientConfig);
}
+ @Test
+ void testAddAndDeleteUser() throws Exception {
+ // Start a server with username/password list config (admin and alice)
+ Configuration serverConfig = new Configuration();
+
serverConfig.setString(ConfigOptions.SERVER_SECURITY_PROTOCOL_MAP.key(),
"CLIENT:sasl");
+ serverConfig.setString("security.sasl.enabled.mechanisms", "plain");
+ serverConfig.setString(
+ "security.sasl.plain.credentials",
"admin:admin-secret,alice:alice-secret");
+
serverConfig.setString(ConfigOptions.NETTY_SERVER_NUM_WORKER_THREADS.key(),
"3");
+
+ MetricGroup metricGroup = NOPMetricsGroup.newInstance();
+ TestingAuthenticateGatewayService service = new
TestingAuthenticateGatewayService();
+ try (NetUtils.Port port = getAvailablePort();
+ NettyServer nettyServer =
+ new NettyServer(
+ serverConfig,
+ Collections.singletonList(
+ new Endpoint("localhost",
port.getPort(), "CLIENT")),
+ service,
+ metricGroup,
+
RequestsMetrics.createCoordinatorServerRequestMetrics(
+ metricGroup))) {
+ nettyServer.start();
+ ServerNode serverNode =
+ new ServerNode(1, "localhost", port.getPort(),
ServerType.TABLET_SERVER);
+ ServerReconfigurable reconfigurable =
nettyServer.getServerReconfigurables().get(0);
+
+ // Verify "admin" can authenticate
+ try (NettyClient client = createSaslClient("admin",
"admin-secret")) {
+ verifyListTables(client, serverNode);
+ }
+
+ // Verify "bob" cannot authenticate initially
+ try (NettyClient client = createSaslClient("bob", "bob-secret")) {
+ assertThatThrownBy(() -> verifyListTables(client, serverNode))
+ .cause()
+ .isExactlyInstanceOf(AuthenticationException.class)
+ .hasMessageContaining("Invalid username or password");
+ }
+
+ // Add user "bob" via reconfigure
+ Configuration addBobConfig = new Configuration();
+
addBobConfig.setString(ConfigOptions.SERVER_SECURITY_PROTOCOL_MAP.key(),
"CLIENT:sasl");
+ addBobConfig.setString("security.sasl.enabled.mechanisms",
"plain");
+ addBobConfig.setString(
+ "security.sasl.plain.credentials",
+ "admin:admin-secret,alice:alice-secret,bob:bob-secret");
+ reconfigurable.validate(addBobConfig);
+ reconfigurable.reconfigure(addBobConfig);
+
+ // Verify "bob" can now authenticate
+ try (NettyClient client = createSaslClient("bob", "bob-secret")) {
+ verifyListTables(client, serverNode);
+ }
+
+ // Delete user "admin" via reconfigure
+ Configuration removeAdminConfig = new Configuration();
+ removeAdminConfig.setString(
+ ConfigOptions.SERVER_SECURITY_PROTOCOL_MAP.key(),
"CLIENT:sasl");
+ removeAdminConfig.setString("security.sasl.enabled.mechanisms",
"plain");
+ removeAdminConfig.setString(
+ "security.sasl.plain.credentials",
"alice:alice-secret,bob:bob-secret");
+ reconfigurable.validate(removeAdminConfig);
+ reconfigurable.reconfigure(removeAdminConfig);
+
+ // Verify "admin" can no longer authenticate
+ try (NettyClient client = createSaslClient("admin",
"admin-secret")) {
+ assertThatThrownBy(() -> verifyListTables(client, serverNode))
+ .cause()
+ .isExactlyInstanceOf(AuthenticationException.class)
+ .hasMessageContaining("Invalid username or password");
+ }
+
+ // Verify "bob" still works
+ try (NettyClient client = createSaslClient("bob", "bob-secret")) {
+ verifyListTables(client, serverNode);
+ }
+ }
+ }
+
+ @Test
+ void testReconfigureMergesUsersWithExistingJaasConfig() throws Exception {
+ // Start server with legacy jaas.config containing "admin" and "alice"
+ Configuration serverConfig = new Configuration();
+
serverConfig.setString(ConfigOptions.SERVER_SECURITY_PROTOCOL_MAP.key(),
"CLIENT:sasl");
+ serverConfig.setString("security.sasl.enabled.mechanisms", "plain");
+ serverConfig.setString(
+ "security.sasl.plain.jaas.config",
+ "org.apache.fluss.security.auth.sasl.plain.PlainLoginModule
required"
+ + " user_admin=\"admin-secret\""
+ + " user_alice=\"alice-secret\";");
+
serverConfig.setString(ConfigOptions.NETTY_SERVER_NUM_WORKER_THREADS.key(),
"3");
+
+ MetricGroup metricGroup = NOPMetricsGroup.newInstance();
+ TestingAuthenticateGatewayService service = new
TestingAuthenticateGatewayService();
+ try (NetUtils.Port port = getAvailablePort();
+ NettyServer nettyServer =
+ new NettyServer(
+ serverConfig,
+ Collections.singletonList(
+ new Endpoint("localhost",
port.getPort(), "CLIENT")),
+ service,
+ metricGroup,
+
RequestsMetrics.createCoordinatorServerRequestMetrics(
+ metricGroup))) {
+ nettyServer.start();
+ ServerNode serverNode =
+ new ServerNode(1, "localhost", port.getPort(),
ServerType.TABLET_SERVER);
+ ServerReconfigurable reconfigurable =
nettyServer.getServerReconfigurables().get(0);
+
+ // Verify existing users from jaas.config can authenticate
+ try (NettyClient client = createSaslClient("admin",
"admin-secret")) {
+ verifyListTables(client, serverNode);
+ }
+ try (NettyClient client = createSaslClient("alice",
"alice-secret")) {
+ verifyListTables(client, serverNode);
+ }
+
+ // Verify "bob" cannot authenticate before reconfigure
+ try (NettyClient client = createSaslClient("bob", "bob-secret")) {
+ assertThatThrownBy(() -> verifyListTables(client, serverNode))
+ .cause()
+ .isExactlyInstanceOf(AuthenticationException.class)
+ .hasMessageContaining("Invalid username or password");
+ }
+
+ // Reconfigure with SERVER_SASL_CREDENTIALS containing only "bob".
+ // The merge logic should keep existing users (admin, alice) AND
add bob.
+ Configuration newConfig = new Configuration();
+
newConfig.setString(ConfigOptions.SERVER_SECURITY_PROTOCOL_MAP.key(),
"CLIENT:sasl");
+ newConfig.setString("security.sasl.plain.credentials",
"bob:bob-secret");
+ reconfigurable.validate(newConfig);
+ reconfigurable.reconfigure(newConfig);
+
+ // After merge: admin, alice (from existing jaas.config) + bob
(from users list)
+ try (NettyClient client = createSaslClient("admin",
"admin-secret")) {
+ verifyListTables(client, serverNode);
+ }
+ try (NettyClient client = createSaslClient("alice",
"alice-secret")) {
+ verifyListTables(client, serverNode);
+ }
+ try (NettyClient client = createSaslClient("bob", "bob-secret")) {
+ verifyListTables(client, serverNode);
+ }
+
+ newConfig.removeKey("security.sasl.plain.credentials");
+ reconfigurable.validate(newConfig);
+ reconfigurable.reconfigure(newConfig);
+ // Verify existing users from jaas.config can authenticate
+ try (NettyClient client = createSaslClient("alice",
"alice-secret")) {
+ verifyListTables(client, serverNode);
+ }
+
+ // Verify "bob" cannot authenticate before reconfigure
+ try (NettyClient client = createSaslClient("bob", "bob-secret")) {
+ assertThatThrownBy(() -> verifyListTables(client, serverNode))
+ .cause()
+ .isExactlyInstanceOf(AuthenticationException.class)
+ .hasMessageContaining("Invalid username or password");
+ }
+ }
+ }
+
+ @Test
+ void testValidateRejectsInvalidUsernameCharacters() throws Exception {
+ Configuration serverConfig = new Configuration();
+
serverConfig.setString(ConfigOptions.SERVER_SECURITY_PROTOCOL_MAP.key(),
"CLIENT:sasl");
+ serverConfig.setString("security.sasl.enabled.mechanisms", "plain");
+ serverConfig.setString(
+ "security.sasl.plain.jaas.config",
+ "org.apache.fluss.security.auth.sasl.plain.PlainLoginModule
required"
+ + " user_admin=\"admin-secret\";");
+
serverConfig.setString(ConfigOptions.NETTY_SERVER_NUM_WORKER_THREADS.key(),
"3");
+
+ MetricGroup metricGroup = NOPMetricsGroup.newInstance();
+ TestingAuthenticateGatewayService service = new
TestingAuthenticateGatewayService();
+ try (NetUtils.Port port = getAvailablePort();
+ NettyServer nettyServer =
+ new NettyServer(
+ serverConfig,
+ Collections.singletonList(
+ new Endpoint("localhost",
port.getPort(), "CLIENT")),
+ service,
+ metricGroup,
+
RequestsMetrics.createCoordinatorServerRequestMetrics(
+ metricGroup))) {
+ nettyServer.start();
+ ServerReconfigurable reconfigurable =
nettyServer.getServerReconfigurables().get(0);
+
+ // Username with spaces
+ Configuration badUsername = new Configuration();
+ badUsername.setString("security.sasl.plain.credentials", "bad
user:password");
+ assertThatThrownBy(() -> reconfigurable.validate(badUsername))
+ .isInstanceOf(ConfigException.class)
+ .hasMessageContaining("contains invalid characters");
+
+ // Username with special chars (@)
+ Configuration atUsername = new Configuration();
+ atUsername.setString("security.sasl.plain.credentials",
"user@domain:password");
+ assertThatThrownBy(() -> reconfigurable.validate(atUsername))
+ .isInstanceOf(ConfigException.class)
+ .hasMessageContaining("contains invalid characters");
+
+ // Username with dots
+ Configuration dotUsername = new Configuration();
+ dotUsername.setString("security.sasl.plain.credentials",
"user.name:password");
+ assertThatThrownBy(() -> reconfigurable.validate(dotUsername))
+ .isInstanceOf(ConfigException.class)
+ .hasMessageContaining("contains invalid characters");
+
+ // Username with hyphens
+ Configuration hyphenUsername = new Configuration();
+ hyphenUsername.setString("security.sasl.plain.credentials",
"user-name:password");
+ assertThatThrownBy(() -> reconfigurable.validate(hyphenUsername))
+ .isInstanceOf(ConfigException.class)
+ .hasMessageContaining("contains invalid characters");
+
+ // Valid username with underscores should pass
+ Configuration validUsername = new Configuration();
+ validUsername.setString("security.sasl.plain.credentials",
"user_name:pw");
+ reconfigurable.validate(validUsername);
+ }
+ }
+
+ @Test
+ void testValidateRejectsInvalidPasswordCharacters() throws Exception {
+ Configuration serverConfig = new Configuration();
+
serverConfig.setString(ConfigOptions.SERVER_SECURITY_PROTOCOL_MAP.key(),
"CLIENT:sasl");
+ serverConfig.setString("security.sasl.enabled.mechanisms", "plain");
+ serverConfig.setString(
+ "security.sasl.plain.jaas.config",
+ "org.apache.fluss.security.auth.sasl.plain.PlainLoginModule
required"
+ + " user_admin=\"admin-secret\";");
+
serverConfig.setString(ConfigOptions.NETTY_SERVER_NUM_WORKER_THREADS.key(),
"3");
+
+ MetricGroup metricGroup = NOPMetricsGroup.newInstance();
+ TestingAuthenticateGatewayService service = new
TestingAuthenticateGatewayService();
+ try (NetUtils.Port port = getAvailablePort();
+ NettyServer nettyServer =
+ new NettyServer(
+ serverConfig,
+ Collections.singletonList(
+ new Endpoint("localhost",
port.getPort(), "CLIENT")),
+ service,
+ metricGroup,
+
RequestsMetrics.createCoordinatorServerRequestMetrics(
+ metricGroup))) {
+ nettyServer.start();
+ ServerReconfigurable reconfigurable =
nettyServer.getServerReconfigurables().get(0);
+
+ // Password with comma (breaks map format)
+ Configuration commaPassword = new Configuration();
+ commaPassword.setString("security.sasl.plain.credentials",
"bob:pass,word");
+ assertThatThrownBy(() -> reconfigurable.validate(commaPassword))
+ .isInstanceOf(ConfigException.class)
+ .hasMessageContaining("Failed to parse
security.sasl.plain.credentials");
+
+ // Password with colon (breaks map key-value format)
+ Configuration colonPassword = new Configuration();
+ colonPassword.setString("security.sasl.plain.credentials",
"bob:'pass:word'");
+ assertThatThrownBy(() -> reconfigurable.validate(colonPassword))
+ .isInstanceOf(ConfigException.class)
+ .hasMessageContaining("contains invalid characters");
+
+ // Password with double-quote (breaks JAAS value)
+ Configuration quotePassword = new Configuration();
+ quotePassword.setString("security.sasl.plain.credentials",
"bob:pass\"word");
+ assertThatThrownBy(() -> reconfigurable.validate(quotePassword))
+ .isInstanceOf(ConfigException.class)
+ .hasMessageContaining("password for user 'bob' contains
invalid characters");
+
+ // Password with semicolon (breaks JAAS statement)
+ Configuration semicolonPassword = new Configuration();
+ semicolonPassword.setString("security.sasl.plain.credentials",
"bob:pass;word");
+ assertThatThrownBy(() ->
reconfigurable.validate(semicolonPassword))
+ .isInstanceOf(ConfigException.class)
+ .hasMessageContaining("contains invalid characters");
+
+ // Password with backslash (escape char)
+ Configuration backslashPassword = new Configuration();
+ backslashPassword.setString("security.sasl.plain.credentials",
"bob:pass\\word");
+ assertThatThrownBy(() ->
reconfigurable.validate(backslashPassword))
+ .isInstanceOf(ConfigException.class)
+ .hasMessageContaining("contains invalid characters");
+
+ // Valid password with special chars (!, @, #, $, etc.) should pass
+ Configuration validPassword = new Configuration();
+ validPassword.setString("security.sasl.plain.credentials",
"bob:P@ss!w0rd#$%^&*()");
+ reconfigurable.validate(validPassword);
+ }
+ }
+
+ private NettyClient createSaslClient(String username, String password) {
+ Configuration clientConfig = new Configuration();
+ clientConfig.setString("client.security.protocol", "sasl");
+ clientConfig.setString("client.security.sasl.mechanism", "plain");
+ clientConfig.setString("client.security.sasl.username", username);
+ clientConfig.setString("client.security.sasl.password", password);
+ return new NettyClient(clientConfig,
TestingClientMetricGroup.newInstance());
+ }
+
+ private void verifyListTables(NettyClient client, ServerNode serverNode)
throws Exception {
+ ListTablesRequest request = new
ListTablesRequest().setDatabaseName("test-database");
+ ListTablesResponse response =
+ (ListTablesResponse)
+ client.sendRequest(serverNode, ApiKeys.LIST_TABLES,
request).get();
+
assertThat(response.getTableNamesList()).isEqualTo(Collections.singletonList("test-table"));
+ }
+
private void testAuthentication(Configuration clientConfig) throws
Exception {
testAuthentication(clientConfig, getDefaultServerConfig());
}
diff --git
a/fluss-server/src/main/java/org/apache/fluss/server/DynamicConfigManager.java
b/fluss-server/src/main/java/org/apache/fluss/server/DynamicConfigManager.java
index a6b678c2e..74d6147c7 100644
---
a/fluss-server/src/main/java/org/apache/fluss/server/DynamicConfigManager.java
+++
b/fluss-server/src/main/java/org/apache/fluss/server/DynamicConfigManager.java
@@ -18,6 +18,8 @@
package org.apache.fluss.server;
import org.apache.fluss.annotation.VisibleForTesting;
+import org.apache.fluss.config.ConfigOption;
+import org.apache.fluss.config.ConfigOptions;
import org.apache.fluss.config.Configuration;
import org.apache.fluss.config.cluster.AlterConfig;
import org.apache.fluss.config.cluster.ConfigEntry;
@@ -36,6 +38,7 @@ import org.slf4j.LoggerFactory;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;
+import java.util.Objects;
/** Manager for dynamic configurations. */
public class DynamicConfigManager {
@@ -104,7 +107,9 @@ public class DynamicConfigManager {
if (!dynamicDefaultConfigs.containsKey(key)) {
ConfigEntry configEntry =
new ConfigEntry(
- key, value,
ConfigEntry.ConfigSource.INITIAL_SERVER_CONFIG);
+ key,
+
dynamicServerConfig.redactConfigValue(key, value),
+
ConfigEntry.ConfigSource.INITIAL_SERVER_CONFIG);
configEntries.add(configEntry);
}
});
@@ -112,7 +117,9 @@ public class DynamicConfigManager {
(key, value) -> {
ConfigEntry configEntry =
new ConfigEntry(
- key, value,
ConfigEntry.ConfigSource.DYNAMIC_SERVER_CONFIG);
+ key,
+ dynamicServerConfig.redactConfigValue(key,
value),
+
ConfigEntry.ConfigSource.DYNAMIC_SERVER_CONFIG);
configEntries.add(configEntry);
});
@@ -129,21 +136,29 @@ public class DynamicConfigManager {
List<AlterConfig> alterConfigs, Map<String, String> configsProps) {
alterConfigs.forEach(
alterConfigOp -> {
- String configPropName = alterConfigOp.key();
- if (!dynamicServerConfig.isAllowedConfig(configPropName)) {
+ String configKey = alterConfigOp.key();
+ if (!dynamicServerConfig.isAllowedConfig(configKey)) {
throw new ConfigException(
String.format(
"The config key %s is not allowed to
be changed dynamically.",
- configPropName));
+ configKey));
}
- String configPropValue = alterConfigOp.value();
+ String configValue = alterConfigOp.value();
switch (alterConfigOp.opType()) {
case SET:
- configsProps.put(configPropName, configPropValue);
+ configsProps.put(configKey, configValue);
break;
case DELETE:
- configsProps.remove(configPropName);
+ configsProps.remove(configKey);
+ break;
+ case APPEND:
+ validateListOrMapType(configKey);
+ appendCollectionConfig(configsProps, configKey,
configValue);
+ break;
+ case SUBTRACT:
+ validateListOrMapType(configKey);
+ subtractCollectionConfig(configsProps, configKey,
configValue);
break;
default:
throw new ConfigException(
@@ -152,6 +167,149 @@ public class DynamicConfigManager {
});
}
+ private void appendCollectionConfig(
+ Map<String, String> dynamicConfigs, String configKey, String
configValue) {
+ if (isMapType(configKey)) {
+ appendMapConfig(dynamicConfigs, configKey, configValue);
+ return;
+ }
+
+ appendListConfig(dynamicConfigs, configKey, configValue);
+ }
+
+ private void appendListConfig(
+ Map<String, String> dynamicConfigs, String configKey, String
configValue) {
+ String existingValue = getExistingConfigValue(dynamicConfigs,
configKey);
+ if (existingValue == null || existingValue.isEmpty()) {
+ dynamicConfigs.put(configKey, configValue);
+ } else {
+ dynamicConfigs.put(configKey, existingValue + "," + configValue);
+ }
+ }
+
+ private void subtractCollectionConfig(
+ Map<String, String> dynamicConfigs, String configKey, String
configValue) {
+ if (isMapType(configKey)) {
+ subtractMapConfig(dynamicConfigs, configKey, configValue);
+ return;
+ }
+
+ subtractListConfig(dynamicConfigs, configKey, configValue);
+ }
+
+ private void subtractListConfig(
+ Map<String, String> dynamicConfigs, String configKey, String
configValue) {
+ String existingValue = getExistingConfigValue(dynamicConfigs,
configKey);
+ if (existingValue == null || existingValue.isEmpty()) {
+ return;
+ }
+
+ List<String> items = new ArrayList<>();
+ for (String item : existingValue.split(",")) {
+ String trimmed = item.trim();
+ if (!trimmed.isEmpty()) {
+ items.add(trimmed);
+ }
+ }
+ items.removeIf(v -> v.equals(configValue));
+ if (items.isEmpty()) {
+ dynamicConfigs.put(configKey, null);
+ } else {
+ dynamicConfigs.put(configKey, String.join(",", items));
+ }
+ }
+
+ private void appendMapConfig(
+ Map<String, String> dynamicConfigs, String configKey, String
configValue) {
+ validateMapEntry(configKey, configValue);
+ String mapEntryKey = getMapEntryKey(configValue);
+ String existingValue = getExistingConfigValue(dynamicConfigs,
configKey);
+ if (existingValue == null || existingValue.isEmpty()) {
+ dynamicConfigs.put(configKey, configValue);
+ return;
+ }
+
+ String existingEntry = findMapEntry(existingValue, mapEntryKey);
+ if (existingEntry != null) {
+ throw new ConfigException(
+ configKey
+ + " must not contain duplicate map entry keys: '"
+ + mapEntryKey
+ + "'.");
+ }
+ dynamicConfigs.put(configKey, existingValue + "," + configValue);
+ }
+
+ private void subtractMapConfig(
+ Map<String, String> dynamicConfigs, String configKey, String
configValue) {
+ validateMapEntry(configKey, configValue);
+ String targetMapEntryKey = getMapEntryKey(configValue);
+ String existingValue = getExistingConfigValue(dynamicConfigs,
configKey);
+ if (existingValue == null || existingValue.isEmpty()) {
+ return;
+ }
+
+ List<String> entries = new ArrayList<>();
+ boolean removed = false;
+ for (String entry : existingValue.split(",")) {
+ String trimmed = entry.trim();
+ if (trimmed.isEmpty()) {
+ continue;
+ }
+ if (Objects.equals(getMapEntryKey(trimmed), targetMapEntryKey)) {
+ removed = true;
+ continue;
+ }
+ entries.add(trimmed);
+ }
+ if (!removed) {
+ return;
+ }
+ if (entries.isEmpty()) {
+ dynamicConfigs.put(configKey, null);
+ } else {
+ dynamicConfigs.put(configKey, String.join(",", entries));
+ }
+ }
+
+ private static String getMapEntryKey(String entry) {
+ int separatorIndex = entry.indexOf(':');
+ if (separatorIndex <= 0) {
+ throw new ConfigException(
+ String.format("Map item is not a key-value pair: '%s'.",
entry));
+ }
+ return entry.substring(0, separatorIndex);
+ }
+
+ private static String findMapEntry(String value, String targetKey) {
+ for (String entry : value.split(",")) {
+ String trimmed = entry.trim();
+ if (!trimmed.isEmpty() && Objects.equals(getMapEntryKey(trimmed),
targetKey)) {
+ return trimmed;
+ }
+ }
+ return null;
+ }
+
+ private static void validateMapEntry(String configKey, String configValue)
{
+ Configuration configuration = new Configuration();
+ configuration.setString(configKey, configValue);
+ try {
+ configuration.get(ConfigOptions.getConfigOption(configKey));
+ } catch (IllegalArgumentException | IllegalStateException e) {
+ throw new ConfigException(
+ String.format("Invalid map entry for config '%s': %s",
configKey, configValue),
+ e);
+ }
+ }
+
+ private String getExistingConfigValue(Map<String, String> dynamicConfigs,
String configKey) {
+ if (dynamicConfigs.containsKey(configKey)) {
+ return dynamicConfigs.get(configKey);
+ }
+ return dynamicServerConfig.getInitialServerConfigs().get(configKey);
+ }
+
@VisibleForTesting
protected void alterServerConfigs(Map<String, String> configsProps) throws
Exception {
dynamicServerConfig.updateDynamicConfig(configsProps, false);
@@ -160,6 +318,23 @@ public class DynamicConfigManager {
zooKeeperClient.upsertServerEntityConfig(configsProps);
}
+ private static void validateListOrMapType(String configKey) {
+ ConfigOption<?> configOption =
ConfigOptions.getConfigOption(configKey);
+ if (configOption == null
+ || (!configOption.isList() && configOption.getClazz() !=
Map.class)) {
+ throw new ConfigException(
+ String.format(
+ "APPEND/SUBTRACT operations are only supported for
list-typed or map-typed config keys, "
+ + "but '%s' is not a list or map type.",
+ configKey));
+ }
+ }
+
+ private static boolean isMapType(String configKey) {
+ ConfigOption<?> configOption =
ConfigOptions.getConfigOption(configKey);
+ return configOption != null && configOption.getClazz() == Map.class;
+ }
+
private class ConfigChangedNotificationHandler
implements ZkNodeChangeNotificationWatcher.NotificationHandler {
diff --git
a/fluss-server/src/main/java/org/apache/fluss/server/DynamicServerConfig.java
b/fluss-server/src/main/java/org/apache/fluss/server/DynamicServerConfig.java
index df82f606d..710c6b675 100644
---
a/fluss-server/src/main/java/org/apache/fluss/server/DynamicServerConfig.java
+++
b/fluss-server/src/main/java/org/apache/fluss/server/DynamicServerConfig.java
@@ -25,10 +25,13 @@ import org.apache.fluss.config.ConfigurationUtils;
import org.apache.fluss.config.cluster.ConfigValidator;
import org.apache.fluss.config.cluster.ServerReconfigurable;
import org.apache.fluss.exception.ConfigException;
+import org.apache.fluss.server.config.ConfigRedactor;
+import org.apache.fluss.server.config.ConfigRedactors;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collections;
import java.util.HashMap;
@@ -50,6 +53,8 @@ import static
org.apache.fluss.config.ConfigOptions.REMOTE_DATA_DIRS;
import static org.apache.fluss.config.ConfigOptions.REMOTE_DATA_DIRS_STRATEGY;
import static org.apache.fluss.config.ConfigOptions.REMOTE_DATA_DIRS_WEIGHTS;
import static
org.apache.fluss.config.ConfigOptions.SERVER_DATA_DISK_WRITE_LIMIT_RATIO;
+import static org.apache.fluss.config.ConfigOptions.SERVER_SASL_CREDENTIALS;
+import static
org.apache.fluss.config.ConfigOptions.SERVER_SASL_PLAIN_JAAS_CONFIG;
import static org.apache.fluss.utils.concurrent.LockUtils.inReadLock;
import static org.apache.fluss.utils.concurrent.LockUtils.inWriteLock;
@@ -74,7 +79,8 @@ class DynamicServerConfig {
// Config options for remote.data.dirs
REMOTE_DATA_DIRS.key(),
REMOTE_DATA_DIRS_STRATEGY.key(),
- REMOTE_DATA_DIRS_WEIGHTS.key()));
+ REMOTE_DATA_DIRS_WEIGHTS.key(),
+ SERVER_SASL_CREDENTIALS.key()));
private static final Set<String> ALLOWED_CONFIG_PREFIXES =
Collections.singleton("datalake.");
private final ReadWriteLock lock = new ReentrantReadWriteLock();
@@ -85,6 +91,9 @@ class DynamicServerConfig {
private final Map<String, List<ConfigValidator<?>>> configValidatorsByKey =
new ConcurrentHashMap<>();
+ /** Registered config redactors for sensitive values exposed through
describe config APIs. */
+ private final List<ConfigRedactor> configRedactors = new ArrayList<>();
+
/** The initial configuration items when the server starts from
server.yaml. */
private final Map<String, String> initialConfigMap;
@@ -106,6 +115,7 @@ class DynamicServerConfig {
this.currentConfig = flussConfig;
this.initialConfigMap = flussConfig.toMap();
this.currentConfigMap = flussConfig.toMap();
+ registerDefaultRedactors();
}
void register(ServerReconfigurable serverReconfigurable) {
@@ -131,6 +141,26 @@ class DynamicServerConfig {
.add(validator);
}
+ String redactConfigValue(String configKey, String value) {
+ for (ConfigRedactor configRedactor : configRedactors) {
+ if (configRedactor.supports(configKey)) {
+ return configRedactor.redact(value);
+ }
+ }
+ return value;
+ }
+
+ private void registerDefaultRedactors() {
+
configRedactors.add(ConfigRedactors.map(SERVER_SASL_CREDENTIALS.key()));
+
configRedactors.add(ConfigRedactors.value(DynamicServerConfig::isPlainJaasConfig));
+ }
+
+ private static boolean isPlainJaasConfig(String configKey) {
+ return SERVER_SASL_PLAIN_JAAS_CONFIG.key().equals(configKey)
+ || (configKey.startsWith("security.sasl.listener.name.")
+ && configKey.endsWith(".plain.jaas.config"));
+ }
+
/**
* Update the dynamic configuration and apply to registered
ServerReconfigurable. If skipping
* error config, only the error one will be ignored.
diff --git
a/fluss-server/src/main/java/org/apache/fluss/server/authorizer/DefaultAuthorizer.java
b/fluss-server/src/main/java/org/apache/fluss/server/authorizer/DefaultAuthorizer.java
index fac4aa740..4c27c43b0 100644
---
a/fluss-server/src/main/java/org/apache/fluss/server/authorizer/DefaultAuthorizer.java
+++
b/fluss-server/src/main/java/org/apache/fluss/server/authorizer/DefaultAuthorizer.java
@@ -119,6 +119,7 @@ public class DefaultAuthorizer extends AbstractAuthorizer
implements FatalErrorH
}
private final Set<FlussPrincipal> superUsers;
+ private final boolean principalIgnoreCase;
private final ZooKeeperClient zooKeeperClient;
private final ZkNodeChangeNotificationWatcher aclChangeNotificationWatcher;
@@ -134,6 +135,8 @@ public class DefaultAuthorizer extends AbstractAuthorizer
implements FatalErrorH
public DefaultAuthorizer(AuthorizationPlugin.Context context) {
Configuration configuration = context.getConfiguration();
this.superUsers = parseSuperUsers(configuration);
+ this.principalIgnoreCase =
+
configuration.get(ConfigOptions.SECURITY_ACL_PRINCIPAL_IGNORE_CASE);
if (context.getZooKeeperClient().isPresent()) {
this.zooKeeperClient = context.getZooKeeperClient().get();
} else {
@@ -167,7 +170,7 @@ public class DefaultAuthorizer extends AbstractAuthorizer
implements FatalErrorH
@Override
public boolean authorizeAction(Session session, Action action) {
FlussPrincipal principal = session.getPrincipal();
- return superUsers.contains(principal)
+ return isSuperUser(principal)
|| aclsAllowAccess(
action.getResource(),
principal,
@@ -175,6 +178,15 @@ public class DefaultAuthorizer extends AbstractAuthorizer
implements FatalErrorH
session.getInetAddress().getHostAddress());
}
+ private boolean isSuperUser(FlussPrincipal principal) {
+ for (FlussPrincipal superUser : superUsers) {
+ if (superUser.matches(principal, principalIgnoreCase)) {
+ return true;
+ }
+ }
+ return false;
+ }
+
@Override
public List<AclCreateResult> addAcls(Session session, List<AclBinding>
aclBindings) {
if (aclBindings.isEmpty()) {
@@ -514,9 +526,12 @@ public class DefaultAuthorizer extends AbstractAuthorizer
implements FatalErrorH
.filter(
acl ->
acl.getPermissionType() == permissionType
- &&
(acl.getPrincipal().equals(principal)
+ && (acl.getPrincipal()
+ .matches(principal,
principalIgnoreCase)
|| acl.getPrincipal()
-
.equals(FlussPrincipal.WILD_CARD_PRINCIPAL))
+ .matches(
+
FlussPrincipal.WILD_CARD_PRINCIPAL,
+
principalIgnoreCase))
&& (operation == acl.getOperationType()
|| acl.getOperationType() ==
OperationType.ALL)
&&
(acl.getHost().equals(AccessControlEntry.WILD_CARD_HOST)
diff --git
a/fluss-server/src/main/java/org/apache/fluss/server/config/ConfigRedactor.java
b/fluss-server/src/main/java/org/apache/fluss/server/config/ConfigRedactor.java
new file mode 100644
index 000000000..bc45e259f
--- /dev/null
+++
b/fluss-server/src/main/java/org/apache/fluss/server/config/ConfigRedactor.java
@@ -0,0 +1,34 @@
+/*
+ * 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.server.config;
+
+import org.apache.fluss.annotation.Internal;
+
+import javax.annotation.Nullable;
+
+/** Redacts sensitive values before exposing server configurations. */
+@Internal
+public interface ConfigRedactor {
+
+ /** Returns whether this redactor handles the given config key. */
+ boolean supports(String configKey);
+
+ /** Returns the redacted display value. */
+ @Nullable
+ String redact(@Nullable String value);
+}
diff --git
a/fluss-server/src/main/java/org/apache/fluss/server/config/ConfigRedactors.java
b/fluss-server/src/main/java/org/apache/fluss/server/config/ConfigRedactors.java
new file mode 100644
index 000000000..2d7e5b9c5
--- /dev/null
+++
b/fluss-server/src/main/java/org/apache/fluss/server/config/ConfigRedactors.java
@@ -0,0 +1,39 @@
+/*
+ * 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.server.config;
+
+import org.apache.fluss.annotation.Internal;
+
+import java.util.function.Predicate;
+
+/** Factory methods for server config redactors. */
+@Internal
+public final class ConfigRedactors {
+
+ private ConfigRedactors() {}
+
+ /** Creates a redactor that masks map values while preserving map keys. */
+ public static ConfigRedactor map(String configKey) {
+ return new MapConfigRedactor(configKey);
+ }
+
+ /** Creates a redactor that masks the whole config value when the matcher
accepts its key. */
+ public static ConfigRedactor value(Predicate<String> configKeyMatcher) {
+ return new ValueConfigRedactor(configKeyMatcher);
+ }
+}
diff --git
a/fluss-server/src/main/java/org/apache/fluss/server/config/MapConfigRedactor.java
b/fluss-server/src/main/java/org/apache/fluss/server/config/MapConfigRedactor.java
new file mode 100644
index 000000000..22be56dec
--- /dev/null
+++
b/fluss-server/src/main/java/org/apache/fluss/server/config/MapConfigRedactor.java
@@ -0,0 +1,61 @@
+/*
+ * 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.server.config;
+
+import javax.annotation.Nullable;
+
+import java.util.ArrayList;
+import java.util.List;
+
+/** Redacts map config values by preserving keys and masking values. */
+class MapConfigRedactor implements ConfigRedactor {
+
+ static final String REDACTED_VALUE = "******";
+
+ private final String configKey;
+
+ MapConfigRedactor(String configKey) {
+ this.configKey = configKey;
+ }
+
+ @Override
+ public boolean supports(String configKey) {
+ return this.configKey.equals(configKey);
+ }
+
+ @Override
+ public @Nullable String redact(@Nullable String value) {
+ if (value == null) {
+ return null;
+ }
+
+ List<String> redactedEntries = new ArrayList<>();
+ for (String rawEntry : value.split(",")) {
+ String entry = rawEntry.trim();
+ if (entry.isEmpty()) {
+ continue;
+ }
+ int separatorIndex = entry.indexOf(':');
+ if (separatorIndex <= 0) {
+ return REDACTED_VALUE;
+ }
+ redactedEntries.add(entry.substring(0, separatorIndex) + ":" +
REDACTED_VALUE);
+ }
+ return String.join(",", redactedEntries);
+ }
+}
diff --git
a/fluss-server/src/main/java/org/apache/fluss/server/config/ValueConfigRedactor.java
b/fluss-server/src/main/java/org/apache/fluss/server/config/ValueConfigRedactor.java
new file mode 100644
index 000000000..d96b7d78d
--- /dev/null
+++
b/fluss-server/src/main/java/org/apache/fluss/server/config/ValueConfigRedactor.java
@@ -0,0 +1,42 @@
+/*
+ * 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.server.config;
+
+import javax.annotation.Nullable;
+
+import java.util.function.Predicate;
+
+/** Redacts sensitive config values as a whole. */
+class ValueConfigRedactor implements ConfigRedactor {
+
+ private final Predicate<String> configKeyMatcher;
+
+ ValueConfigRedactor(Predicate<String> configKeyMatcher) {
+ this.configKeyMatcher = configKeyMatcher;
+ }
+
+ @Override
+ public boolean supports(String configKey) {
+ return configKeyMatcher.test(configKey);
+ }
+
+ @Override
+ public @Nullable String redact(@Nullable String value) {
+ return value == null ? null : MapConfigRedactor.REDACTED_VALUE;
+ }
+}
diff --git
a/fluss-server/src/main/java/org/apache/fluss/server/coordinator/CoordinatorServer.java
b/fluss-server/src/main/java/org/apache/fluss/server/coordinator/CoordinatorServer.java
index 549eaf657..d4fbcbb97 100644
---
a/fluss-server/src/main/java/org/apache/fluss/server/coordinator/CoordinatorServer.java
+++
b/fluss-server/src/main/java/org/apache/fluss/server/coordinator/CoordinatorServer.java
@@ -242,16 +242,6 @@ public class CoordinatorServer extends ServerBase {
this.remoteDirDynamicLoader = new RemoteDirDynamicLoader(conf);
this.dynamicConfigManager = new DynamicConfigManager(zkClient,
conf, true);
-
- // Register server reconfigurable components
- dynamicConfigManager.register(lakeCatalogDynamicLoader);
- dynamicConfigManager.register(remoteDirDynamicLoader);
-
- // Register stateless validators for coordinator-side upfront
validation
- dynamicConfigManager.registerValidator(new
DiskWriteLimitRatioValidator());
-
- dynamicConfigManager.startup();
-
this.metadataCache = new CoordinatorMetadataCache();
this.authorizer = AuthorizerLoader.createAuthorizer(conf,
zkClient, pluginManager);
@@ -305,6 +295,14 @@ public class CoordinatorServer extends ServerBase {
serverMetricGroup,
RequestsMetrics.createCoordinatorServerRequestMetrics(
serverMetricGroup));
+ // Register server reconfigurable components
+ dynamicConfigManager.register(lakeCatalogDynamicLoader);
+ dynamicConfigManager.register(remoteDirDynamicLoader);
+ // Register stateless validators for coordinator-side upfront
validation
+ dynamicConfigManager.registerValidator(new
DiskWriteLimitRatioValidator());
+
rpcServer.getServerReconfigurables().forEach(dynamicConfigManager::register);
+ dynamicConfigManager.startup();
+
rpcServer.start();
registerCoordinatorServer();
diff --git
a/fluss-server/src/main/java/org/apache/fluss/server/tablet/TabletServer.java
b/fluss-server/src/main/java/org/apache/fluss/server/tablet/TabletServer.java
index c7c5fdd78..adf9a59bd 100644
---
a/fluss-server/src/main/java/org/apache/fluss/server/tablet/TabletServer.java
+++
b/fluss-server/src/main/java/org/apache/fluss/server/tablet/TabletServer.java
@@ -227,7 +227,6 @@ public class TabletServer extends ServerBase {
MetadataManager metadataManager =
new MetadataManager(zkClient, conf,
lakeCatalogDynamicLoader);
this.dynamicConfigManager = new DynamicConfigManager(zkClient,
conf, false);
- dynamicConfigManager.register(lakeCatalogDynamicLoader);
this.metadataCache = new
TabletServerMetadataCache(metadataManager);
@@ -247,9 +246,6 @@ public class TabletServer extends ServerBase {
conf, zkClient, logManager,
tabletServerMetricGroup, localDiskManager);
kvManager.startup();
- // Register kvManager to dynamicConfigManager for dynamic
reconfiguration
- dynamicConfigManager.register(kvManager);
-
this.authorizer = AuthorizerLoader.createAuthorizer(conf,
zkClient, pluginManager);
if (authorizer != null) {
authorizer.startup();
@@ -295,15 +291,6 @@ public class TabletServer extends ServerBase {
localDiskManager);
replicaManager.startup();
- // Register DefaultSnapshotContext for dynamic kv.snapshot.interval
-
dynamicConfigManager.register(replicaManager.getKvSnapshotContext());
- // Register replicaManager to dynamicConfigManager for dynamic
config
- dynamicConfigManager.register(replicaManager);
- // Register localDiskManager for dynamic
server.data-disk.write-limit-ratio
- dynamicConfigManager.register(localDiskManager);
- // Start dynamicConfigManager after all reconfigurable components
are registered
- dynamicConfigManager.startup();
-
this.tabletService =
new TabletService(
serverId,
@@ -328,6 +315,21 @@ public class TabletServer extends ServerBase {
tabletService,
tabletServerMetricGroup,
requestsMetrics);
+
+ dynamicConfigManager.register(lakeCatalogDynamicLoader);
+ // Register kvManager to dynamicConfigManager for dynamic
reconfiguration
+ dynamicConfigManager.register(kvManager);
+ // Register DefaultSnapshotContext for dynamic kv.snapshot.interval
+
dynamicConfigManager.register(replicaManager.getKvSnapshotContext());
+ // Register replicaManager to dynamicConfigManager for dynamic
config
+ dynamicConfigManager.register(replicaManager);
+ // Register localDiskManager for dynamic
server.data-disk.write-limit-ratio
+ dynamicConfigManager.register(localDiskManager);
+
rpcServer.getServerReconfigurables().forEach(dynamicConfigManager::register);
+
+ // Start dynamicConfigManager after all reconfigurable components
are registered
+ dynamicConfigManager.startup();
+
rpcServer.start();
registerTabletServer();
diff --git
a/fluss-server/src/test/java/org/apache/fluss/server/DynamicConfigChangeTest.java
b/fluss-server/src/test/java/org/apache/fluss/server/DynamicConfigChangeTest.java
index c41afa0f6..839c8c69a 100644
---
a/fluss-server/src/test/java/org/apache/fluss/server/DynamicConfigChangeTest.java
+++
b/fluss-server/src/test/java/org/apache/fluss/server/DynamicConfigChangeTest.java
@@ -21,6 +21,7 @@ import org.apache.fluss.config.ConfigOptions;
import org.apache.fluss.config.Configuration;
import org.apache.fluss.config.cluster.AlterConfig;
import org.apache.fluss.config.cluster.AlterConfigOpType;
+import org.apache.fluss.config.cluster.ConfigEntry;
import org.apache.fluss.config.cluster.ServerReconfigurable;
import org.apache.fluss.exception.ConfigException;
import org.apache.fluss.server.coordinator.LakeCatalogDynamicLoader;
@@ -697,4 +698,246 @@ public class DynamicConfigChangeTest {
assertThat(localDiskManager.getDiskWriteLimitRatio()).isEqualTo(0.85);
}
}
+
+ @Test
+ void testAppendAndSubtractOnMapConfig() throws Exception {
+ Configuration configuration = new Configuration();
+ DynamicConfigManager dynamicConfigManager =
+ new DynamicConfigManager(zookeeperClient, configuration, true);
+
+ AtomicReference<Map<String, String>> reconfiguredUsers = new
AtomicReference<>();
+ dynamicConfigManager.register(
+ new ServerReconfigurable() {
+ @Override
+ public void validate(Configuration newConfig) throws
ConfigException {}
+
+ @Override
+ public void reconfigure(Configuration newConfig) {
+
reconfiguredUsers.set(newConfig.get(ConfigOptions.SERVER_SASL_CREDENTIALS));
+ }
+ });
+ dynamicConfigManager.startup();
+
+ // APPEND first user
+ dynamicConfigManager.alterConfigs(
+ Collections.singletonList(
+ new AlterConfig(
+ ConfigOptions.SERVER_SASL_CREDENTIALS.key(),
+ "admin:admin-secret",
+ AlterConfigOpType.APPEND)));
+
+ Map<String, String> zkConfig = zookeeperClient.fetchEntityConfig();
+ assertThat(zkConfig.get(ConfigOptions.SERVER_SASL_CREDENTIALS.key()))
+ .isEqualTo("admin:admin-secret");
+ assertThat(reconfiguredUsers.get()).containsEntry("admin",
"admin-secret");
+
+ // APPEND second user
+ dynamicConfigManager.alterConfigs(
+ Collections.singletonList(
+ new AlterConfig(
+ ConfigOptions.SERVER_SASL_CREDENTIALS.key(),
+ "bob:bob-secret",
+ AlterConfigOpType.APPEND)));
+
+ zkConfig = zookeeperClient.fetchEntityConfig();
+ assertThat(zkConfig.get(ConfigOptions.SERVER_SASL_CREDENTIALS.key()))
+ .isEqualTo("admin:admin-secret,bob:bob-secret");
+ assertThat(reconfiguredUsers.get())
+ .containsEntry("admin", "admin-secret")
+ .containsEntry("bob", "bob-secret");
+
+ // SUBTRACT with an existing key but a different value should remove
by map key.
+ dynamicConfigManager.alterConfigs(
+ Collections.singletonList(
+ new AlterConfig(
+ ConfigOptions.SERVER_SASL_CREDENTIALS.key(),
+ "admin:wrong-secret",
+ AlterConfigOpType.SUBTRACT)));
+
+ zkConfig = zookeeperClient.fetchEntityConfig();
+ assertThat(zkConfig.get(ConfigOptions.SERVER_SASL_CREDENTIALS.key()))
+ .isEqualTo("bob:bob-secret");
+ assertThat(reconfiguredUsers.get())
+ .containsEntry("bob", "bob-secret")
+ .doesNotContainKey("admin");
+
+ // SUBTRACT last user - should write null to override static config
+ dynamicConfigManager.alterConfigs(
+ Collections.singletonList(
+ new AlterConfig(
+ ConfigOptions.SERVER_SASL_CREDENTIALS.key(),
+ "bob:bob-secret",
+ AlterConfigOpType.SUBTRACT)));
+
+ zkConfig = zookeeperClient.fetchEntityConfig();
+
assertThat(zkConfig.containsKey(ConfigOptions.SERVER_SASL_CREDENTIALS.key())).isTrue();
+
assertThat(zkConfig.get(ConfigOptions.SERVER_SASL_CREDENTIALS.key())).isNull();
+ assertThat(reconfiguredUsers.get()).isNull();
+ }
+
+ @Test
+ void testAppendOnNonListOrMapConfigIsRejected() throws Exception {
+ Configuration configuration = new Configuration();
+ DynamicConfigManager dynamicConfigManager =
+ new DynamicConfigManager(zookeeperClient, configuration, true);
+ dynamicConfigManager.startup();
+
+ // APPEND on a non-list/non-map config (kv.snapshot.interval is
Duration type)
+ // should be rejected immediately by the collection-type check.
+ assertThatThrownBy(
+ () ->
+ dynamicConfigManager.alterConfigs(
+ Collections.singletonList(
+ new AlterConfig(
+
ConfigOptions.KV_SNAPSHOT_INTERVAL.key(),
+ "5min",
+
AlterConfigOpType.APPEND))))
+ .isInstanceOf(ConfigException.class)
+ .hasMessageContaining(
+ "APPEND/SUBTRACT operations are only supported for
list-typed or map-typed config keys");
+
+ // SUBTRACT on a non-list config should also be rejected.
+ assertThatThrownBy(
+ () ->
+ dynamicConfigManager.alterConfigs(
+ Collections.singletonList(
+ new AlterConfig(
+
ConfigOptions.KV_SNAPSHOT_INTERVAL.key(),
+ "5min",
+
AlterConfigOpType.SUBTRACT))))
+ .isInstanceOf(ConfigException.class)
+ .hasMessageContaining(
+ "APPEND/SUBTRACT operations are only supported for
list-typed or map-typed config keys");
+ }
+
+ @Test
+ void testDescribeRedactsSensitiveConfigs() throws Exception {
+ Configuration configuration = new Configuration();
+ configuration.setString(ConfigOptions.SERVER_SASL_CREDENTIALS.key(),
"admin:admin-secret");
+ configuration.setString(
+ ConfigOptions.SERVER_SASL_PLAIN_JAAS_CONFIG.key(),
+ "org.apache.fluss.security.auth.sasl.plain.PlainLoginModule
required "
+ + "user_admin=\"admin-secret\";");
+
+ DynamicConfigManager dynamicConfigManager =
+ new DynamicConfigManager(zookeeperClient, configuration, true);
+ dynamicConfigManager.startup();
+
+ assertThat(dynamicConfigManager.describeConfigs())
+ .contains(
+ new ConfigEntry(
+ ConfigOptions.SERVER_SASL_CREDENTIALS.key(),
+ "admin:******",
+
ConfigEntry.ConfigSource.INITIAL_SERVER_CONFIG),
+ new ConfigEntry(
+
ConfigOptions.SERVER_SASL_PLAIN_JAAS_CONFIG.key(),
+ "******",
+
ConfigEntry.ConfigSource.INITIAL_SERVER_CONFIG));
+ }
+
+ @Test
+ void testAppendReadsStaticConfig() throws Exception {
+ Configuration configuration = new Configuration();
+ // Pre-configure a static user in server.yaml equivalent
+ configuration.setString(ConfigOptions.SERVER_SASL_CREDENTIALS.key(),
"admin:admin-secret");
+
+ DynamicConfigManager dynamicConfigManager =
+ new DynamicConfigManager(zookeeperClient, configuration, true);
+
+ AtomicReference<Map<String, String>> reconfiguredUsers = new
AtomicReference<>();
+ dynamicConfigManager.register(
+ new ServerReconfigurable() {
+ @Override
+ public void validate(Configuration newConfig) throws
ConfigException {}
+
+ @Override
+ public void reconfigure(Configuration newConfig) {
+
reconfiguredUsers.set(newConfig.get(ConfigOptions.SERVER_SASL_CREDENTIALS));
+ }
+ });
+ dynamicConfigManager.startup();
+
+ // APPEND should build on top of static config
+ dynamicConfigManager.alterConfigs(
+ Collections.singletonList(
+ new AlterConfig(
+ ConfigOptions.SERVER_SASL_CREDENTIALS.key(),
+ "bob:bob-secret",
+ AlterConfigOpType.APPEND)));
+
+ Map<String, String> zkConfig = zookeeperClient.fetchEntityConfig();
+ assertThat(zkConfig.get(ConfigOptions.SERVER_SASL_CREDENTIALS.key()))
+ .isEqualTo("admin:admin-secret,bob:bob-secret");
+ assertThat(reconfiguredUsers.get())
+ .containsEntry("admin", "admin-secret")
+ .containsEntry("bob", "bob-secret");
+ }
+
+ @Test
+ void testSubtractFromStaticConfigWritesNull() throws Exception {
+ Configuration configuration = new Configuration();
+ // Pre-configure a static user in server.yaml equivalent
+ configuration.setString(ConfigOptions.SERVER_SASL_CREDENTIALS.key(),
"admin:admin-secret");
+
+ DynamicConfigManager dynamicConfigManager =
+ new DynamicConfigManager(zookeeperClient, configuration, true);
+
+ AtomicReference<Map<String, String>> reconfiguredUsers = new
AtomicReference<>();
+ dynamicConfigManager.register(
+ new ServerReconfigurable() {
+ @Override
+ public void validate(Configuration newConfig) throws
ConfigException {}
+
+ @Override
+ public void reconfigure(Configuration newConfig) {
+
reconfiguredUsers.set(newConfig.get(ConfigOptions.SERVER_SASL_CREDENTIALS));
+ }
+ });
+ dynamicConfigManager.startup();
+
+ // SUBTRACT the static user - should write null to override static
config
+ dynamicConfigManager.alterConfigs(
+ Collections.singletonList(
+ new AlterConfig(
+ ConfigOptions.SERVER_SASL_CREDENTIALS.key(),
+ "admin:admin-secret",
+ AlterConfigOpType.SUBTRACT)));
+
+ Map<String, String> zkConfig = zookeeperClient.fetchEntityConfig();
+
assertThat(zkConfig.containsKey(ConfigOptions.SERVER_SASL_CREDENTIALS.key())).isTrue();
+
assertThat(zkConfig.get(ConfigOptions.SERVER_SASL_CREDENTIALS.key())).isNull();
+ assertThat(reconfiguredUsers.get()).isNull();
+ }
+
+ @Test
+ void testSubtractTrimsWhitespaceAndRemovesByKey() throws Exception {
+ Configuration configuration = new Configuration();
+ DynamicConfigManager dynamicConfigManager =
+ new DynamicConfigManager(zookeeperClient, configuration, true);
+ dynamicConfigManager.startup();
+
+ // Set up a config with whitespace around entries
+ dynamicConfigManager.alterConfigs(
+ Collections.singletonList(
+ new AlterConfig(
+ ConfigOptions.SERVER_SASL_CREDENTIALS.key(),
+ "admin:admin-secret, bob:bob-secret",
+ AlterConfigOpType.SET)));
+
+ Map<String, String> zkConfig = zookeeperClient.fetchEntityConfig();
+ assertThat(zkConfig.get(ConfigOptions.SERVER_SASL_CREDENTIALS.key()))
+ .isEqualTo("admin:admin-secret, bob:bob-secret");
+
+ // SUBTRACT should trim entries and remove the matching map key
+ dynamicConfigManager.alterConfigs(
+ Collections.singletonList(
+ new AlterConfig(
+ ConfigOptions.SERVER_SASL_CREDENTIALS.key(),
+ "admin:admin-secret",
+ AlterConfigOpType.SUBTRACT)));
+
+ zkConfig = zookeeperClient.fetchEntityConfig();
+ assertThat(zkConfig.get(ConfigOptions.SERVER_SASL_CREDENTIALS.key()))
+ .isEqualTo("bob:bob-secret");
+ }
}
diff --git
a/fluss-server/src/test/java/org/apache/fluss/server/authorizer/DefaultAuthorizerTest.java
b/fluss-server/src/test/java/org/apache/fluss/server/authorizer/DefaultAuthorizerTest.java
index 68fd6eed2..7a597c2e3 100644
---
a/fluss-server/src/test/java/org/apache/fluss/server/authorizer/DefaultAuthorizerTest.java
+++
b/fluss-server/src/test/java/org/apache/fluss/server/authorizer/DefaultAuthorizerTest.java
@@ -223,6 +223,35 @@ public class DefaultAuthorizerTest {
.isTrue();
}
+ @Test
+ void testPrincipalIgnoreCaseMatchesConfiguredUser() throws Exception {
+ // The setup configures USER:root as a super user with full access.
USER:user1 has no
+ // matching ACLs and is not a super user, so READ on database1 should
always be denied.
+ // The root session intentionally uses user:ROOT to verify
case-sensitive matching first.
+ Session normalUserSession = createSession("USER", "user1",
"192.168.1.1");
+ Session superUserSession = createSession("user", "ROOT",
"192.168.1.1");
+ assertThat(authorizer.isAuthorized(normalUserSession, READ,
Resource.database("database1")))
+ .isFalse();
+ assertThat(authorizer.isAuthorized(superUserSession, READ,
Resource.database("database1")))
+ .isFalse();
+ Configuration ignoreCaseConfiguration = new
Configuration(configuration);
+
ignoreCaseConfiguration.setBoolean(ConfigOptions.SECURITY_ACL_PRINCIPAL_IGNORE_CASE,
true);
+ try (Authorizer ignoreCaseAuthorizer =
+ AuthorizerLoader.createAuthorizer(ignoreCaseConfiguration,
zooKeeperClient, null)) {
+ assertThat(ignoreCaseAuthorizer).isNotNull();
+ // Enabling ignore-case lets user:ROOT match the configured
USER:root super user, but it
+ // does not grant privileges to USER:user1.
+ assertThat(
+ ignoreCaseAuthorizer.isAuthorized(
+ normalUserSession, READ,
Resource.database("database1")))
+ .isFalse();
+ assertThat(
+ ignoreCaseAuthorizer.isAuthorized(
+ superUserSession, READ,
Resource.database("database1")))
+ .isTrue();
+ }
+ }
+
@Test
void testAclWithOperationAll() throws Exception {
Session session = createSession("user1", "192.168.1.1");
@@ -664,12 +693,16 @@ public class DefaultAuthorizerTest {
}
private Session createSession(String username, String host) throws
Exception {
+ return createSession("USER", username, host);
+ }
+
+ private Session createSession(String userType, String username, String
host) throws Exception {
return new Session(
(byte) 1,
"FLUSS",
false,
InetAddress.getByName(host),
- new FlussPrincipal(username, "USER"));
+ new FlussPrincipal(username, userType));
}
private AclBinding createAclBinding(
diff --git a/website/docs/engine-flink/procedures.md
b/website/docs/engine-flink/procedures.md
index b3035e836..24baa9c08 100644
--- a/website/docs/engine-flink/procedures.md
+++ b/website/docs/engine-flink/procedures.md
@@ -246,6 +246,78 @@ CALL sys.set_cluster_configs(
);
```
+### append_cluster_configs
+
+Append values to list-type or map-type cluster configurations dynamically.
+
+**Syntax:**
+
+```sql
+-- Append collection configuration values
+CALL [catalog_name.]sys.append_cluster_configs(
+ config_pairs => 'key1', 'value1' [, 'key2', 'value2' ...]
+)
+```
+
+**Parameters:**
+
+- `config_pairs`(required): For key-value pairs in configuration items, the
number of parameters must be even.
+
+**Important Notes:**
+
+- APPEND operations are only supported for list-type or map-type configuration
keys
+- For map-type configurations, values must use the `map-key:map-value` format
+- Appending a map entry whose key already exists is rejected
+
+**Example:**
+
+```sql title="Flink SQL"
+-- Use the Fluss catalog (replace 'fluss_catalog' with your catalog name if
different)
+USE fluss_catalog;
+
+-- Add a SASL Plain credential
+CALL sys.append_cluster_configs(
+ config_pairs => 'security.sasl.plain.credentials', 'bob:bob-secret'
+);
+```
+
+### subtract_cluster_configs
+
+Subtract values from list-type or map-type cluster configurations dynamically.
+
+**Syntax:**
+
+```sql
+-- Subtract collection configuration values
+CALL [catalog_name.]sys.subtract_cluster_configs(
+ config_pairs => 'key1', 'value1' [, 'key2', 'value2' ...]
+)
+```
+
+**Parameters:**
+
+- `config_pairs`(required): For key-value pairs in configuration items, the
number of parameters must be even.
+
+**Important Notes:**
+
+- SUBTRACT operations are only supported for list-type or map-type
configuration keys
+- For list-type configurations, SUBTRACT removes the exact list value
+- For map-type configurations, values must use the `map-key:map-value` format,
but removal is matched only by `map-key`; the supplied `map-value` is ignored
for matching
+- Subtracting a list value or map key that does not exist is a no-op
+
+**Example:**
+
+```sql title="Flink SQL"
+-- Use the Fluss catalog (replace 'fluss_catalog' with your catalog name if
different)
+USE fluss_catalog;
+
+-- Remove user "bob" from the SASL Plain credentials map.
+-- This removes the "bob" entry even if the stored secret is different.
+CALL sys.subtract_cluster_configs(
+ config_pairs => 'security.sasl.plain.credentials', 'bob:any-secret'
+);
+```
+
### reset_cluster_configs
reset cluster configurations dynamically.
@@ -521,4 +593,4 @@ USE fluss_catalog;
-- Drop KV snapshots leased under the given leaseId
CALL sys.drop_kv_snapshot_lease('test-lease-id');
-```
\ No newline at end of file
+```
diff --git a/website/docs/maintenance/operations/updating-configs.md
b/website/docs/maintenance/operations/updating-configs.md
index 6ab797239..f6d3b12e3 100644
--- a/website/docs/maintenance/operations/updating-configs.md
+++ b/website/docs/maintenance/operations/updating-configs.md
@@ -74,7 +74,7 @@ CALL sys.set_cluster_configs(
);
```
-See [Procedures](engine-flink/procedures.md#cluster-configuration-procedures)
for detailed documentation on `set_cluster_configs`, `reset_cluster_configs`,
and `get_cluster_configs` procedures.
+See [Procedures](engine-flink/procedures.md#cluster-configuration-procedures)
for detailed documentation on `get_cluster_configs`, `set_cluster_configs`,
`append_cluster_configs`, `subtract_cluster_configs`, and
`reset_cluster_configs` procedures.
</TabItem>
</Tabs>
