@Override
    public FilteredMessage apply(Message msg) {
        try (var ignored = executionTime.time()) {
            return doApply(msg);
        }
    }
@Test
    void applyWithFilter(MessageFactory messageFactory) {
        final var filterRules = List.of(
                RuleDao.builder()
                        .id("668cff8ed9f9636b4b629c29")
                        .title("[668cff8ed9f9636b4b629c29] Test Filter 1")
                        .source("""
                                rule "[668cff8ed9f9636b4b629c29] Test Filter 1"
                                when has_field(field : "source")
                                then
                                  __remove_from_stream_destination__(stream_id : "000000000000000000000001", destination_type : "indexer");
                                end
                                """)
                        .build()
        );

        final var pipeline = Pipeline.builder()
                .id(defaultStream.getId()) // Must be the stream ID to make PipelineRuleOutputFilterState#getPipelinesForMessage work
                .name("Test")
                .stages(ImmutableSortedSet.of(
                        Stage.builder()
                                .stage(0)
                                .match(Stage.Match.EITHER)
                                .ruleReferences(filterRules.stream().map(RuleDao::title).toList())
                                .build()
                ))
                .build();

        final var resolver = createResolver(filterRules);
        final var pipelines = resolver.resolveFunctions(Set.of(pipeline), PipelineMetricRegistry.create(new MetricRegistry(), "P", "R"));

        final var filter = createFilter(pipelines, Set.of("indexer", "other"));
        final var message = messageFactory.createMessage("msg", "src", Tools.nowUTC());
        message.addStream(defaultStream);

        final var filteredMessage = filter.apply(message);

        assertThat(filteredMessage.message()).isEqualTo(ImmutableMessage.wrap(message));
        // The filter rules removes the default stream from the indexer destination, so it shouldn't be in the
        // destinations after running the filter.
        assertThat(filteredMessage.destinations().keySet()).containsExactlyInAnyOrder("other");
        assertThat(filteredMessage.destinations().get("indexer")).isEmpty();
        assertThat(filteredMessage.destinations().get("other")).containsExactlyInAnyOrder(defaultStream);
    }