Fixed Ratis query not retrying when DataNode restarts

iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/exception/RatisReadUnavailableException.java

@@ -22,7 +22,7 @@
 /** RaftServer is unable to serve linearizable read requests. */
 public class RatisReadUnavailableException extends ConsensusException {
 
-  public static final String RATIS_READ_UNAVAILABLE =
+  private static final String RATIS_READ_UNAVAILABLE =
       "Raft Server cannot serve read requests now (leader is unknown or under recovery). "
           + "Please try read later: ";
 

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

@@ -308,7 +308,7 @@ public TSendFragmentInstanceResp sendFragmentInstance(TSendFragmentInstanceReq r
     TSendFragmentInstanceResp resp = new TSendFragmentInstanceResp();
     resp.setAccepted(executionResult.isAccepted());
     resp.setMessage(executionResult.getMessage());
-    // TODO
+    resp.setNeedRetry(executionResult.isNeedRetry());
     return resp;
   }
 

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/executor/RegionExecutionResult.java

@@ -28,6 +28,7 @@ public class RegionExecutionResult {
   private String message;
 
   private TSStatus status;
+  private boolean needRetry;
 
   public boolean isAccepted() {
     return accepted;
@@ -52,4 +53,12 @@ public TSStatus getStatus() {
   public void setStatus(TSStatus status) {
     this.status = status;
   }
+
+  public boolean isNeedRetry() {
+    return needRetry;
+  }
+
+  public void setNeedRetry(boolean needRetry) {
+    this.needRetry = needRetry;
+  }
 }

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/executor/RegionReadExecutor.java

@@ -32,6 +32,10 @@
 import org.apache.iotdb.db.storageengine.dataregion.VirtualDataRegion;
 import org.apache.iotdb.db.utils.SetThreadName;
 
+import org.apache.ratis.protocol.exceptions.NotLeaderException;
+import org.apache.ratis.protocol.exceptions.ReadException;
+import org.apache.ratis.protocol.exceptions.ReadIndexException;
+import org.apache.ratis.protocol.exceptions.ServerNotReadyException;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -69,14 +73,14 @@ public RegionReadExecutor(
   public RegionExecutionResult execute(
       ConsensusGroupId groupId, FragmentInstance fragmentInstance) {
     // execute fragment instance in state machine
-    DataSet readResponse;
+    RegionExecutionResult resp = new RegionExecutionResult();
     try (SetThreadName threadName = new SetThreadName(fragmentInstance.getId().getFullId())) {
+      DataSet readResponse;
       if (groupId instanceof DataRegionId) {
         readResponse = dataRegionConsensus.read(groupId, fragmentInstance);
       } else {
         readResponse = schemaRegionConsensus.read(groupId, fragmentInstance);
       }
-      RegionExecutionResult resp = new RegionExecutionResult();
       if (readResponse == null) {
         LOGGER.error(RESPONSE_NULL_ERROR_MSG);
         resp.setAccepted(false);
@@ -87,11 +91,16 @@ public RegionExecutionResult execute(
         resp.setMessage(info.getMessage());
       }
       return resp;
-    } catch (Throwable t) {
-      LOGGER.error("Execute FragmentInstance in ConsensusGroup {} failed.", groupId, t);
-      RegionExecutionResult resp = new RegionExecutionResult();
-      resp.setAccepted(false);
-      resp.setMessage(String.format(ERROR_MSG_FORMAT, t.getMessage()));
+    } catch (Throwable e) {
+      LOGGER.error("Execute FragmentInstance in ConsensusGroup {} failed.", groupId, e);
+      resp.setMessage(String.format(ERROR_MSG_FORMAT, e.getMessage()));
+      Throwable t = e.getCause();
+      if (t instanceof ReadException
+          || t instanceof ReadIndexException
+          || t instanceof NotLeaderException
+          || t instanceof ServerNotReadyException) {
+        resp.setNeedRetry(true);
+      }
       return resp;
     }
   }

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/scheduler/FragmentInstanceDispatcherImpl.java

@@ -301,8 +301,8 @@ private void dispatchRemoteHelper(FragmentInstance instance, TEndPoint endPoint)
               client.sendFragmentInstance(sendFragmentInstanceReq);
           if (!sendFragmentInstanceResp.accepted) {
             logger.warn(sendFragmentInstanceResp.message);
-            if (sendFragmentInstanceResp.message.contains(
-                RatisReadUnavailableException.RATIS_READ_UNAVAILABLE)) {
+            if (sendFragmentInstanceResp.isSetNeedRetry()
+                && sendFragmentInstanceResp.isNeedRetry()) {
               throw new RatisReadUnavailableException(sendFragmentInstanceResp.message);
             } else {
               throw new FragmentInstanceDispatchException(

iotdb-protocol/thrift-datanode/src/main/thrift/datanode.thrift

@@ -116,6 +116,7 @@ struct TSendFragmentInstanceReq {
 struct TSendFragmentInstanceResp {
   1: required bool accepted
   2: optional string message
+  3: optional bool needRetry
 }
 
 struct TSendSinglePlanNodeReq {