@Override
    public boolean updateTaskExecutionState(TaskExecutionStateTransition taskExecutionState) {
        return state.tryCall(
                        StateWithExecutionGraph.class,
                        stateWithExecutionGraph ->
                                stateWithExecutionGraph.updateTaskExecutionState(
                                        taskExecutionState, labelFailure(taskExecutionState)),
                        "updateTaskExecutionState")
                .orElse(false);
    }
@Test
    void testExceptionHistoryWithTaskFailureLabels() throws Exception {
        final Exception taskException = new Exception("Task Exception");
        BiConsumer<AdaptiveScheduler, List<ExecutionAttemptID>> testLogic =
                (scheduler, attemptIds) -> {
                    final ExecutionAttemptID attemptId = attemptIds.get(1);
                    scheduler.updateTaskExecutionState(
                            new TaskExecutionStateTransition(
                                    new TaskExecutionState(
                                            attemptId, ExecutionState.FAILED, taskException)));
                };

        final TestingFailureEnricher failureEnricher = new TestingFailureEnricher();
        final Iterable<RootExceptionHistoryEntry> actualExceptionHistory =
                new ExceptionHistoryTester(singleThreadMainThreadExecutor)
                        .withFailureEnrichers(Collections.singletonList(failureEnricher))
                        .withTestLogic(testLogic)
                        .run();

        assertThat(actualExceptionHistory).hasSize(1);

        final RootExceptionHistoryEntry failure = actualExceptionHistory.iterator().next();

        assertThat(failure.getException().deserializeError(classLoader)).isEqualTo(taskException);
        assertThat(failure.getFailureLabels()).isEqualTo(failureEnricher.getFailureLabels());
    }