@Override
    public void onFinished(ISourceHandle sourceHandle) {
      if (LOGGER.isDebugEnabled()) {
        LOGGER.debug("[ScHListenerOnFinish]");
      }
      Map<String, ISourceHandle> sourceHandleMap =
          sourceHandles.get(sourceHandle.getLocalFragmentInstanceId());
      if ((sourceHandleMap == null
              || sourceHandleMap.remove(sourceHandle.getLocalPlanNodeId()) == null)
          && LOGGER.isDebugEnabled()) {
        LOGGER.debug("[ScHListenerAlreadyReleased]");
      }

      if (sourceHandleMap != null && sourceHandleMap.isEmpty()) {
        sourceHandles.remove(sourceHandle.getLocalFragmentInstanceId());
      }
    }
Listen to the state changes of a source handle.
iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/exchange/MPPDataExchangeManager.java