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 run_whenCustomHandlersProvided_thenFallbackHandlerIgnored() throws Exception {
        // given
        final FailureBatch indexingFailureBatch = indexingFailureBatch(createIndexingFailure());

        final FailureHandler customIndexingFailureHandler1 = enabledFailureHandler(indexingFailureBatch);
        final FailureHandler customIndexingFailureHandler2 = enabledFailureHandler(indexingFailureBatch);
        final FailureHandler fallbackIndexingFailureHandler = enabledFailureHandler(indexingFailureBatch);

        final FailureHandlingService underTest = new FailureHandlingService(fallbackIndexingFailureHandler,
                ImmutableSet.of(customIndexingFailureHandler1, customIndexingFailureHandler2), failureSubmissionQueue, configuration, acknowledger);

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

        // when
        failureSubmissionQueue.submitBlocking(indexingFailureBatch);

        Awaitility.waitAtMost(Durations.ONE_SECOND)
                .until(() -> failureSubmissionQueue.queueSize() == 0);

        // then
        verify(customIndexingFailureHandler1).handle(indexingFailureBatch);
        verify(customIndexingFailureHandler2).handle(indexingFailureBatch);
        verify(fallbackIndexingFailureHandler, times(0)).handle(any());
    }