@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 testMessagesRecordProcessingFailures() {
        final MessageFilter first = new ExceptingMessageFilter();
        final Set<MessageFilter> filters = ImmutableSet.of(first);
        final MessageFilterChainProcessor processor = new MessageFilterChainProcessor(new MetricRegistry(),
                filters,
                acknowledger,
                serverStatus);

        final Message message = messageFactory.createMessage("message", "source", new DateTime(2016, 1, 1, 0, 0, DateTimeZone.UTC));
        final Messages result = processor.process(message);

        assertThat(result).hasSize(1);
        // passed message is mutated, so we can assert on that
        assertThat(message.processingErrors()).hasSize(1);
        assertThat(message.processingErrors().get(0)).satisfies(pe -> {
            assertThat(pe.getCause()).isEqualTo(ProcessingFailureCause.MessageFilterException);
            assertThat(pe.getMessage()).startsWith("Could not apply filter [Excepting filter] on message <");
            assertThat(pe.getDetails()).isEqualTo("BOOM!");
        });
    }