public void dropPipe(DropPipeStmt stmt) throws DdlException {
        Pipe pipe = null;
        try {
            lock.writeLock().lock();
            Pair<Long, String> dbAndName = resolvePipeNameUnlock(stmt.getPipeName());
            PipeId pipeId = nameToId.get(dbAndName);
            if (pipeId == null) {
                if (stmt.isIfExists()) {
                    return;
                }
                ErrorReport.reportSemanticException(ErrorCode.ERR_UNKNOWN_PIPE, stmt.getPipeName());
            }
            pipe = pipeMap.get(nameToId.get(dbAndName));

            dropPipeImpl(pipe);
        } catch (Throwable e) {
            LOG.error("drop pipe {} failed", pipe, e);
            throw e;
        } finally {
            lock.writeLock().unlock();
        }
    }
@Test
    public void testExecuteFailed() throws Exception {
        TaskManager taskManager = GlobalStateMgr.getCurrentState().getTaskManager();
        mockRepoExecutor();

        // mock execution failed
        for (boolean retryAll : Lists.newArrayList(true, false)) {
            final String pipeName = "p3";
            Pipe p3 = preparePipe(pipeName);
            new mockit.Expectations(taskManager) {
                {
                    taskManager.executeTaskAsync((Task) any, (ExecuteOption) any);
                    SubmitResult submit = new SubmitResult("queryid", SubmitResult.SubmitStatus.SUBMITTED);
                    FutureTask<Constants.TaskRunState> future = new FutureTask<>(() -> Constants.TaskRunState.FAILED);
                    submit.setFuture(future);
                    future.run();
                    result = submit;
                }
            };
            Assert.assertEquals(0, p3.getRunningTasks().size());
            pipeRetryFailedTask(p3, retryAll);
            dropPipe(pipeName);
        }

        // mock execution cancelled
        for (boolean retryAll : Lists.newArrayList(true, false)) {
            final String pipeName = "p4";
            Pipe p4 = preparePipe(pipeName);
            new mockit.Expectations(taskManager) {
                {
                    taskManager.executeTaskAsync((Task) any, (ExecuteOption) any);
                    SubmitResult submit = new SubmitResult("queryid", SubmitResult.SubmitStatus.SUBMITTED);
                    FutureTask<Constants.TaskRunState> future = new FutureTask<>(() -> Constants.TaskRunState.FAILED);
                    submit.setFuture(future);
                    future.cancel(true);
                    result = submit;
                }
            };
            Assert.assertEquals(0, p4.getRunningTasks().size());
            pipeRetryFailedTask(p4, retryAll);
            dropPipe(pipeName);
        }
    }