@Override
    public void writeFiltered(List<FilteredMessage> filteredMessages) throws Exception {
        final var messages = filteredMessages.stream()
                .filter(message -> !message.destinations().get(FILTER_KEY).isEmpty())
                .toList();

        writes.mark(messages.size());
        ignores.mark(filteredMessages.size() - messages.size());

        writeMessageEntries(messages);
    }
@Test
    public void writeFiltered() throws Exception {
        final List<Message> messageList = buildMessages(2);

        output.writeFiltered(List.of(
                // The first message should not be written to the output because the output's filter key is not included.
                DefaultFilteredMessage.forDestinationKeys(messageList.get(0), Set.of("foo")),
                DefaultFilteredMessage.forDestinationKeys(messageList.get(1), Set.of("foo", ElasticSearchOutput.FILTER_KEY))
        ));

        verify(messages, times(1)).bulkIndex(eq(List.of(
                new MessageWithIndex(wrap(messageList.get(1)), defaultIndexSet)
        )));

        verifyNoMoreInteractions(messages);
    }