public void discardEventsOfPipe(final String pipeNameToDrop, int regionId) {
    // Try to remove the events as much as possible
    inputPendingQueue.discardEventsOfPipe(pipeNameToDrop, regionId);

    highPriorityLockTaskCount.incrementAndGet();
    try {
      synchronized (highPriorityLockTaskCount) {
        highPriorityLockTaskCount.notifyAll();
      }

      // synchronized to use the lastEvent & lastExceptionEvent
      synchronized (this) {
        // Here we discard the last event, and re-submit the pipe task to avoid that the pipe task
        // has stopped submission but will not be stopped by critical exceptions, because when it
        // acquires lock, the pipe is already dropped, thus it will do nothing. Note that since we
        // use a new thread to stop all the pipes, we will not encounter deadlock here. Or else we
        // will.
        if (lastEvent instanceof EnrichedEvent
            && pipeNameToDrop.equals(((EnrichedEvent) lastEvent).getPipeName())
            && regionId == ((EnrichedEvent) lastEvent).getRegionId()) {
          // Do not clear the last event's reference counts because it may be on transferring
          lastEvent = null;
          // Submit self to avoid that the lastEvent has been retried "max times" times and has
          // stopped executing.
          // 1. If the last event is still on execution, or submitted by the previous "onSuccess" or
          //    "onFailure", the "submitSelf" causes nothing.
          // 2. If the last event is waiting the instance lock to call "onSuccess", then the
          //    callback method will skip this turn of submission.
          // 3. If the last event is waiting to call "onFailure", then it will be ignored because
          //    the last event has been set to null.
          // 4. If the last event has called "onFailure" and caused the subtask to stop submission,
          //    it's submitted here and the "report" will wait for the "drop pipe" lock to stop all
          //    the pipes with critical exceptions. As illustrated above, the "report" will do
          //    nothing.
          submitSelf();
        }

        // We only clear the lastEvent's reference counts when it's already on failure. Namely, we
        // clear the lastExceptionEvent. It's safe to potentially clear it twice because we have the
        // "nonnull" detection.
        if (lastExceptionEvent instanceof EnrichedEvent
            && pipeNameToDrop.equals(((EnrichedEvent) lastExceptionEvent).getPipeName())
            && regionId == ((EnrichedEvent) lastExceptionEvent).getRegionId()) {
          clearReferenceCountAndReleaseLastExceptionEvent();
        }
      }
    } finally {
      highPriorityLockTaskCount.decrementAndGet();
    }

    if (outputPipeConnector instanceof IoTDBConnector) {
      ((IoTDBConnector) outputPipeConnector).discardEventsOfPipe(pipeNameToDrop, regionId);
    }
  }
When a pipe is dropped, the connector maybe reused and will not be closed. So we just discard
its queued events in the output pipe connector.
iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/task/subtask/connector/PipeConnectorSubtask.java