Merge remote-tracking branch 'origin/master' into codex/generic-ssl-config

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

Author: HTHou

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 {
 
@@ -75,6 +79,27 @@ public void testIsSetSqlDialect() {
     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"));
+    }
+  }
+
   @Test
   public void testStandardTlsProtocol() {
     Assert.assertEquals("TLS", RpcSslUtils.normalizeStandardTlsProtocol(null));

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();
   }
 }