void onFatalError(final Throwable t) {
        try {
            log.error("Fatal error occurred in TaskExecutor {}.", getAddress(), t);
        } catch (Throwable ignored) {
        }

        // The fatal error handler implementation should make sure that this call is non-blocking
        fatalErrorHandler.onFatalError(t);
    }
@Test
    @Timeout(10)
    void testTerminationOnFatalError() throws Throwable {
        try (TaskSubmissionTestEnvironment env =
                new Builder(jobId)
                        .setConfiguration(configuration)
                        .build(EXECUTOR_EXTENSION.getExecutor())) {
            String testExceptionMsg = "Test exception of fatal error.";

            env.getTaskExecutor().onFatalError(new Exception(testExceptionMsg));

            Throwable exception = env.getTestingFatalErrorHandler().getErrorFuture().get();
            env.getTestingFatalErrorHandler().clearError();

            assertThat(exception.getMessage()).startsWith(testExceptionMsg);
        }
    }