private boolean recordDataNodePushPipeMetaExceptionsInternal(
      final Map<Integer, TPushPipeMetaResp> respMap) {
    boolean hasException = false;

    for (final Map.Entry<Integer, TPushPipeMetaResp> respEntry : respMap.entrySet()) {
      final int dataNodeId = respEntry.getKey();
      final TPushPipeMetaResp resp = respEntry.getValue();

      if (resp.getStatus().getCode() == TSStatusCode.PIPE_PUSH_META_ERROR.getStatusCode()) {
        hasException = true;

        if (!resp.isSetExceptionMessages()) {
          // The pushPipeMeta process on dataNode encountered internal errors
          continue;
        }

        resp.getExceptionMessages().stream()
            .filter(message -> pipeMetaKeeper.containsPipeMeta(message.getPipeName()))
            .forEach(
                message -> {
                  final PipeRuntimeMeta runtimeMeta =
                      pipeMetaKeeper.getPipeMeta(message.getPipeName()).getRuntimeMeta();

                  // Mark the status of the pipe with exception as stopped
                  runtimeMeta.getStatus().set(PipeStatus.STOPPED);
                  runtimeMeta.onSetPipeDroppedOrStopped();
                  runtimeMeta.setIsStoppedByRuntimeException(true);

                  final Map<Integer, PipeRuntimeException> exceptionMap =
                      runtimeMeta.getNodeId2PipeRuntimeExceptionMap();
                  if (!exceptionMap.containsKey(dataNodeId)
                      || exceptionMap.get(dataNodeId).getTimeStamp() < message.getTimeStamp()) {
                    exceptionMap.put(
                        dataNodeId,
                        new PipeRuntimeCriticalException(
                            message.getMessage(), message.getTimeStamp()));
                  }
                });
      }
    }

    return hasException;
  }
Record the exceptions of all pipes locally if they encountered failure when pushing {@link
PipeMeta}s to dataNodes.

<p>If there are exceptions recorded, the related pipes will be stopped, and the exception
messages will then be updated to all the nodes through {@link PipeHandleMetaChangeProcedure}.

@param respMap The responseMap after pushing pipe meta
@return {@code true} if there are exceptions encountered
iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/persistence/pipe/PipeTaskInfo.java