@Override
    public void handleAsyncException(String message, Throwable exception) {
        if (isRestoring || isRunning) {
            // only fail if the task is still in restoring or running
            asyncExceptionHandler.handleAsyncException(message, exception);
        }
    }
@Test
    void testAsyncExceptionHandlerHandleExceptionForwardsMessageProperly() {
        MockEnvironment mockEnvironment = MockEnvironment.builder().build();
        RuntimeException expectedException = new RuntimeException("RUNTIME EXCEPTION");

        final StreamTask.StreamTaskAsyncExceptionHandler asyncExceptionHandler =
                new StreamTask.StreamTaskAsyncExceptionHandler(mockEnvironment);

        mockEnvironment.setExpectedExternalFailureCause(AsynchronousException.class);
        final String expectedErrorMessage = "EXPECTED_ERROR MESSAGE";

        asyncExceptionHandler.handleAsyncException(expectedErrorMessage, expectedException);

        // expect an AsynchronousException containing the supplied error details
        Optional<? extends Throwable> actualExternalFailureCause =
                mockEnvironment.getActualExternalFailureCause();
        final Throwable actualException =
                actualExternalFailureCause.orElseThrow(
                        () -> new AssertionError("Expected exceptional completion"));

        assertThat(actualException)
                .isInstanceOf(AsynchronousException.class)
                .hasMessage(expectedErrorMessage)
                .hasCause(expectedException);
    }