@WithSpan
    @Override
    public SearchResponse apply(SearchResponse searchResponse) {

        final LookupTable table = lookupTable.getTable();

        if (table != null) {
            Span.current().setAttribute(LOOKUP_TABLE_NAME, table.name())
                    .setAttribute(LOOKUP_CACHE_NAME, table.cache().name())
                    .setAttribute(LOOKUP_CACHE_TYPE, table.cache().getConfig().type())
                    .setAttribute(LOOKUP_DATA_ADAPTER_NAME, table.dataAdapter().name())
                    .setAttribute(LOOKUP_DATA_ADAPTER_TYPE, table.dataAdapter().getConfig().type());
        }

        final List<ResultMessageSummary> summaries = searchResponse.messages().stream()
                .map(summary -> {
                    // Do not touch the message if the field does not exist.
                    if (!summary.message().containsKey(sourceField)) {
                        return summary;
                    }

                    final LookupResult result = lookupTable.lookup(summary.message().get(sourceField));

                    // Do not touch the message if there is no result
                    if (result == null || result.isEmpty()) {
                        return summary;
                    }

                    final Message message = messageFactory.createMessage(ImmutableMap.copyOf(summary.message()));

                    message.addField(targetField, result.singleValue());

                    return summary.toBuilder().message(message.getFields()).build();
                })
                .collect(Collectors.toList());

        return searchResponse.toBuilder().messages(summaries).build();
    }
@Test
    public void decorate() throws Exception {
        final String sourceField = "source";
        final String targetField = "source_decorated";
        final String lookupTableName = "test";

        final Decorator decorator = createDecorator(sourceField, targetField, lookupTableName);
        final Pair<LookupTableDecorator, LookupTableService.Function> lookupTableDecoratorPair = createLookupTableDecorator(decorator);

        final LookupTableDecorator lookupTableDecorator = lookupTableDecoratorPair.getLeft();
        final LookupTableService.Function function = lookupTableDecoratorPair.getRight();

        final List<ResultMessageSummary> messages = ImmutableList.of(
                ResultMessageSummary.create(ImmutableMultimap.of(), ImmutableMap.of("_id", "a", sourceField, "0"), "graylog_0"),
                ResultMessageSummary.create(ImmutableMultimap.of(), ImmutableMap.of("_id", "b", sourceField, "1"), "graylog_0"),
                ResultMessageSummary.create(ImmutableMultimap.of(), ImmutableMap.of("_id", "c", sourceField, "2"), "graylog_0"),
                ResultMessageSummary.create(ImmutableMultimap.of(), ImmutableMap.of("_id", "d", sourceField, "3"), "graylog_0"),
                ResultMessageSummary.create(ImmutableMultimap.of(), ImmutableMap.of("_id", "e", "invalid", "4"), "graylog_0")
        );

        final SearchResponse searchResponse = createSearchResponse(messages);

        when(function.lookup("0")).thenReturn(LookupResult.single("zero"));
        when(function.lookup("1")).thenReturn(LookupResult.single("one"));
        when(function.lookup("2")).thenReturn(LookupResult.empty());
        when(function.lookup("3")).thenReturn(null);

        final SearchResponse response = lookupTableDecorator.apply(searchResponse);

        assertThat(response.messages().get(0).message().get(sourceField)).isEqualTo("0");
        assertThat(response.messages().get(0).message().get(targetField)).isEqualTo("zero");

        assertThat(response.messages().get(1).message().get(sourceField)).isEqualTo("1");
        assertThat(response.messages().get(1).message().get(targetField)).isEqualTo("one");

        assertThat(response.messages().get(2).message().get(sourceField)).isEqualTo("2");
        assertThat(response.messages().get(2).message()).doesNotContainKey(targetField);

        assertThat(response.messages().get(3).message().get(sourceField)).isEqualTo("3");
        assertThat(response.messages().get(3).message()).doesNotContainKey(targetField);

        assertThat(response.messages().get(4).message().get("invalid")).isEqualTo("4");
        assertThat(response.messages().get(4).message()).doesNotContainKey(targetField);
    }