private boolean onPipeConnectionException(final Throwable throwable) {
    LOGGER.warn(
        "PipeConnectionException occurred, {} retries to handshake with the target system.",
        outputPipeConnector.getClass().getName(),
        throwable);

    int retry = 0;
    while (retry < MAX_RETRY_TIMES) {
      try {
        outputPipeConnector.handshake();
        LOGGER.info(
            "{} handshakes with the target system successfully.",
            outputPipeConnector.getClass().getName());
        break;
      } catch (final Exception e) {
        retry++;
        LOGGER.warn(
            "{} failed to handshake with the target system for {} times, "
                + "will retry at most {} times.",
            outputPipeConnector.getClass().getName(),
            retry,
            MAX_RETRY_TIMES,
            e);
        try {
          synchronized (highPriorityLockTaskCount) {
            // The wait operation will release the highPriorityLockTaskCount lock, so there will be
            // no deadlock.
            if (highPriorityLockTaskCount.get() == 0) {
              highPriorityLockTaskCount.wait(
                  retry * PipeConfig.getInstance().getPipeConnectorRetryIntervalMs());
            }
          }
        } catch (final InterruptedException interruptedException) {
          LOGGER.info(
              "Interrupted while sleeping, will retry to handshake with the target system.",
              interruptedException);
          Thread.currentThread().interrupt();
        }
      }
    }

    // Stop current pipe task directly if failed to reconnect to
    // the target system after MAX_RETRY_TIMES times
    if (retry == MAX_RETRY_TIMES && lastEvent instanceof EnrichedEvent) {
      report(
          (EnrichedEvent) lastEvent,
          new PipeRuntimeConnectorCriticalException(
              throwable.getMessage() + ", root cause: " + getRootCause(throwable)));
      LOGGER.warn(
          "{} failed to handshake with the target system after {} times, "
              + "stopping current subtask {} (creation time: {}, simple class: {}). "
              + "Status shown when query the pipe will be 'STOPPED'. "
              + "Please restart the task by executing 'START PIPE' manually if needed.",
          outputPipeConnector.getClass().getName(),
          MAX_RETRY_TIMES,
          taskID,
          creationTime,
          this.getClass().getSimpleName(),
          throwable);

      // Although the pipe task will be stopped, we still don't release the last event here
      // Because we need to keep it for the next retry. If user wants to restart the task,
      // the last event will be processed again. The last event will be released when the task
      // is dropped or the process is running normally.

      // Stop current pipe task if failed to reconnect to the target system after MAX_RETRY_TIMES
      return true;
    }

    // For non enriched event, forever retry.
    // For enriched event, retry if connection is set up successfully.
    return false;
  }
@return {@code true} if the {@link PipeSubtask} should be stopped, {@code false} otherwise
iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/agent/task/subtask/PipeAbstractConnectorSubtask.java