fix zk limit

Author: luoyuxia

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

@@ -572,6 +572,15 @@ 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

@@ -104,12 +104,13 @@ public static ZooKeeperClient startZookeeperClient(
                     new SessionConnectionStateErrorPolicy());
         }
 
+        // Configure ZKClientConfig with jute.maxbuffer
+        ZKClientConfig zkClientConfig;
         Optional<String> configPath =
                 configuration.getOptional(ConfigOptions.ZOOKEEPER_CONFIG_PATH);
         if (configPath.isPresent()) {
             try {
-                ZKClientConfig zkClientConfig = new ZKClientConfig(configPath.get());
-                curatorFrameworkBuilder.zkClientConfig(zkClientConfig);
+                zkClientConfig = new ZKClientConfig(configPath.get());
             } catch (QuorumPeerConfig.ConfigException e) {
                 LOG.warn("Fail to load zookeeper client config from path {}", configPath.get(), e);
                 throw new RuntimeException(
@@ -118,7 +119,15 @@ public static ZooKeeperClient startZookeeperClient(
                                 configPath.get()),
                         e);
             }
+        } else {
+            zkClientConfig = new ZKClientConfig();
         }
+
+        // Set jute.maxbuffer to match RPC frame length limit
+        int maxBufferSize = configuration.getInt(ConfigOptions.ZOOKEEPER_MAX_BUFFER_SIZE);
+        zkClientConfig.setProperty("jute.maxbuffer", String.valueOf(maxBufferSize));
+        LOG.debug("Set ZooKeeper jute.maxbuffer to {} bytes", maxBufferSize);
+        curatorFrameworkBuilder.zkClientConfig(zkClientConfig);
         return new ZooKeeperClient(
                 startZookeeperClient(curatorFrameworkBuilder, fatalErrorHandler), configuration);
     }