@Override
    public void acknowledge(List<? extends Acknowledgeable> messages) {
        @SuppressWarnings("ConstantConditions")
        final Optional<Long> max =
                messages.stream()
                        .map(Acknowledgeable::getMessageQueueId)
                        .filter(this::isValidMessageQueueId)
                        .map(Long.class::cast)
                        .max(Long::compare);
        max.ifPresent(this::doAcknowledge);
        metrics.acknowledgedMessages().mark(messages.size());
    }
@Test
    void acknowledgeMessages(MessageFactory messageFactory) {
        final Message firstMessage = messageFactory.createMessage("message", "source", DateTime.now(UTC));
        firstMessage.setMessageQueueId(1L);

        final Message nullOffsetMessage = messageFactory.createMessage("message", "source", DateTime.now(UTC));

        final Message secondMessage = messageFactory.createMessage("message", "source", DateTime.now(UTC));
        secondMessage.setMessageQueueId(2L);

        final Message wrongOffsetTypeMessage = messageFactory.createMessage("message", "source", DateTime.now(UTC));
        wrongOffsetTypeMessage.setMessageQueueId("foo");

        acknowledger.acknowledge(ImmutableList.of(firstMessage, nullOffsetMessage, secondMessage, wrongOffsetTypeMessage));

        verify(kafkaJournal).markJournalOffsetCommitted(2L);
    }