private synchronized void trySubmitGetDataBlocksTask() {
    if (aborted || closed) {
      return;
    }
    if (blockedOnMemory != null && !blockedOnMemory.isDone()) {
      return;
    }

    final int startSequenceId = nextSequenceId;
    int endSequenceId = nextSequenceId;
    long reservedBytes = 0L;
    Pair<ListenableFuture<Void>, Boolean> pair = null;
    long blockedSize = 0L;
    while (sequenceIdToDataBlockSize.containsKey(endSequenceId)) {
      Long bytesToReserve = sequenceIdToDataBlockSize.get(endSequenceId);
      if (bytesToReserve == null) {
        throw new IllegalStateException("Data block size is null.");
      }
      pair =
          localMemoryManager
              .getQueryPool()
              .reserve(
                  localFragmentInstanceId.getQueryId(),
                  fullFragmentInstanceId,
                  localPlanNodeId,
                  bytesToReserve,
                  maxBytesCanReserve);
      bufferRetainedSizeInBytes += bytesToReserve;
      endSequenceId += 1;
      reservedBytes += bytesToReserve;
      if (!Boolean.TRUE.equals(pair.right)) {
        blockedSize = bytesToReserve;
        break;
      }
    }

    if (pair == null) {
      // Next data block not generated yet. Do nothing.
      return;
    }
    nextSequenceId = endSequenceId;

    if (!Boolean.TRUE.equals(pair.right)) {
      endSequenceId--;
      reservedBytes -= blockedSize;
      // The future being not completed indicates,
      //   1. Memory has been reserved for blocks in [startSequenceId, endSequenceId).
      //   2. Memory reservation for block whose sequence ID equals endSequenceId - 1 is blocked.
      //   3. Have not reserve memory for the rest of blocks.
      //
      //  startSequenceId          endSequenceId - 1  endSequenceId
      //         |-------- reserved --------|--- blocked ---|--- not reserved ---|

      // Schedule another call of trySubmitGetDataBlocksTask for the rest of blocks.
      blockedOnMemory = pair.left;
      final int blockedSequenceId = endSequenceId;
      final long blockedRetainedSize = blockedSize;
      blockedOnMemory.addListener(
          () ->
              executorService.submit(
                  new GetDataBlocksTask(
                      blockedSequenceId, blockedSequenceId + 1, blockedRetainedSize)),
          executorService);
    }

    if (endSequenceId > startSequenceId) {
      executorService.submit(new GetDataBlocksTask(startSequenceId, endSequenceId, reservedBytes));
    }
  }
this is set to true after calling isBlocked() at least once which indicates that this
SourceHandle needs to output data.
iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/exchange/source/SourceHandle.java