private <T> RetryResult<T> callWithRetryAndReconnect(
      TFunction<T> rpc, Predicate<T> shouldRetry, Predicate<T> forceReconnect) {
    TException lastTException = null;
    T result = null;
    int retryAttempt;
    int maxRetryCountRead = 10;
    for (retryAttempt = 0; retryAttempt <= maxRetryCountRead; retryAttempt++) {
      // 1. try to execute the rpc
      try {
        result = rpc.run();
        lastTException = null;
      } catch (TException e) {
        result = null;
        lastTException = e;
      }

      // success, return immediately
      if (result != null && !shouldRetry.test(result)) {
        return new RetryResult<>(result, null, retryAttempt);
      }

      logger.debug(
          "Retry attempt #{}, result {}, exception {}", retryAttempt, result, lastTException);
      // prepare for the next retry
      if (lastTException != null
          || !availableNodes.get().contains(this.endPoint)
          || (result != null && forceReconnect.test(result))) {
        // 1. the current datanode is unreachable (TException)
        // 2. the current datanode is partitioned with other nodes (not in availableNodes)
        // 3. asymmetric network partition
        logger.debug("Retry attempt #{}, Reconnecting to other datanode", retryAttempt);
        reconnect();
      }
      try {
        TimeUnit.MILLISECONDS.sleep(retryIntervalInMs);
      } catch (InterruptedException e) {
        Thread.currentThread().interrupt();
        logger.warn(
            "Thread {} was interrupted during retry {} with wait time {} ms. Exiting retry loop.",
            Thread.currentThread().getName(),
            retryAttempt,
            retryIntervalInMs);
        break;
      }
    }

    return new RetryResult<>(result, lastTException, retryAttempt);
  }
reconnect if the remote datanode is unreachable retry if the status is set to needRetry
iotdb-client/session/src/main/java/org/apache/iotdb/session/SessionConnection.java