protected List<TPushPipeMetaRespExceptionMessage> handlePipeMetaChangesInternal(
      final List<PipeMeta> pipeMetaListFromCoordinator) {
    // Do nothing if the node is removing or removed
    if (isShutdown()) {
      return Collections.emptyList();
    }

    final List<TPushPipeMetaRespExceptionMessage> exceptionMessages = new ArrayList<>();
    final List<PipeMeta> copyPipeMetaListFromCoordinator =
        new LinkedList<>(pipeMetaListFromCoordinator);

    while (!copyPipeMetaListFromCoordinator.isEmpty()) {
      // Iterate through pipe meta list from coordinator, check if pipe meta exists on local agent
      // or has changed
      exceptionMessages.clear();
      Iterator<PipeMeta> pipeMetas = copyPipeMetaListFromCoordinator.iterator();
      int successfulPipeCount = 0;
      while (pipeMetas.hasNext()) {
        PipeMeta metaFromCoordinator = pipeMetas.next();
        try {
          executeSinglePipeMetaChanges(metaFromCoordinator);
          pipeMetas.remove();
          successfulPipeCount++;
        } catch (final Exception | Error e) {
          // To ensure that all Pipes cannot be started due to an Error in a certain pipeline, you
          // need to capture the Error. The reason for the Error may be a class loading failure. For
          // example, plugin A depends on plugin B, but because plugin B is not loaded, plugin A
          // cannot be loaded, so we need to ignore the Error of the loading process.
          final String pipeName = metaFromCoordinator.getStaticMeta().getPipeName();
          final String errorMessage =
              String.format(
                  "Failed to handle pipe meta changes for %s, because %s",
                  pipeName, e.getMessage());
          LOGGER.warn("Failed to handle pipe meta changes for {}", pipeName, e);
          exceptionMessages.add(
              new TPushPipeMetaRespExceptionMessage(
                  pipeName, errorMessage, System.currentTimeMillis()));
        }
      }
      // If the number of successful changes to pipe meta is 0, it means that the failure has
      // nothing to do with the loading order, so we can exit.
      if (successfulPipeCount == 0) {
        break;
      }
    }

    // Check if there are pipes on local agent that do not exist on coordinator, if so, drop them
    final Set<String> pipeNamesFromCoordinator =
        pipeMetaListFromCoordinator.stream()
            .map(meta -> meta.getStaticMeta().getPipeName())
            .collect(Collectors.toSet());
    for (final PipeMeta metaInAgent : pipeMetaKeeper.getPipeMetaList()) {
      final String pipeName = metaInAgent.getStaticMeta().getPipeName();

      try {
        if (!pipeNamesFromCoordinator.contains(pipeName)) {
          dropPipe(metaInAgent.getStaticMeta().getPipeName());
        }
      } catch (final Exception e) {
        // Report the exception message for CN to sense the failure of meta sync
        final String errorMessage =
            String.format(
                "Failed to handle pipe meta changes for %s, because %s", pipeName, e.getMessage());
        LOGGER.warn("Failed to handle pipe meta changes for {}", pipeName, e);
        exceptionMessages.add(
            new TPushPipeMetaRespExceptionMessage(
                pipeName, errorMessage, System.currentTimeMillis()));
      }
    }

    return exceptionMessages;
  }
State transition diagram of a pipe task:

<p><code>
|----------------|                     |---------| --> stop  pipe --> |---------|                   |---------|
| initial status | --> create pipe --> | RUNNING |                    | STOPPED | --> drop pipe --> | DROPPED |
|----------------|                     |---------| <-- start pipe <-- |---------|                   |---------|
                                            |                                                            |
                                            | ----------------------> drop pipe -----------------------> |
</code>

<p>Other transitions are not allowed, will be ignored when received in the {@link PipeTaskAgent}.
iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/agent/task/PipeTaskAgent.java