[zk] Allow to set jute.maxbuffer for zk client (#2048)

Author: luoyuxia

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

@@ -572,6 +572,16 @@ public class ConfigOptions {
                                     + "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.");
+
+    public static final ConfigOption<Integer> ZOOKEEPER_MAX_BUFFER_SIZE =
+            key("zookeeper.client.max-buffer-size")
+                    .intType()
+                    .defaultValue(100 * 1024 * 1024) // 100MB
+                    .withDescription(
+                            "The maximum buffer size (in bytes) for ZooKeeper client. "
+                                    + "This corresponds to the jute.maxbuffer property. "
+                                    + "Default is 100MB to match the RPC frame length limit.");
+
     // ------------------------------------------------------------------------
     //  ConfigOptions for Log
     // ------------------------------------------------------------------------

fluss-server/src/main/java/org/apache/fluss/server/zk/ZooKeeperUtils.java

@@ -40,6 +40,7 @@
 import java.util.Optional;
 import java.util.stream.Collectors;
 
+import static org.apache.fluss.shaded.zookeeper3.org.apache.zookeeper.common.ZKConfig.JUTE_MAXBUFFER;
 import static org.apache.fluss.utils.Preconditions.checkNotNull;
 
 /** Class containing helper functions to interact with ZooKeeper. */
@@ -104,11 +105,12 @@ public static ZooKeeperClient startZookeeperClient(
                     new SessionConnectionStateErrorPolicy());
         }
 
+        ZKClientConfig zkClientConfig;
         Optional<String> configPath =
                 configuration.getOptional(ConfigOptions.ZOOKEEPER_CONFIG_PATH);
         if (configPath.isPresent()) {
             try {
-                ZKClientConfig zkClientConfig = new ZKClientConfig(configPath.get());
+                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);
@@ -118,7 +120,14 @@ public static ZooKeeperClient startZookeeperClient(
                                 configPath.get()),
                         e);
             }
+        } else {
+            zkClientConfig = new ZKClientConfig();
         }
+        // set jute.max buffer
+        zkClientConfig.setProperty(
+                JUTE_MAXBUFFER,
+                String.valueOf(configuration.getInt(ConfigOptions.ZOOKEEPER_MAX_BUFFER_SIZE)));
+
         return new ZooKeeperClient(
                 startZookeeperClient(curatorFrameworkBuilder, fatalErrorHandler), configuration);
     }