public void submitIndexingErrors(Collection<IndexingError> indexingErrors) {
        try {
            final FailureBatch fb = FailureBatch.indexingFailureBatch(
                    indexingErrors.stream()
                            .filter(ie -> {
                                if (!ie.message().supportsFailureHandling()) {
                                    logger.warn("Submitted a message with indexing errors, which doesn't support failure handling!");
                                    return false;
                                } else {
                                    return true;
                                }
                            })
                            .map(this::fromIndexingError)
                            .collect(Collectors.toList()));

            if (fb.size() > 0) {
                failureSubmissionQueue.submitBlocking(fb);
            }

        } catch (InterruptedException ignored) {
            logger.warn("Failed to submit {} indexing errors for failure handling. The thread has been interrupted!",
                    indexingErrors.size());
            Thread.currentThread().interrupt();
        }
    }
@Test
    public void submitIndexingErrors_allIndexingErrorsTransformedAndSubmittedToFailureQueue() throws Exception {
        // given
        final Message msg1 = Mockito.mock(Message.class);
        when(msg1.getMessageId()).thenReturn("msg-1");
        when(msg1.supportsFailureHandling()).thenReturn(true);
        final Message msg2 = Mockito.mock(Message.class);
        when(msg2.getMessageId()).thenReturn("msg-2");
        when(msg2.supportsFailureHandling()).thenReturn(true);

        final List<IndexingError> indexingErrors = List.of(
                IndexingError.create(msg1, "index-1", MappingError, "Error"),
                IndexingError.create(msg2, "index-2", Unknown, "Error2")
        );

        // when
        underTest.submitIndexingErrors(indexingErrors);

        // then
        verify(failureSubmissionQueue, times(1)).submitBlocking(failureBatchCaptor.capture());

        assertThat(failureBatchCaptor.getValue()).satisfies(fb -> {
            assertThat(fb.containsIndexingFailures()).isTrue();
            assertThat(fb.size()).isEqualTo(2);

            assertThat(fb.getFailures().get(0)).satisfies(indexingFailure -> {
                assertThat(indexingFailure.failureType()).isEqualTo(FailureType.INDEXING);
                assertThat(indexingFailure.failureCause().label()).isEqualTo("MappingError");
                assertThat(indexingFailure.message()).isEqualTo("Failed to index message with id 'msg-1' targeting 'index-1'");
                assertThat(indexingFailure.failureDetails()).isEqualTo("Error");
                assertThat(indexingFailure.failureTimestamp()).isNotNull();
                assertThat(indexingFailure.failedMessage()).isEqualTo(msg1);
                assertThat(indexingFailure.targetIndex()).isEqualTo("index-1");
                assertThat(indexingFailure.requiresAcknowledgement()).isFalse();
            });

            assertThat(fb.getFailures().get(1)).satisfies(indexingFailure -> {
                assertThat(indexingFailure.failureType()).isEqualTo(FailureType.INDEXING);
                assertThat(indexingFailure.failureCause().label()).isEqualTo("UNKNOWN");
                assertThat(indexingFailure.message()).isEqualTo("Failed to index message with id 'msg-2' targeting 'index-2'");
                assertThat(indexingFailure.failureDetails()).isEqualTo("Error2");
                assertThat(indexingFailure.failureTimestamp()).isNotNull();
                assertThat(indexingFailure.failedMessage()).isEqualTo(msg2);
                assertThat(indexingFailure.targetIndex()).isEqualTo("index-2");
                assertThat(indexingFailure.requiresAcknowledgement()).isFalse();
            });
        });
    }