Added filter order optimizations

data-plane/benchmarks/resources/filter-class-list.txt

@@ -2,3 +2,5 @@ ExactFilterBenchmark
 NotFilterBenchmark
 PrefixFilterBenchmark
 SuffixFilterBenchmark
+AnyFilterBenchmark
+AllFilterBenchmark

data-plane/benchmarks/src/main/java/dev/knative/eventing/kafka/broker/dispatcher/impl/filter/AllFilterBenchmark.java

@@ -16,16 +16,17 @@
 
 package dev.knative.eventing.kafka.broker.dispatcher.impl.filter;
 
+import com.google.common.collect.ImmutableList;
 import dev.knative.eventing.kafka.broker.dispatcher.Filter;
 import dev.knative.eventing.kafka.broker.dispatcher.impl.filter.subscriptionsapi.AllFilter;
 import dev.knative.eventing.kafka.broker.dispatcher.impl.filter.subscriptionsapi.ExactFilter;
 import dev.knative.eventing.kafka.broker.dispatcher.impl.filter.subscriptionsapi.PrefixFilter;
 import dev.knative.eventing.kafka.broker.dispatcher.impl.filter.subscriptionsapi.SuffixFilter;
 import io.cloudevents.CloudEvent;
-import java.util.List;
 import java.util.Map;
 
 public class AllFilterBenchmark {
+
     public static ExactFilter makeExactFilter() {
         return new ExactFilter(Map.of("type", "com.github.pull.create"));
     }
@@ -50,7 +51,7 @@ public static class AllFilterWithExactFilter extends FilterBenchmark {
 
         @Override
         protected Filter createFilter() {
-            return new AllFilter(List.of(makeExactFilter()));
+            return new AllFilter(ImmutableList.of(makeExactFilter()), vertx, FILTER_REORDER_TIME_MILLISECONDS);
         }
 
         @Override
@@ -63,7 +64,10 @@ public static class AllFilterMatchAllSubFilters extends FilterBenchmark {
 
         @Override
         protected Filter createFilter() {
-            return new AllFilter(List.of(makeExactFilter(), makePrefixFilter(), makeSuffixFilter()));
+            return new AllFilter(
+                    ImmutableList.of(makeExactFilter(), makePrefixFilter(), makeSuffixFilter()),
+                    vertx,
+                    FILTER_REORDER_TIME_MILLISECONDS);
         }
 
         @Override
@@ -76,7 +80,10 @@ public static class AllFilterFirstMatchEndOfArray extends FilterBenchmark {
 
         @Override
         protected Filter createFilter() {
-            return new AllFilter(List.of(makePrefixFilterNoMatch(), makeSuffixFilterNoMatch(), makeExactFilter()));
+            return new AllFilter(
+                    ImmutableList.of(makePrefixFilterNoMatch(), makeSuffixFilterNoMatch(), makeExactFilter()),
+                    vertx,
+                    FILTER_REORDER_TIME_MILLISECONDS);
         }
 
         @Override
@@ -89,7 +96,10 @@ public static class AllFilterFirstMatchStartOfArray extends FilterBenchmark {
 
         @Override
         protected Filter createFilter() {
-            return new AllFilter(List.of(makeExactFilter(), makePrefixFilterNoMatch(), makeSuffixFilterNoMatch()));
+            return new AllFilter(
+                    ImmutableList.of(makeExactFilter(), makePrefixFilterNoMatch(), makeSuffixFilterNoMatch()),
+                    vertx,
+                    FILTER_REORDER_TIME_MILLISECONDS);
         }
 
         @Override
@@ -102,7 +112,10 @@ public static class AllFilterOneNonMatchingFilterInMiddle extends FilterBenchmar
 
         @Override
         protected Filter createFilter() {
-            return new AllFilter(List.of(makeExactFilter(), makePrefixFilterNoMatch(), makePrefixFilter()));
+            return new AllFilter(
+                    ImmutableList.of(makeExactFilter(), makePrefixFilterNoMatch(), makePrefixFilter()),
+                    vertx,
+                    FILTER_REORDER_TIME_MILLISECONDS);
         }
 
         @Override
@@ -115,7 +128,10 @@ public static class AllFilterNoMatchingFilters extends FilterBenchmark {
 
         @Override
         protected Filter createFilter() {
-            return new AllFilter(List.of(makePrefixFilterNoMatch(), makeSuffixFilterNoMatch()));
+            return new AllFilter(
+                    ImmutableList.of(makePrefixFilterNoMatch(), makeSuffixFilterNoMatch()),
+                    vertx,
+                    FILTER_REORDER_TIME_MILLISECONDS);
         }
 
         @Override

data-plane/benchmarks/src/main/java/dev/knative/eventing/kafka/broker/dispatcher/impl/filter/AnyFilterBenchmark.java

@@ -16,10 +16,10 @@
 
 package dev.knative.eventing.kafka.broker.dispatcher.impl.filter;
 
+import com.google.common.collect.ImmutableList;
 import dev.knative.eventing.kafka.broker.dispatcher.Filter;
 import dev.knative.eventing.kafka.broker.dispatcher.impl.filter.subscriptionsapi.*;
 import io.cloudevents.CloudEvent;
-import java.util.List;
 import java.util.Map;
 
 public class AnyFilterBenchmark {
@@ -48,15 +48,15 @@ public static PrefixFilter makePrefixFilterNoMatch() {
         return new PrefixFilter(Map.of("type", "other.event"));
     }
 
-    public static SuffixFilter makeSufficFilterNoMatch() {
+    public static SuffixFilter makeSuffixFilterNoMatch() {
         return new SuffixFilter(Map.of("source", "qwertyuiop"));
     }
 
     public static class AnyFilterWithExactFilterBenchmark extends FilterBenchmark {
 
         @Override
         protected Filter createFilter() {
-            return new AnyFilter(List.of(makeExactFilter()));
+            return new AnyFilter(ImmutableList.of(makeExactFilter()), vertx, FILTER_REORDER_TIME_MILLISECONDS);
         }
 
         @Override
@@ -69,7 +69,10 @@ public static class AnyFilterMatchAllSubfilters extends FilterBenchmark {
 
         @Override
         protected Filter createFilter() {
-            return new AnyFilter(List.of(makeExactFilter(), makePrefixFilter(), makeSuffixFilter()));
+            return new AnyFilter(
+                    ImmutableList.of(makeExactFilter(), makePrefixFilter(), makeSuffixFilter()),
+                    vertx,
+                    FILTER_REORDER_TIME_MILLISECONDS);
         }
 
         @Override
@@ -82,7 +85,10 @@ public static class AnyFilterFirstMatchAtEnd extends FilterBenchmark {
 
         @Override
         protected Filter createFilter() {
-            return new AnyFilter(List.of(makePrefixFilterNoMatch(), makeSufficFilterNoMatch(), makeExactFilter()));
+            return new AnyFilter(
+                    ImmutableList.of(makePrefixFilterNoMatch(), makeSuffixFilterNoMatch(), makeExactFilter()),
+                    vertx,
+                    FILTER_REORDER_TIME_MILLISECONDS);
         }
 
         @Override
@@ -95,7 +101,10 @@ public static class AnyFilterFirstMatchAtStart extends FilterBenchmark {
 
         @Override
         protected Filter createFilter() {
-            return new AnyFilter(List.of(makeExactFilter(), makePrefixFilterNoMatch(), makeSufficFilterNoMatch()));
+            return new AnyFilter(
+                    ImmutableList.of(makeExactFilter(), makePrefixFilterNoMatch(), makeSuffixFilterNoMatch()),
+                    vertx,
+                    FILTER_REORDER_TIME_MILLISECONDS);
         }
 
         @Override
@@ -108,7 +117,10 @@ public static class AnyFilter2EventsMatch2DifferentFilters extends FilterBenchma
 
         @Override
         protected Filter createFilter() {
-            return new AnyFilter(List.of(makePrefixFilter(), makePrefixFilterNoMatch()));
+            return new AnyFilter(
+                    ImmutableList.of(makePrefixFilter(), makePrefixFilterNoMatch()),
+                    vertx,
+                    FILTER_REORDER_TIME_MILLISECONDS);
         }
 
         @Override
@@ -120,7 +132,10 @@ protected CloudEvent createEvent() {
     public static class AnyFilter2EventsMatch2DifferentFiltersOneFilterMatchesNeither extends FilterBenchmark {
         @Override
         protected Filter createFilter() {
-            return new AnyFilter(List.of(makeSufficFilterNoMatch(), makePrefixFilter(), makePrefixFilterNoMatch()));
+            return new AnyFilter(
+                    ImmutableList.of(makeSuffixFilterNoMatch(), makePrefixFilter(), makePrefixFilterNoMatch()),
+                    vertx,
+                    FILTER_REORDER_TIME_MILLISECONDS);
         }
 
         @Override

data-plane/benchmarks/src/main/java/dev/knative/eventing/kafka/broker/dispatcher/impl/filter/FilterBenchmark.java

@@ -18,6 +18,7 @@
 
 import dev.knative.eventing.kafka.broker.dispatcher.Filter;
 import io.cloudevents.CloudEvent;
+import io.vertx.core.Vertx;
 import java.util.concurrent.TimeUnit;
 import org.openjdk.jmh.annotations.*;
 import org.openjdk.jmh.infra.Blackhole;
@@ -32,8 +33,18 @@ public abstract class FilterBenchmark {
     Filter filter;
     CloudEvent cloudEvent;
 
+    Vertx vertx;
+
+    public static final long FILTER_REORDER_TIME_MILLISECONDS = 10000; // 1 seconds
+
+    @TearDown
+    public void closeVertx() {
+        this.vertx.close();
+    }
+
     @Setup(Level.Trial)
     public void setupFilter() {
+        this.vertx = Vertx.vertx();
         this.filter = createFilter();
     }
 
@@ -42,13 +53,20 @@ public void setupCloudEvent() {
         this.cloudEvent = createEvent();
     }
 
+    @TearDown(Level.Trial)
+    public void teardown() {
+        this.filter.close(vertx);
+    }
+
     protected abstract Filter createFilter();
 
     protected abstract CloudEvent createEvent();
 
     @Benchmark
     public void benchmarkFilterCreation(Blackhole bh) {
-        bh.consume(this.createFilter());
+        final var filter = this.createFilter();
+        filter.close(vertx);
+        bh.consume(filter);
     }
 
     @Benchmark

data-plane/dispatcher/src/main/java/dev/knative/eventing/kafka/broker/dispatcher/Filter.java

@@ -16,6 +16,7 @@
 package dev.knative.eventing.kafka.broker.dispatcher;
 
 import io.cloudevents.CloudEvent;
+import io.vertx.core.Vertx;
 import java.util.function.Predicate;
 
 /**
@@ -30,4 +31,18 @@ public interface Filter extends Predicate<CloudEvent> {
     static Filter noop() {
         return ce -> true;
     }
+
+    default int getCount() {
+        return 0;
+    }
+    ;
+
+    default int incrementCount() {
+        return 0;
+    }
+    ;
+
+    default void close(Vertx vertx) {
+        return;
+    }
 }

data-plane/dispatcher/src/main/java/dev/knative/eventing/kafka/broker/dispatcher/impl/RecordDispatcherImpl.java

@@ -77,6 +77,8 @@ public class RecordDispatcherImpl implements RecordDispatcher {
     private static final String EKB_ERROR_PREFIX = "kne-";
     private static final int KN_ERROR_DATA_MAX_BYTES = 1024;
 
+    private final Vertx vertx;
+
     private final Filter filter;
     private final Function<ConsumerRecord<Object, CloudEvent>, Future<HttpResponse<?>>> subscriberSender;
     private final Function<ConsumerRecord<Object, CloudEvent>, Future<HttpResponse<?>>> dlsSender;
@@ -103,6 +105,7 @@ public class RecordDispatcherImpl implements RecordDispatcher {
      * @param consumerTracer           consumer tracer
      */
     public RecordDispatcherImpl(
+            final Vertx vertx,
             final ConsumerVerticleContext consumerVerticleContext,
             final Filter filter,
             final CloudEventSender subscriberSender,
@@ -111,13 +114,15 @@ public RecordDispatcherImpl(
             final RecordDispatcherListener recordDispatcherListener,
             final ConsumerTracer consumerTracer,
             final MeterRegistry meterRegistry) {
+        Objects.requireNonNull(vertx, "provide vertx");
         Objects.requireNonNull(consumerVerticleContext, "provide consumerVerticleContext");
         Objects.requireNonNull(filter, "provide filter");
         Objects.requireNonNull(subscriberSender, "provide subscriberSender");
         Objects.requireNonNull(deadLetterSinkSender, "provide deadLetterSinkSender");
         Objects.requireNonNull(recordDispatcherListener, "provide offsetStrategy");
         Objects.requireNonNull(responseHandler, "provide sinkResponseHandler");
 
+        this.vertx = vertx;
         this.consumerVerticleContext = consumerVerticleContext;
         this.filter = filter;
         this.subscriberSender = composeSenderAndSinkHandler(subscriberSender, responseHandler, "subscriber");
@@ -503,6 +508,8 @@ private void recordReceived(final ConsumerRecordContext recordContext) {
     public Future<Void> close() {
         this.closed.set(true);
 
+        this.filter.close(vertx);
+
         Metrics.searchEgressMeters(
                         meterRegistry, consumerVerticleContext.getEgress().getReference())
                 .forEach(meterRegistry::remove);

data-plane/dispatcher/src/main/java/dev/knative/eventing/kafka/broker/dispatcher/impl/filter/AttributesFilter.java

@@ -64,6 +64,8 @@ public AttributeEntry(String name, String expectedValue, Function<CloudEvent, St
 
     private final List<AttributeEntry> attributes;
 
+    private int count;
+
     /**
      * All args constructor.
      *
@@ -81,6 +83,7 @@ public AttributesFilter(final Map<String, String> attributes) {
                             }
                         })))
                 .collect(Collectors.toUnmodifiableList());
+        this.count = 0;
     }
 
     /**
@@ -132,4 +135,14 @@ private static <T> String getOrDefault(@Nullable final T s, final Function<T, St
     private static boolean isNotEmpty(final String value) {
         return !(value == null || value.isEmpty());
     }
+
+    @Override
+    public int getCount() {
+        return this.count;
+    }
+
+    @Override
+    public int incrementCount() {
+        return this.count++;
+    }
 }