@Override
  protected void rollbackState(ConfigNodeProcedureEnv env, OperatePipeTaskState state)
      throws IOException, InterruptedException, ProcedureException {
    if (pipeTaskInfo == null) {
      LOGGER.warn(
          "ProcedureId {}: Pipe lock is not acquired, rollbackState({})'s execution will be skipped.",
          getProcId(),
          state);
      return;
    }

    switch (state) {
      case VALIDATE_TASK:
        if (!isRollbackFromValidateTaskSuccessful) {
          try {
            rollbackFromValidateTask(env);
            isRollbackFromValidateTaskSuccessful = true;
          } catch (Exception e) {
            LOGGER.warn("ProcedureId {}: Failed to rollback from validate task.", getProcId(), e);
          }
        }
        break;
      case CALCULATE_INFO_FOR_TASK:
        try {
          rollbackFromCalculateInfoForTask(env);
        } catch (Exception e) {
          LOGGER.warn(
              "ProcedureId {}: Failed to rollback from calculate info for task.", getProcId(), e);
        }
        break;
      case WRITE_CONFIG_NODE_CONSENSUS:
        try {
          // rollbackFromWriteConfigNodeConsensus can be called before
          // rollbackFromOperateOnDataNodes.
          // So we need to check if rollbackFromOperateOnDataNodes is successfully executed.
          // If yes, we don't need to call rollbackFromWriteConfigNodeConsensus again.
          if (!isRollbackFromOperateOnDataNodesSuccessful) {
            rollbackFromWriteConfigNodeConsensus(env);
          }
        } catch (Exception e) {
          LOGGER.warn(
              "ProcedureId {}: Failed to rollback from write config node consensus.",
              getProcId(),
              e);
        }
        break;
      case OPERATE_ON_DATA_NODES:
        try {
          // We have to make sure that rollbackFromOperateOnDataNodes is executed after
          // rollbackFromWriteConfigNodeConsensus, because rollbackFromOperateOnDataNodes is
          // executed based on the consensus of config nodes that is written by
          // rollbackFromWriteConfigNodeConsensus
          rollbackFromWriteConfigNodeConsensus(env);
          rollbackFromOperateOnDataNodes(env);
          isRollbackFromOperateOnDataNodesSuccessful = true;
        } catch (Exception e) {
          LOGGER.warn(
              "ProcedureId {}: Failed to rollback from operate on data nodes.", getProcId(), e);
        }
        break;
      default:
        LOGGER.error("Unsupported roll back STATE [{}]", state);
    }
  }
Execute at state {@link OperatePipeTaskState#OPERATE_ON_DATA_NODES}.

@throws PipeException if push pipe metas to dataNodes failed
@throws IOException Exception when Serializing to byte buffer
iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/pipe/AbstractOperatePipeProcedureV2.java