@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 acknowledgeMessageWithWrongTypeOfMessageQueueId(MessageFactory messageFactory) {
        final Message message = messageFactory.createMessage("message", "source", DateTime.now(UTC));
        message.setMessageQueueId("foo");
        acknowledger.acknowledge(message);
        verifyNoMoreInteractions(kafkaJournal);
    }