@Override
    public boolean updateTaskExecutionState(TaskExecutionStateTransition taskExecutionState) {
        return state.tryCall(
                        StateWithExecutionGraph.class,
                        stateWithExecutionGraph ->
                                stateWithExecutionGraph.updateTaskExecutionState(
                                        taskExecutionState, labelFailure(taskExecutionState)),
                        "updateTaskExecutionState")
                .orElse(false);
    }
@Test
    void testExceptionHistoryWithTaskConcurrentFailure() throws Exception {
        final Exception expectedException1 = new Exception("Expected Local Exception 1");
        final Exception expectedException2 = new Exception("Expected Local Exception 2");
        BiConsumer<AdaptiveScheduler, List<ExecutionAttemptID>> testLogic =
                (scheduler, attemptIds) -> {
                    final ExecutionAttemptID attemptId = attemptIds.remove(0);
                    final ExecutionAttemptID attemptId2 = attemptIds.remove(0);

                    scheduler.updateTaskExecutionState(
                            new TaskExecutionStateTransition(
                                    new TaskExecutionState(
                                            attemptId, ExecutionState.FAILED, expectedException1)));
                    scheduler.updateTaskExecutionState(
                            new TaskExecutionStateTransition(
                                    new TaskExecutionState(
                                            attemptId2,
                                            ExecutionState.FAILED,
                                            expectedException2)));
                };

        final Iterable<RootExceptionHistoryEntry> entries =
                new ExceptionHistoryTester(singleThreadMainThreadExecutor)
                        .withTestLogic(testLogic)
                        .run();
        assertThat(entries).hasSize(1);
        final RootExceptionHistoryEntry failure = entries.iterator().next();
        assertThat(failure.getException().deserializeError(classLoader))
                .isEqualTo(expectedException1);
        final Iterable<ExceptionHistoryEntry> concurrentExceptions =
                failure.getConcurrentExceptions();
        final List<Throwable> foundExceptions =
                IterableUtils.toStream(concurrentExceptions)
                        .map(ExceptionHistoryEntry::getException)
                        .map(exception -> exception.deserializeError(classLoader))
                        .collect(Collectors.toList());

        // In the future, concurrent local failures should be stored.
        assertThat(foundExceptions).isEmpty();
    }