@Override
  public synchronized ByteBuffer getSerializedTsBlock() {
    long startTime = System.nanoTime();
    try (SetThreadName sourceHandleName = new SetThreadName(threadName)) {
      checkState();

      if (!blocked.isDone()) {
        throw new IllegalStateException("Source handle is blocked.");
      }

      ByteBuffer tsBlock = sequenceIdToTsBlock.remove(currSequenceId);
      if (tsBlock == null) {
        return null;
      }
      long retainedSize = sequenceIdToDataBlockSize.remove(currSequenceId);
      if (LOGGER.isDebugEnabled()) {
        LOGGER.debug("[GetTsBlockFromBuffer] sequenceId:{}, size:{}", currSequenceId, retainedSize);
      }
      currSequenceId += 1;
      bufferRetainedSizeInBytes -= retainedSize;
      localMemoryManager
          .getQueryPool()
          .free(
              localFragmentInstanceId.getQueryId(),
              fullFragmentInstanceId,
              localPlanNodeId,
              retainedSize);

      if (sequenceIdToTsBlock.isEmpty() && !isFinished()) {
        if (LOGGER.isDebugEnabled()) {
          LOGGER.debug("[WaitForMoreTsBlock]");
        }
        blocked = SettableFuture.create();
      }
      if (isFinished()) {
        sourceHandleListener.onFinished(this);
      }
      trySubmitGetDataBlocksTask();
      return tsBlock;
    } finally {
      DATA_EXCHANGE_COST_METRIC_SET.recordDataExchangeCost(
          SOURCE_HANDLE_GET_TSBLOCK_REMOTE, System.nanoTime() - startTime);
    }
  }
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