@Override
    public Messages process(Messages messages) {

        for (final MessageFilter filter : filterRegistry) {
            for (Message msg : messages) {
                final String timerName = name(filter.getClass(), "executionTime");
                final Timer timer = metricRegistry.timer(timerName);
                final Timer.Context timerContext = timer.time();

                try {
                    LOG.trace("Applying filter [{}] on message <{}>.", filter.getName(), msg.getId());

                    if (filter.filter(msg)) {
                        LOG.debug("Filter [{}] marked message <{}> to be discarded. Dropping message.",
                                filter.getName(),
                                msg.getId());
                        msg.setFilterOut(true);
                        filteredOutMessages.mark();
                        messageQueueAcknowledger.acknowledge(msg);
                    }
                } catch (Exception e) {
                    final String shortError = String.format(Locale.US, "Could not apply filter [%s] on message <%s>",
                            filter.getName(), msg.getId());
                    if (LOG.isDebugEnabled()) {
                        LOG.error("{}:", shortError, e);
                    } else {
                        LOG.error("{}:\n{}", shortError, ExceptionUtils.getShortenedStackTrace(e));
                    }
                    msg.addProcessingError(new Message.ProcessingError(ProcessingFailureCause.MessageFilterException,
                            shortError, ExceptionUtils.getRootCauseMessage(e)));
                } finally {
                    final long elapsedNanos = timerContext.stop();
                    msg.recordTiming(serverStatus, timerName, elapsedNanos);
                }
            }
        }
        return messages;
    }
@Test
    public void testHandleMessage() {

        MessageFilter filterOnlyFirst = new MessageFilter() {
            private boolean filterOut = true;

            @Override
            public boolean filter(Message msg) {
                if (filterOut) {
                    msg.setFilterOut(true);
                    filterOut = false;
                    return true;
                }
                return false;
            }

            @Override
            public String getName() {
                return "first filtered out, subsequent pass";
            }

            @Override
            public int getPriority() {
                return 0;
            }
        };

        final MessageFilterChainProcessor filterTest = new MessageFilterChainProcessor(new MetricRegistry(),
                                                                                       Collections.singleton(filterOnlyFirst),
                                                                                       acknowledger,
                                                                                       serverStatus);
        Message filteredoutMessage = messageFactory.createMessage("filtered out", "source", Tools.nowUTC());
        filteredoutMessage.setJournalOffset(1);
        Message unfilteredMessage = messageFactory.createMessage("filtered out", "source", Tools.nowUTC());

        final Messages messages1 = filterTest.process(filteredoutMessage);
        final Messages messages2 = filterTest.process(unfilteredMessage);

        Assert.assertTrue(filteredoutMessage.getFilterOut());
        Assert.assertFalse(unfilteredMessage.getFilterOut());
        Assert.assertEquals(0, Iterables.size(messages1));
        Assert.assertEquals(1, Iterables.size(messages2));
    }