[server] Use unified batch-concurrent implementation for ZooKeeper operations and remove redundant code

This commit replaces all single-operation async calls
with a unified batch-concurrent implementation. This eliminates the need
to maintain three separate code paths (batch concurrent, single async,
and sync), significantly reducing code duplication in ZooKeeperClient.

Going forward, all sync ZooKeeper operations will be replaced by the batch-concurrent
approach, which improves coordinator performance through better I/O
utilization and reduced latency.

Additionally, the `cacheOnly` flag has been removed from the current
metadata request. Support for cache-only semantics will be reintroduced
in a future `MetadataRequestV2` with a cleaner design.

Author: wuchong

fluss-client/src/test/java/org/apache/fluss/client/table/AutoPartitionedTableITCase.java

@@ -327,7 +327,7 @@ void testOperateNotExistPartitionShouldThrowException() throws Exception {
         LogScanner logScanner = table.newScan().createLogScanner();
         assertThatThrownBy(() -> logScanner.subscribe(100L, 0, 0))
                 .isInstanceOf(PartitionNotExistException.class)
-                .hasMessageContaining("Partition not exist for partition ids: [100]");
+                .hasMessageContaining("The partition id '100' does not exist");
         logScanner.close();
 
         // todo: test the case that client produce to a partition to a server, but

fluss-common/src/main/java/org/apache/fluss/exception/CacheMissException.java

@@ -0,0 +1,19 @@
+#
+# 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.
+#
+
+org.apache.fluss.testutils.common.TestLoggerExtension

fluss-flink/fluss-flink-1.20/src/test/resources/META-INF/services/org.junit.jupiter.api.extension.Extension

@@ -0,0 +1,32 @@
+#
+#  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.
+#
+
+# Set root logger level to OFF to not flood build logs
+# set manually to INFO for debugging purposes
+rootLogger.level = OFF
+rootLogger.appenderRef.test.ref = TestLogger
+
+appender.testlogger.name = TestLogger
+appender.testlogger.type = CONSOLE
+appender.testlogger.target = SYSTEM_ERR
+appender.testlogger.layout.type = PatternLayout
+appender.testlogger.layout.pattern = %-4r [%t] %-5p %c %x - %m%n
+
+# suppress the duplicated logger extension
+logger.flink.name = org.apache.flink.util.TestLoggerExtension
+logger.flink.level = OFF

fluss-flink/fluss-flink-1.20/src/test/resources/log4j2-test.properties

@@ -0,0 +1,19 @@
+#
+# 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.
+#
+
+org.apache.fluss.testutils.common.TestLoggerExtension

fluss-flink/fluss-flink-2.1/src/test/resources/META-INF/services/org.junit.jupiter.api.extension.Extension

@@ -0,0 +1,32 @@
+#
+#  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.
+#
+
+# Set root logger level to OFF to not flood build logs
+# set manually to INFO for debugging purposes
+rootLogger.level = OFF
+rootLogger.appenderRef.test.ref = TestLogger
+
+appender.testlogger.name = TestLogger
+appender.testlogger.type = CONSOLE
+appender.testlogger.target = SYSTEM_ERR
+appender.testlogger.layout.type = PatternLayout
+appender.testlogger.layout.pattern = %-4r [%t] %-5p %c %x - %m%n
+
+# suppress the duplicated logger extension
+logger.flink.name = org.apache.flink.util.TestLoggerExtension
+logger.flink.level = OFF

fluss-flink/fluss-flink-2.1/src/test/resources/log4j2-test.properties

@@ -20,7 +20,6 @@
 import org.apache.fluss.exception.ApiException;
 import org.apache.fluss.exception.AuthenticationException;
 import org.apache.fluss.exception.AuthorizationException;
-import org.apache.fluss.exception.CacheMissException;
 import org.apache.fluss.exception.CorruptMessageException;
 import org.apache.fluss.exception.CorruptRecordException;
 import org.apache.fluss.exception.DatabaseAlreadyExistException;
@@ -225,9 +224,6 @@ public enum Errors {
             55,
             "The new ISR contains at least one ineligible replica.",
             IneligibleReplicaException::new),
-    CACHE_MISS(
-            56, "The requested metadata is not available in local cache.", CacheMissException::new);
-            IneligibleReplicaException::new),
     INVALID_ALTER_TABLE_EXCEPTION(
             56, "The alter table is invalid.", InvalidAlterTableException::new);
 

fluss-rpc/src/main/java/org/apache/fluss/rpc/protocol/Errors.java

@@ -176,10 +176,6 @@ message MetadataRequest {
   // metadata request
   // todo: we won't need the assumption after we introduce metadata cache in server
   repeated int64 partitions_id = 3 [packed = true];
-  
-  // if cache_only is true, return immediately if metadata is not in local cache
-  // instead of waiting for ZooKeeper access to complete
-  optional bool cache_only = 4;
 }
 
 message MetadataResponse {