private void handle(FailureBatch failureBatch) {
        suitableHandlers(failureBatch)
                .forEach(handler -> {
                    try {
                        handler.handle(failureBatch);
                    } catch (Exception e) {
                        logger.error("Error occurred while handling failures by {}", handler.getClass().getName());
                    }
                });

        final List<Message> requiresAcknowledgement = failureBatch.getFailures().stream()
                .filter(Failure::requiresAcknowledgement)
                .map(Failure::failedMessage)
                .filter(Message.class::isInstance)
                .map(Message.class::cast)
                .collect(Collectors.toList());

        if (!requiresAcknowledgement.isEmpty()) {
            acknowledger.acknowledge(requiresAcknowledgement);
        }
    }
@Test
    public void shutDown_uponShutdownAllRemainingFailuresAreHandled() throws Exception {
        // given
        final FailureBatch indexingFailureBatch = indexingFailureBatch(createIndexingFailure());

        final FailureHandler fallbackFailureHandler = enabledFailureHandler(indexingFailureBatch);

        final FailureSubmissionQueue failureSubmissionQueue = mock(FailureSubmissionQueue.class);
        final FailureHandlingService underTest = new FailureHandlingService(fallbackFailureHandler,
                ImmutableSet.of(), failureSubmissionQueue, configuration, acknowledger);

        when(configuration.getFailureHandlingShutdownAwait()).thenReturn(com.github.joschi.jadconfig.util.Duration.milliseconds(300));
        when(failureSubmissionQueue.consumeBlockingWithTimeout(300L))
                .thenReturn(indexingFailureBatch)
                .thenReturn(null);

        underTest.startAsync();
        underTest.awaitRunning();

        //when
        underTest.stopAsync();
        underTest.awaitTerminated();

        // then
        verify(failureSubmissionQueue, times(2)).consumeBlockingWithTimeout(300L);
        verify(fallbackFailureHandler, times(1)).handle(indexingFailureBatch);
    }