[server] Add ConfigOption to pass config path to zookeeper client (#909)

Author: wangfengpro

fluss-common/src/main/java/com/alibaba/fluss/config/ConfigOptions.java

@@ -486,6 +486,15 @@ public class ConfigOptions {
                                     + "Under certain configurations EnsembleTracking can lead to setting of ZooKeeper connection string "
                                     + "with unresolvable hostnames.");
 
+    public static final ConfigOption<String> ZOOKEEPER_CONFIG_PATH =
+            key("zookeeper.client.config-path")
+                    .stringType()
+                    .noDefaultValue()
+                    .withDescription(
+                            "The file path from which the ZooKeeper client reads its configuration. "
+                                    + "This allows each ZooKeeper client instance to load its own configuration file, "
+                                    + "instead of relying on shared JVM-level environment settings. "
+                                    + "This enables fine-grained control over ZooKeeper client behavior.");
     // ------------------------------------------------------------------------
     //  ConfigOptions for Log
     // ------------------------------------------------------------------------

fluss-server/src/main/java/com/alibaba/fluss/server/zk/ZooKeeperUtils.java

@@ -25,12 +25,15 @@
 import com.alibaba.fluss.shaded.curator5.org.apache.curator.framework.api.UnhandledErrorListener;
 import com.alibaba.fluss.shaded.curator5.org.apache.curator.framework.state.SessionConnectionStateErrorPolicy;
 import com.alibaba.fluss.shaded.curator5.org.apache.curator.retry.ExponentialBackoffRetry;
+import com.alibaba.fluss.shaded.zookeeper3.org.apache.zookeeper.client.ZKClientConfig;
+import com.alibaba.fluss.shaded.zookeeper3.org.apache.zookeeper.server.quorum.QuorumPeerConfig;
 
 import org.apache.commons.lang3.StringUtils;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
 import java.util.Arrays;
+import java.util.Optional;
 import java.util.stream.Collectors;
 
 import static com.alibaba.fluss.utils.Preconditions.checkNotNull;
@@ -96,6 +99,22 @@ public static ZooKeeperClient startZookeeperClient(
             curatorFrameworkBuilder.connectionStateErrorPolicy(
                     new SessionConnectionStateErrorPolicy());
         }
+
+        Optional<String> configPath =
+                configuration.getOptional(ConfigOptions.ZOOKEEPER_CONFIG_PATH);
+        if (configPath.isPresent()) {
+            try {
+                ZKClientConfig zkClientConfig = new ZKClientConfig(configPath.get());
+                curatorFrameworkBuilder.zkClientConfig(zkClientConfig);
+            } catch (QuorumPeerConfig.ConfigException e) {
+                LOG.warn("Fail to load zookeeper client config from path {}", configPath.get(), e);
+                throw new RuntimeException(
+                        String.format(
+                                "Fail to load zookeeper client config from path %s",
+                                configPath.get()),
+                        e);
+            }
+        }
         return new ZooKeeperClient(
                 startZookeeperClient(curatorFrameworkBuilder, fatalErrorHandler));
     }

fluss-server/src/test/java/com/alibaba/fluss/server/zk/ZooKeeperClientTest.java

@@ -17,6 +17,8 @@
 package com.alibaba.fluss.server.zk;
 
 import com.alibaba.fluss.cluster.Endpoint;
+import com.alibaba.fluss.config.ConfigOptions;
+import com.alibaba.fluss.config.Configuration;
 import com.alibaba.fluss.metadata.Schema;
 import com.alibaba.fluss.metadata.SchemaInfo;
 import com.alibaba.fluss.metadata.TableBucket;
@@ -30,6 +32,10 @@
 import com.alibaba.fluss.server.zk.data.TableAssignment;
 import com.alibaba.fluss.server.zk.data.TableRegistration;
 import com.alibaba.fluss.server.zk.data.TabletServerRegistration;
+import com.alibaba.fluss.shaded.curator5.org.apache.curator.CuratorZookeeperClient;
+import com.alibaba.fluss.shaded.curator5.org.apache.curator.framework.CuratorFramework;
+import com.alibaba.fluss.shaded.zookeeper3.org.apache.zookeeper.ZooKeeper;
+import com.alibaba.fluss.shaded.zookeeper3.org.apache.zookeeper.client.ZKClientConfig;
 import com.alibaba.fluss.testutils.common.AllCallbackWrapper;
 import com.alibaba.fluss.types.DataTypes;
 import com.alibaba.fluss.utils.types.Tuple2;
