Invalidate schema cache when data region leader change

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

@@ -68,6 +68,7 @@
 import org.apache.iotdb.mpp.rpc.thrift.TDropTriggerInstanceReq;
 import org.apache.iotdb.mpp.rpc.thrift.TFetchSchemaBlackListReq;
 import org.apache.iotdb.mpp.rpc.thrift.TInactiveTriggerInstanceReq;
+import org.apache.iotdb.mpp.rpc.thrift.TInvalidateCacheReq;
 import org.apache.iotdb.mpp.rpc.thrift.TInvalidateMatchedSchemaCacheReq;
 import org.apache.iotdb.mpp.rpc.thrift.TPipeHeartbeatReq;
 import org.apache.iotdb.mpp.rpc.thrift.TPushConsumerGroupMetaReq;
@@ -260,6 +261,11 @@ protected void initActionMapBuilder() {
         (req, client, handler) ->
             client.fetchSchemaBlackList(
                 (TFetchSchemaBlackListReq) req, (FetchSchemaBlackListRPCHandler) handler));
+    actionMapBuilder.put(
+        CnToDnRequestType.INVALIDATE_SCHEMA_CACHE,
+        (req, client, handler) ->
+            client.invalidateSchemaCache(
+                (TInvalidateCacheReq) req, (DataNodeTSStatusRPCHandler) handler));
     actionMapBuilder.put(
         CnToDnRequestType.INVALIDATE_MATCHED_SCHEMA_CACHE,
         (req, client, handler) ->

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

@@ -200,6 +200,7 @@ public static DataNodeAsyncRequestRPCHandler<?> buildHandler(
       case LOAD_CONFIGURATION:
       case SET_SYSTEM_STATUS:
       case UPDATE_REGION_ROUTE_MAP:
+      case INVALIDATE_SCHEMA_CACHE:
       case INVALIDATE_MATCHED_SCHEMA_CACHE:
       case UPDATE_TEMPLATE:
       case UPDATE_TABLE:

iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/load/balancer/RouteBalancer.java

@@ -47,6 +47,7 @@
 import org.apache.iotdb.confignode.manager.node.NodeManager;
 import org.apache.iotdb.confignode.manager.partition.PartitionManager;
 import org.apache.iotdb.consensus.ConsensusFactory;
+import org.apache.iotdb.mpp.rpc.thrift.TInvalidateCacheReq;
 import org.apache.iotdb.mpp.rpc.thrift.TRegionLeaderChangeReq;
 import org.apache.iotdb.mpp.rpc.thrift.TRegionLeaderChangeResp;
 import org.apache.iotdb.mpp.rpc.thrift.TRegionRouteReq;
@@ -59,6 +60,7 @@
 import java.util.List;
 import java.util.Map;
 import java.util.Objects;
+import java.util.Set;
 import java.util.TreeMap;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicBoolean;
@@ -245,7 +247,40 @@ private void balanceRegionLeader(
         }
       }
     }
+
     getLoadManager().forceUpdateConsensusGroupCache(successTransferMap);
+
+    invalidateSchemaCacheOfOldLeaders(currentLeaderMap, successTransferMap.keySet());
+  }
+
+  private void invalidateSchemaCacheOfOldLeaders(
+      Map<TConsensusGroupId, Integer> oldLeaderMap, Set<TConsensusGroupId> successTransferSet) {
+    DataNodeAsyncRequestContext<TInvalidateCacheReq, TSStatus> invalidateSchemaCacheRequestHandler =
+        new DataNodeAsyncRequestContext<>(CnToDnRequestType.INVALIDATE_SCHEMA_CACHE);
+    AtomicInteger requestIndex = new AtomicInteger(0);
+    oldLeaderMap.entrySet().stream()
+        .filter(entry -> TConsensusGroupType.DataRegion == entry.getKey().getType())
+        .filter(entry -> successTransferSet.contains(entry.getKey()))
+        .forEach(
+            entry -> {
+              // set target
+              Integer dataNodeId = entry.getValue();
+              TDataNodeLocation dataNodeLocation =
+                  getNodeManager().getRegisteredDataNode(dataNodeId).getLocation();
+              if (dataNodeLocation == null) {
+                LOGGER.warn("DataNodeLocation is null, datanodeId {}", dataNodeId);
+                return;
+              }
+              invalidateSchemaCacheRequestHandler.putNodeLocation(
+                  requestIndex.getAndIncrement(), dataNodeLocation);
+              // set req
+              TConsensusGroupId consensusGroupId = entry.getKey();
+              String database = getPartitionManager().getRegionStorageGroup(consensusGroupId);
+              invalidateSchemaCacheRequestHandler.putRequest(
+                  requestIndex.get(), new TInvalidateCacheReq(true, database));
+            });
+    CnToDnInternalServiceAsyncRequestManager.getInstance()
+        .sendAsyncRequest(invalidateSchemaCacheRequestHandler);
   }
 
   public synchronized void balanceRegionLeaderAndPriority() {

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/protocol/thrift/impl/DataNodeInternalRPCServiceImpl.java

@@ -522,6 +522,7 @@ public TSStatus invalidateSchemaCache(TInvalidateCacheReq req) {
       String database = req.getFullPath().substring(5);
       DataNodeTableCache.getInstance().invalid(database);
       TableDeviceSchemaFetcher.getInstance().getTableDeviceCache().invalidate(database);
+      LOGGER.info("Schema cache of {} has been invalidated", req.getFullPath());
       return new TSStatus(TSStatusCode.SUCCESS_STATUS.getStatusCode());
     } finally {
       DataNodeSchemaCache.getInstance().releaseWriteLock();

iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/client/request/AsyncRequestManager.java

@@ -94,7 +94,7 @@ public final void sendAsyncRequest(
     sendAsyncRequest(requestContext, 1, null);
   }
 
-  private void sendAsyncRequest(
+  public void sendAsyncRequest(
       AsyncRequestContext<?, ?, RequestType, NodeLocation> requestContext,
       int retryNum,
       Long timeoutInMs) {