docs: public void create(final Pipeline pipeline) {
          Assert.assertNotNull(pipeline);
          transactionTemplate.execute(new TransactionCallbackWithoutResult() {
  
              @Override
              protected void doInTransactionWithoutResult(TransactionStatus status) {
  
                  try {
                      PipelineDO pipelineDo = modelToDo(pipeline);
                      pipelineDo.setId(0L);
  
                      if (!pipelineDao.checkUnique(pipelineDo)) {
                          String exceptionCause = "exist the same name pipeline under the channel("
                                                  + pipelineDo.getChannelId() + ") in the database.";
                          logger.warn("WARN ## " + exceptionCause);
                          throw new RepeatConfigureException(exceptionCause);
                      }
                      pipelineDao.insert(pipelineDo);
  
                      List<PipelineNodeRelationDO> pipelineNodeRelationDos = new ArrayList<PipelineNodeRelationDO>();
  
                      for (Node node : pipeline.getSelectNodes()) {
                          PipelineNodeRelationDO pipelineNodeRelationDo = new PipelineNodeRelationDO();
                          pipelineNodeRelationDo.setPipelineId(pipelineDo.getId());
                          pipelineNodeRelationDo.setNodeId(node.getId());
                          pipelineNodeRelationDo.setLocation(Location.SELECT);
                          pipelineNodeRelationDos.add(pipelineNodeRelationDo);
                      }
  
                      for (Node node : pipeline.getExtractNodes()) {
                          PipelineNodeRelationDO pipelineNodeRelationDo = new PipelineNodeRelationDO();
                          pipelineNodeRelationDo.setPipelineId(pipelineDo.getId());
                          pipelineNodeRelationDo.setNodeId(node.getId());
                          pipelineNodeRelationDo.setLocation(Location.EXTRACT);
                          pipelineNodeRelationDos.add(pipelineNodeRelationDo);
                      }
  
                      for (Node node : pipeline.getLoadNodes()) {
                          PipelineNodeRelationDO pipelineNodeRelationDo = new PipelineNodeRelationDO();
                          pipelineNodeRelationDo.setPipelineId(pipelineDo.getId());
                          pipelineNodeRelationDo.setNodeId(node.getId());
                          pipelineNodeRelationDo.setLocation(Location.LOAD);
                          pipelineNodeRelationDos.add(pipelineNodeRelationDo);
                      }
  
                      pipelineNodeRelationDao.insertBatch(pipelineNodeRelationDos);
                      arbitrateManageService.pipelineEvent().init(pipelineDo.getChannelId(), pipelineDo.getId());
                  } catch (RepeatConfigureException rce) {
                      throw rce;
                  } catch (Exception e) {
                      logger.error("ERROR ## create pipeline has an exception!");
                      throw new ManagerException(e);
                  }
              }
          });
  
      }