@Deprecated
    public PassiveCompletableFuture<TaskExecutionState> deployLocalTask(
            @NonNull TaskGroup taskGroup) {
        return deployLocalTask(
                taskGroup, Thread.currentThread().getContextClassLoader(), emptyList());
    }
@Test
    public void testThrowException() throws InterruptedException {
        TaskExecutionService taskExecutionService = server.getTaskExecutionService();

        AtomicBoolean stopMark = new AtomicBoolean(false);

        long t1Sleep = 100;
        long t2Sleep = 50;

        long lowLagSleep = 50;
        long highLagSleep = 300;

        List<Throwable> t1throwable = new ArrayList<>();
        ExceptionTestTask t1 = new ExceptionTestTask(t1Sleep, "t1", t1throwable);

        List<Throwable> t2throwable = new ArrayList<>();
        ExceptionTestTask t2 = new ExceptionTestTask(t2Sleep, "t2", t2throwable);

        // Create low lat tasks
        List<Task> lowLagTask =
                buildFixedTestTask(lowLagSleep, 10, stopMark, new CopyOnWriteArrayList<>());

        // Create high lat tasks
        List<Task> highLagTask =
                buildFixedTestTask(highLagSleep, 5, stopMark, new CopyOnWriteArrayList<>());

        List<Task> tasks = new ArrayList<>();
        tasks.addAll(highLagTask);
        tasks.addAll(lowLagTask);
        Collections.shuffle(tasks);

        CompletableFuture<TaskExecutionState> taskCts =
                taskExecutionService.deployLocalTask(
                        new TaskGroupDefaultImpl(
                                new TaskGroupLocation(
                                        jobId, pipeLineId, FLAKE_ID_GENERATOR.newId()),
                                "ts",
                                Lists.newArrayList(tasks)));

        CompletableFuture<TaskExecutionState> t1c =
                taskExecutionService.deployLocalTask(
                        new TaskGroupDefaultImpl(
                                new TaskGroupLocation(
                                        jobId, pipeLineId, FLAKE_ID_GENERATOR.newId()),
                                "t1",
                                Lists.newArrayList(t1)));

        CompletableFuture<TaskExecutionState> t2c =
                taskExecutionService.deployLocalTask(
                        new TaskGroupDefaultImpl(
                                new TaskGroupLocation(
                                        jobId, pipeLineId, FLAKE_ID_GENERATOR.newId()),
                                "t2",
                                Lists.newArrayList(t2)));

        Thread.sleep(taskRunTime);

        t1throwable.add(new IOException());
        t2throwable.add(new IOException());

        await().atMost(t1Sleep + t2Sleep + 1000, TimeUnit.MILLISECONDS)
                .untilAsserted(
                        () -> {
                            assertEquals(FAILED, t1c.get().getExecutionState());
                            assertEquals(FAILED, t2c.get().getExecutionState());
                        });

        stopMark.set(true);

        await().atMost(lowLagSleep * 10 + highLagSleep + 1000, TimeUnit.MILLISECONDS)
                .untilAsserted(() -> assertEquals(FINISHED, taskCts.get().getExecutionState()));
    }