@@ -49,6 +55,7 @@
 import java.util.Set;
 
 import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
 
 /** Test for {@link ZooKeeperClient}. */
 class ZooKeeperClientTest {
@@ -371,4 +378,34 @@ void testPartition() throws Exception {
         partitions = zookeeperClient.getPartitions(tablePath);
         assertThat(partitions).containsExactly("p2");
     }
+
+    @Test
+    void testZookeeperConfigPath() throws Exception {
+        final Configuration config = new Configuration();
+        config.setString(
+                ConfigOptions.ZOOKEEPER_ADDRESS,
+                ZOO_KEEPER_EXTENSION_WRAPPER.getCustomExtension().getConnectString());
+        config.setString(ConfigOptions.ZOOKEEPER_CONFIG_PATH, "./no-file.properties");
+        assertThatThrownBy(
+                        () -> ZooKeeperUtils.startZookeeperClient(config, NOPErrorHandler.INSTANCE))
+                .isExactlyInstanceOf(RuntimeException.class)
+                .hasMessageContaining("Fail to load zookeeper client config from path");
+
+        config.setString(
+                ConfigOptions.ZOOKEEPER_CONFIG_PATH,
+                getClass().getClassLoader().getResource("zk.properties").getPath());
+        try (ZooKeeperClient zookeeperClient =
+                        ZooKeeperUtils.startZookeeperClient(config, NOPErrorHandler.INSTANCE);
+                CuratorFramework curatorClient = zookeeperClient.getCuratorClient();
+                CuratorZookeeperClient curatorZookeeperClient = curatorClient.getZookeeperClient();
+                ZooKeeper zooKeeper = curatorZookeeperClient.getZooKeeper()) {
+            ZKClientConfig clientConfig = zooKeeper.getClientConfig();
+            assertThat(clientConfig.getProperty(ZKClientConfig.ENABLE_CLIENT_SASL_KEY))
+                    .isEqualTo("true");
+            assertThat(clientConfig.getProperty(ZKClientConfig.LOGIN_CONTEXT_NAME_KEY))
+                    .isEqualTo("ZookeeperClient");
+            assertThat(clientConfig.getProperty(ZKClientConfig.ZK_SASL_CLIENT_USERNAME))
+                    .isEqualTo("zookeeper2");
+        }
+    }
 }

fluss-server/src/test/resources/zk.properties

@@ -0,0 +1,19 @@
+#
+#  Copyright (c) 2025 Alibaba Group Holding Ltd.
+#
+#  Licensed 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.
+#
+
+zookeeper.sasl.client=true
+zookeeper.sasl.clientconfig=ZookeeperClient
+zookeeper.sasl.client.username=zookeeper2

website/docs/maintenance/configuration.md

@@ -97,6 +97,7 @@ during the Fluss cluster working.
 | zookeeper.client.max-retry-attempts             | Integer  | 3       | Defines the number of connection retries before the client gives up.                                                                                                                                                                                                                                                                                                                                                                                                                                                          |
 | zookeeper.client.tolerate-suspended-connections | Boolean  | false   | Defines whether a suspended ZooKeeper connection will be treated as an error that causes the leader information to be invalidated or not. In case you set this option to %s, Fluss will wait until a ZooKeeper connection is marked as lost before it revokes the leadership of components. This has the effect that Fluss is more resilient against temporary connection instabilities at the cost of running more likely into timing issues with ZooKeeper.                                                                 |
 | zookeeper.client.ensemble-tracker               | Boolean  | true    | Defines whether Curator should enable ensemble tracker. This can be useful in certain scenarios in which CuratorFramework is accessing to ZK clusters via load balancer or Virtual IPs. Default Curator EnsembleTracking logic watches CuratorEventType.GET_CONFIG events and changes ZooKeeper connection string. It is not desired behaviour when ZooKeeper is running under the Virtual IPs. Under certain configurations EnsembleTracking can lead to setting of ZooKeeper connection string with unresolvable hostnames. |
+| zookeeper.client.config-path                    | String   | (None)  | The file path from which the ZooKeeper client reads its configuration. This allows each ZooKeeper client instance to load its own configuration file, instead of relying on shared JVM-level environment settings. This enables fine-grained control over ZooKeeper client behavior.                                                                                                                                                                                                                                          |
 
 ## Netty