Fix pipe runtime meta push blocking user operations

iotdb-client/service-rpc/src/main/java/org/apache/iotdb/rpc/RpcUtils.java

@@ -151,16 +151,16 @@ public static void verifySuccessWithRedirectionForMultiDevices(
   }
 
   public static void verifySuccess(List<TSStatus> statuses) throws BatchExecutionException {
-    StringBuilder errMsgs =
-        new StringBuilder().append(TSStatusCode.MULTIPLE_ERROR.getStatusCode()).append(": ");
+    StringBuilder errMsgs = new StringBuilder();
     for (TSStatus status : statuses) {
       if (status.getCode() != TSStatusCode.SUCCESS_STATUS.getStatusCode()
           && status.getCode() != TSStatusCode.REDIRECTION_RECOMMEND.getStatusCode()) {
         errMsgs.append(status.getMessage()).append("; ");
       }
     }
     if (errMsgs.length() > 0) {
-      throw new BatchExecutionException(statuses, errMsgs.toString());
+      throw new BatchExecutionException(
+          statuses, TSStatusCode.MULTIPLE_ERROR.getStatusCode() + ": " + errMsgs);
     }
   }
 
@@ -181,9 +181,9 @@ public static TSStatus getStatus(List<TSStatus> statusList) {
       for (TSStatus subStatus : statusList) {
         if (subStatus.getCode() != TSStatusCode.SUCCESS_STATUS.getStatusCode()
             && subStatus.getCode() != TSStatusCode.REDIRECTION_RECOMMEND.getStatusCode()) {
-          if (!msgSet.contains(status)) {
-            errMsg.append(status).append("; ");
-            msgSet.add(status);
+          if (!msgSet.contains(subStatus)) {
+            errMsg.append(subStatus).append("; ");
+            msgSet.add(subStatus);
           }
         }
       }

iotdb-client/service-rpc/src/test/java/org/apache/iotdb/rpc/RpcUtilsTest.java

@@ -19,11 +19,15 @@
 
 package org.apache.iotdb.rpc;
 
+import org.apache.iotdb.common.rpc.thrift.TSStatus;
+
 import org.junit.Assert;
 import org.junit.Test;
 
 import java.time.ZoneId;
 import java.time.format.DateTimeFormatter;
+import java.util.Arrays;
+import java.util.Collections;
 
 public class RpcUtilsTest {
 
@@ -74,4 +78,25 @@ public void testIsSetSqlDialect() {
     Assert.assertFalse(RpcUtils.isSetSqlDialect("setsql_dialect =table"));
     Assert.assertFalse(RpcUtils.isSetSqlDialect("set           sql_dia"));
   }
+
+  @Test
+  public void testVerifySuccessListAllowsSuccessfulStatuses() throws BatchExecutionException {
+    RpcUtils.verifySuccess(
+        Arrays.asList(
+            RpcUtils.getStatus(TSStatusCode.SUCCESS_STATUS),
+            RpcUtils.getStatus(TSStatusCode.REDIRECTION_RECOMMEND)));
+  }
+
+  @Test
+  public void testVerifySuccessListThrowsOnFailure() {
+    TSStatus failedStatus = RpcUtils.getStatus(TSStatusCode.EXECUTE_STATEMENT_ERROR, "failed");
+
+    try {
+      RpcUtils.verifySuccess(Collections.singletonList(failedStatus));
+      Assert.fail("Expected BatchExecutionException");
+    } catch (BatchExecutionException e) {
+      Assert.assertEquals(Collections.singletonList(failedStatus), e.getStatusList());
+      Assert.assertTrue(e.getMessage().contains("failed"));
+    }
+  }
 }

iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/client/async/AsyncAINodeHeartbeatClientPool.java

@@ -54,8 +54,8 @@ public void getAINodeHeartBeat(
       client = clientManager.borrowClient(endPoint);
       client.getAIHeartbeat(req, handler);
       dispatched = true;
-    } catch (Exception ignore) {
-      // Just ignore
+    } catch (Exception e) {
+      handleError(handler, e);
     } finally {
       // After the async call is dispatched, the client's onComplete/onError callback is
       // responsible for returning the client. If the RPC was not dispatched (exception
@@ -67,6 +67,14 @@ public void getAINodeHeartBeat(
     }
   }
 
+  private void handleError(final AINodeHeartbeatHandler handler, final Exception e) {
+    try {
+      handler.onError(e);
+    } catch (final Exception ignore) {
+      // Ignore handler failures in heartbeat best-effort path.
+    }
+  }
+
   private static class AsyncAINodeHeartbeatClientPoolHolder {
 
     private static final AsyncAINodeHeartbeatClientPool INSTANCE =

iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/client/async/AsyncConfigNodeHeartbeatClientPool.java

@@ -55,8 +55,8 @@ public void getConfigNodeHeartBeat(
       client = clientManager.borrowClient(endPoint);
       client.getConfigNodeHeartBeat(heartbeatReq, handler);
       dispatched = true;
-    } catch (Exception ignore) {
-      // Just ignore
+    } catch (Exception e) {
+      handleError(handler, e);
     } finally {
       // After the async call is dispatched, the client's onComplete/onError callback is
       // responsible for returning the client. If the RPC was not dispatched (exception
@@ -68,6 +68,14 @@ public void getConfigNodeHeartBeat(
     }
   }
 
+  private void handleError(final ConfigNodeHeartbeatHandler handler, final Exception e) {
+    try {
+      handler.onError(e);
+    } catch (final Exception ignore) {
+      // Ignore handler failures in heartbeat best-effort path.
+    }
+  }
+
   private static class AsyncConfigNodeHeartbeatClientPoolHolder {
 
     private static final AsyncConfigNodeHeartbeatClientPool INSTANCE =

iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/client/async/AsyncDataNodeHeartbeatClientPool.java

@@ -56,8 +56,8 @@ public void getDataNodeHeartBeat(
       client = clientManager.borrowClient(endPoint);
       client.getDataNodeHeartBeat(req, handler);
       dispatched = true;
-    } catch (Exception ignore) {
-      // Just ignore
+    } catch (Exception e) {
+      handleError(handler, e);
     } finally {
       returnClientIfNotDispatched(endPoint, client, dispatched);
     }
@@ -72,7 +72,7 @@ public void writeAuditLog(
       client.writeAuditLog(req, handler);
       dispatched = true;
     } catch (Exception e) {
-      // Just ignore
+      handleError(handler, e);
     } finally {
       returnClientIfNotDispatched(endPoint, client, dispatched);
     }
@@ -89,6 +89,22 @@ private void returnClientIfNotDispatched(
     }
   }
 
+  private void handleError(final DataNodeHeartbeatHandler handler, final Exception e) {
+    try {
+      handler.onError(e);
+    } catch (final Exception ignore) {
+      // Ignore handler failures in heartbeat best-effort path.
+    }
+  }
+
+  private void handleError(final DataNodeWriteAuditLogHandler handler, final Exception e) {
+    try {
+      handler.onError(e);
+    } catch (final Exception ignore) {
+      // Ignore handler failures in audit-log best-effort path.
+    }
+  }
+
   private static class AsyncDataNodeHeartbeatClientPoolHolder {
 
     private static final AsyncDataNodeHeartbeatClientPool INSTANCE =

iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/client/async/handlers/rpc/CheckTimeSeriesExistenceRPCHandler.java

@@ -83,11 +83,11 @@ public void onError(Exception e) {
             + e.getMessage();
     LOGGER.error(errorMsg);
 
-    countDownLatch.countDown();
     TCheckTimeSeriesExistenceResp resp = new TCheckTimeSeriesExistenceResp();
     resp.setStatus(
         new TSStatus(
             RpcUtils.getStatus(TSStatusCode.EXECUTE_STATEMENT_ERROR.getStatusCode(), errorMsg)));
     responseMap.put(requestId, resp);
+    countDownLatch.countDown();
   }
 }

iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/client/async/handlers/rpc/CountPathsUsingTemplateRPCHandler.java

@@ -83,11 +83,11 @@ public void onError(Exception e) {
             + e.getMessage();
     LOGGER.error(errorMsg);
 
-    countDownLatch.countDown();
     TCountPathsUsingTemplateResp resp = new TCountPathsUsingTemplateResp();
     resp.setStatus(
         new TSStatus(
             RpcUtils.getStatus(TSStatusCode.EXECUTE_STATEMENT_ERROR.getStatusCode(), errorMsg)));
     responseMap.put(requestId, resp);
+    countDownLatch.countDown();
   }
 }

iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/client/async/handlers/rpc/FetchSchemaBlackListRPCHandler.java

@@ -83,11 +83,11 @@ public void onError(Exception e) {
             + e.getMessage();
     LOGGER.error(errorMsg);
 
-    countDownLatch.countDown();
     TFetchSchemaBlackListResp resp = new TFetchSchemaBlackListResp();
     resp.setStatus(
         new TSStatus(
             RpcUtils.getStatus(TSStatusCode.EXECUTE_STATEMENT_ERROR.getStatusCode(), errorMsg)));
     responseMap.put(requestId, resp);
+    countDownLatch.countDown();
   }
 }

iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/client/async/handlers/rpc/SchemaUpdateRPCHandler.java

@@ -77,10 +77,10 @@ public void onError(Exception e) {
             + e.getMessage();
     LOGGER.warn(errorMsg);
 
-    countDownLatch.countDown();
     responseMap.put(
         requestId,
         new TSStatus(
             RpcUtils.getStatus(TSStatusCode.EXECUTE_STATEMENT_ERROR.getStatusCode(), errorMsg)));
+    countDownLatch.countDown();
   }
 }

iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/client/async/handlers/rpc/subscription/CheckSchemaRegionUsingTemplateRPCHandler.java

@@ -85,11 +85,11 @@ public void onError(Exception e) {
             + e.getMessage();
     LOGGER.error(errorMsg);
 
-    countDownLatch.countDown();
     TCheckSchemaRegionUsingTemplateResp resp = new TCheckSchemaRegionUsingTemplateResp();
     resp.setStatus(
         new TSStatus(
             RpcUtils.getStatus(TSStatusCode.EXECUTE_STATEMENT_ERROR.getStatusCode(), errorMsg)));
     responseMap.put(requestId, resp);
+    countDownLatch.countDown();
   }
 }

iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/load/service/TopologyService.java

@@ -44,6 +44,7 @@
 import org.apache.iotdb.confignode.manager.load.subscriber.IClusterStatusSubscriber;
 import org.apache.iotdb.confignode.manager.load.subscriber.NodeStatisticsChangeEvent;
 import org.apache.iotdb.mpp.rpc.thrift.TUpdateClusterTopologyReq;
+import org.apache.iotdb.rpc.TSStatusCode;
 
 import org.apache.ratis.util.AwaitForSignal;
 import org.apache.tsfile.utils.Pair;
@@ -72,6 +73,7 @@
 public class TopologyService implements Runnable, IClusterStatusSubscriber {
   private static final Logger LOGGER = LoggerFactory.getLogger(TopologyService.class);
   private static final int SAMPLING_WINDOW_SIZE = 100;
+  private static final int TOPOLOGY_PROBING_RETRY_NUM = 1;
 
   private final ExecutorService topologyThread =
       IoTDBThreadPoolFactory.newSingleThreadExecutor(
@@ -222,7 +224,7 @@ private synchronized void topologyProbing() {
                 nodeLocations,
                 proberLocationMap);
     CnToDnInternalServiceAsyncRequestManager.getInstance()
-        .sendAsyncRequestWithTimeoutInMs(dataNodeAsyncRequestContext, timeout);
+        .sendAsyncRequest(dataNodeAsyncRequestContext, TOPOLOGY_PROBING_RETRY_NUM, timeout, true);
     final List<TTestConnectionResult> results = new ArrayList<>();
     dataNodeAsyncRequestContext
         .getResponseMap()
@@ -360,15 +362,18 @@ private void pushTopologyToDataNodes(
     }
 
     CnToDnInternalServiceAsyncRequestManager.getInstance()
-        .sendAsyncRequestWithTimeoutInMs(context, CONF.getTopologyProbingBaseIntervalInMs());
+        .sendAsyncRequest(
+            context, TOPOLOGY_PROBING_RETRY_NUM, CONF.getTopologyProbingBaseIntervalInMs(), true);
 
     context
         .getResponseMap()
         .forEach(
             (nodeId, resp) -> {
-              Set<Integer> reachableSet =
-                  computedTopology.getOrDefault(nodeId, Collections.emptySet());
-              lastPushedTopology.put(nodeId, new HashSet<>(reachableSet));
+              if (resp.getCode() == TSStatusCode.SUCCESS_STATUS.getStatusCode()) {
+                Set<Integer> reachableSet =
+                    computedTopology.getOrDefault(nodeId, Collections.emptySet());
+                lastPushedTopology.put(nodeId, new HashSet<>(reachableSet));
+              }
             });
   }
 

iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/pipe/coordinator/runtime/heartbeat/PipeHeartbeatScheduler.java

@@ -51,6 +51,7 @@ public class PipeHeartbeatScheduler {
       PipeConfig.getInstance().isSeperatedPipeHeartbeatEnabled();
   private static final long HEARTBEAT_INTERVAL_SECONDS =
       PipeConfig.getInstance().getPipeHeartbeatIntervalSecondsForCollectingPipeMeta();
+  private static final int PIPE_HEARTBEAT_RETRY_NUM = 1;
 
   private static final ScheduledExecutorService HEARTBEAT_EXECUTOR =
       IoTDBThreadPoolFactory.newSingleThreadScheduledExecutor(
@@ -100,12 +101,8 @@ private synchronized void heartbeat() {
         new DataNodeAsyncRequestContext<>(
             CnToDnAsyncRequestType.PIPE_HEARTBEAT, request, dataNodeLocationMap);
     CnToDnInternalServiceAsyncRequestManager.getInstance()
-        .sendAsyncRequestToNodeWithRetryAndTimeoutInMs(
-            clientHandler,
-            PipeConfig.getInstance().getPipeHeartbeatIntervalSecondsForCollectingPipeMeta()
-                * 1000L
-                * 2
-                / 3);
+        .sendAsyncRequest(
+            clientHandler, PIPE_HEARTBEAT_RETRY_NUM, getPipeHeartbeatRequestTimeoutInMs(), true);
     clientHandler
         .getResponseMap()
         .forEach(
@@ -134,6 +131,10 @@ private synchronized void heartbeat() {
     }
   }
 
+  private static long getPipeHeartbeatRequestTimeoutInMs() {
+    return TimeUnit.SECONDS.toMillis(HEARTBEAT_INTERVAL_SECONDS) * 2 / 3;
+  }
+
   public synchronized void stop() {
     if (IS_SEPERATED_PIPE_HEARTBEAT_ENABLED && heartbeatFuture != null) {
       heartbeatFuture.cancel(false);