@Override
  public synchronized void close() {
    try (SetThreadName sourceHandleName = new SetThreadName(threadName)) {
      if (aborted || closed) {
        return;
      }
      if (blocked != null && !blocked.isDone()) {
        blocked.set(null);
      }
      if (blockedOnMemory != null) {
        bufferRetainedSizeInBytes -= localMemoryManager.getQueryPool().tryCancel(blockedOnMemory);
      }
      sequenceIdToDataBlockSize.clear();
      if (bufferRetainedSizeInBytes > 0) {
        localMemoryManager
            .getQueryPool()
            .free(
                localFragmentInstanceId.getQueryId(),
                fullFragmentInstanceId,
                localPlanNodeId,
                bufferRetainedSizeInBytes);
        bufferRetainedSizeInBytes = 0;
      }
      closed = true;
      executorService.submit(new SendCloseSinkChannelEventTask());
      currSequenceId = lastSequenceId + 1;
      sourceHandleListener.onFinished(this);
    }
  }
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