public void createPipe(CreatePipeStmt stmt) throws DdlException {
        try {
            lock.writeLock().lock();
            Pair<Long, String> dbIdAndName = resolvePipeNameUnlock(stmt.getPipeName());
            boolean existed = nameToId.containsKey(dbIdAndName);
            if (existed) {
                if (!stmt.isIfNotExists() && !stmt.isReplace()) {
                    ErrorReport.reportSemanticException(ErrorCode.ERR_PIPE_EXISTS);
                }
                if (stmt.isIfNotExists()) {
                    return;
                } else if (stmt.isReplace()) {
                    LOG.info("Pipe {} already exist, replace it with a new one", stmt.getPipeName());
                    Pipe pipe = pipeMap.get(nameToId.get(dbIdAndName));
                    dropPipeImpl(pipe);
                }
            }

            // Add pipe
            long id = GlobalStateMgr.getCurrentState().getNextId();
            Pipe pipe = Pipe.fromStatement(id, stmt);
            putPipe(pipe);

            repo.addPipe(pipe);
        } finally {
            lock.writeLock().unlock();
        }

    }
@Test
    @Ignore("flaky test")
    public void testExecuteTaskSubmitFailed() throws Exception {
        mockRepoExecutor();
        final String pipeName = "p3";
        String sql = "create pipe p3 as insert into tbl1 select * from files('path'='fake://pipe', 'format'='parquet')";
        createPipe(sql);

        // poll error
        mockPollError(1);

        Pipe p3 = getPipe(pipeName);
        p3.poll();
        Assert.assertEquals(Pipe.State.ERROR, p3.getState());

        // clear the error and resume the pipe
        resumePipe(pipeName);
        p3.setLastPolledTime(0);
        Assert.assertEquals(Pipe.State.RUNNING, p3.getState());
        p3.poll();
        p3.schedule();
        Assert.assertEquals(Pipe.State.RUNNING, p3.getState());
        Assert.assertEquals(1, p3.getRunningTasks().size());

        TaskManager taskManager = GlobalStateMgr.getCurrentState().getTaskManager();
        new mockit.Expectations(taskManager) {
            {
                // submit error
                taskManager.executeTaskAsync((Task) any, (ExecuteOption) any);
                result = new SubmitResult("queryid", SubmitResult.SubmitStatus.FAILED);

            }
        };

        Thread.sleep(1000);
        Assert.assertEquals(1, p3.getRunningTasks().size());
        // retry several times, until failed
        for (int i = 0; i < Pipe.FAILED_TASK_THRESHOLD; i++) {
            p3.schedule();
            Assert.assertEquals(Pipe.State.RUNNING, p3.getState());
            Assert.assertEquals(1, p3.getRunningTasks().size());
            Assert.assertTrue(String.format("iteration %d: %s", i, p3.getRunningTasks()),
                    p3.getRunningTasks().stream().allMatch(PipeTaskDesc::isError));

            p3.schedule();
            Assert.assertEquals(Pipe.State.RUNNING, p3.getState());
            Assert.assertTrue(String.format("iteration %d: %s", i, p3.getRunningTasks()),
                    p3.getRunningTasks().stream().allMatch(PipeTaskDesc::isRunnable));
        }
        p3.schedule();
        Assert.assertEquals(Pipe.FAILED_TASK_THRESHOLD + 1, p3.getFailedTaskExecutionCount());
        Assert.assertEquals(Pipe.State.ERROR, p3.getState());

        // retry all
        {
            AlterPipeStmt alter = (AlterPipeStmt) UtFrameUtils.parseStmtWithNewParser("alter pipe p3 retry all", ctx);
            p3.retry((AlterPipeClauseRetry) alter.getAlterPipeClause());
            List<PipeFileRecord> unloadedFiles =
                    p3.getPipeSource().getFileListRepo().listFilesByState(FileListRepo.PipeFileState.UNLOADED, 0);
            Assert.assertEquals(1, unloadedFiles.size());
        }
    }