private TSStatus cacheAndInsertLatestNode(DeserializedBatchIndexedConsensusRequest request) {
      logger.debug(
          "cacheAndInsert start: source = {}, region = {}, queue size {}, startSyncIndex = {}, endSyncIndex = {}",
          sourcePeerId,
          consensusGroupId,
          requestCache.size(),
          request.getStartSyncIndex(),
          request.getEndSyncIndex());
      queueLock.lock();
      try {
        long insertStartTime = System.nanoTime();
        requestCache.add(request);
        // If the peek is not hold by current thread, it should notify the corresponding thread to
        // process the peek when the queue is full
        if (requestCache.size() == config.getReplication().getMaxPendingBatchesNum()
            && requestCache.peek() != null
            && requestCache.peek().getStartSyncIndex() != request.getStartSyncIndex()) {
          queueSortCondition.signalAll();
        }
        while (true) {
          // If current InsertNode is the next target InsertNode, write it
          if (request.getStartSyncIndex() == nextSyncIndex) {
            requestCache.remove(request);
            nextSyncIndex = request.getEndSyncIndex() + 1;
            break;
          }
          // If all write thread doesn't hit nextSyncIndex and the heap is full, write
          // the peek request. This is used to keep the whole write correct when nextSyncIndex
          // is not set. We won't persist the value of nextSyncIndex to reduce the complexity.
          // There are some cases that nextSyncIndex is not set:
          //   1. When the system was just started
          //   2. When some exception occurs during SyncLog
          if (requestCache.size() == config.getReplication().getMaxPendingBatchesNum()
              && requestCache.peek() != null
              && requestCache.peek().getStartSyncIndex() == request.getStartSyncIndex()) {
            requestCache.remove();
            nextSyncIndex = request.getEndSyncIndex() + 1;
            break;
          }
          try {
            boolean timeout =
                !queueSortCondition.await(
                    config.getReplication().getMaxWaitingTimeForWaitBatchInMs(),
                    TimeUnit.MILLISECONDS);
            // although the timeout is triggered, current thread cannot write its request
            // if current thread does not hold the peek request. And there should be some
            // other thread who hold the peek request. In this scenario, current thread
            // should go into await again and wait until its request becoming peek request
            if (timeout
                && requestCache.peek() != null
                && requestCache.peek().getStartSyncIndex() == request.getStartSyncIndex()) {
              // current thread hold the peek request thus it can write the peek immediately.
              logger.info(
                  "waiting target request timeout. current index: {}, target index: {}",
                  request.getStartSyncIndex(),
                  nextSyncIndex);
              requestCache.remove(request);
              nextSyncIndex = Math.max(nextSyncIndex, request.getEndSyncIndex() + 1);
              break;
            }
          } catch (InterruptedException e) {
            logger.warn(
                "current waiting is interrupted. SyncIndex: {}. Exception: ",
                request.getStartSyncIndex(),
                e);
            Thread.currentThread().interrupt();
            break;
          }
        }
        long sortTime = System.nanoTime();
        ioTConsensusServerMetrics.recordSortCost(sortTime - insertStartTime);
        List<TSStatus> subStatus = new LinkedList<>();
        for (IConsensusRequest insertNode : request.getInsertNodes()) {
          insertNode.markAsGeneratedByRemoteConsensusLeader();
          subStatus.add(stateMachine.write(insertNode));
        }
        long applyTime = System.nanoTime();
        ioTConsensusServerMetrics.recordApplyCost(applyTime - sortTime);
        queueSortCondition.signalAll();
        logger.debug(
            "cacheAndInsert end: source = {}, region = {}, queue size {}, startSyncIndex = {}, endSyncIndex = {}, sortTime = {}ms, applyTime = {}ms",
            sourcePeerId,
            consensusGroupId,
            requestCache.size(),
            request.getStartSyncIndex(),
            request.getEndSyncIndex(),
            TimeUnit.NANOSECONDS.toMillis(sortTime - insertStartTime),
            TimeUnit.NANOSECONDS.toMillis(applyTime - sortTime));
        return new TSStatus().setSubStatus(subStatus);
      } finally {
        queueLock.unlock();
      }
    }
This method is used for write of IoTConsensus SyncLog. By this method, we can keep write
order in follower the same as the leader. And besides order insurance, we can make the
deserialization of PlanNode to be concurrent
iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/iot/IoTConsensusServerImpl.java