Add initial filter classes

consumer/src/main/java/com/flipkart/varadhi/consumer/filtering/FilterProvider.java

@@ -0,0 +1,14 @@
+package com.flipkart.varadhi.consumer.filtering;
+
+import lombok.RequiredArgsConstructor;
+import lombok.extern.slf4j.Slf4j;
+
+@Slf4j
+@RequiredArgsConstructor
+public class FilterProvider {
+
+
+    public static MessageFilter get() {
+        return new MessageFilter();
+    }
+}

consumer/src/main/java/com/flipkart/varadhi/consumer/filtering/MessageFilter.java

@@ -0,0 +1,12 @@
+package com.flipkart.varadhi.consumer.filtering;
+
+import com.flipkart.varadhi.entities.Message;
+
+import java.util.function.Predicate;
+
+public class MessageFilter implements Predicate<Message> {
+    @Override
+    public boolean test(Message message) {
+        return true;
+    }
+}

entities/src/main/java/com/flipkart/varadhi/entities/SubscriptionResource.java

@@ -1,6 +1,7 @@
 package com.flipkart.varadhi.entities;
 
 import com.fasterxml.jackson.annotation.JsonIgnore;
+import com.flipkart.varadhi.entities.filter.FilterPolicy;
 import jakarta.validation.constraints.NotBlank;
 import jakarta.validation.constraints.NotNull;
 import lombok.Getter;
@@ -31,6 +32,8 @@ public class SubscriptionResource extends VersionedEntity implements Validatable
     @NotNull
     ConsumptionPolicy consumptionPolicy;
 
+    FilterPolicy filterPolicy;
+
     private SubscriptionResource(
             String name,
             int version,
@@ -41,7 +44,8 @@ private SubscriptionResource(
             boolean grouped,
             Endpoint endpoint,
             RetryPolicy retryPolicy,
-            ConsumptionPolicy consumptionPolicy
+            ConsumptionPolicy consumptionPolicy,
+            FilterPolicy filterPolicy
     ) {
         super(name, version);
         this.project = project;
@@ -52,6 +56,7 @@ private SubscriptionResource(
         this.endpoint = endpoint;
         this.retryPolicy = retryPolicy;
         this.consumptionPolicy = consumptionPolicy;
+        this.filterPolicy = filterPolicy;
     }
 
     public static SubscriptionResource of(
@@ -63,11 +68,12 @@ public static SubscriptionResource of(
             boolean grouped,
             Endpoint endpoint,
             RetryPolicy retryPolicy,
-            ConsumptionPolicy consumptionPolicy
+            ConsumptionPolicy consumptionPolicy,
+            FilterPolicy filterPolicy
     ) {
         return new SubscriptionResource(
                 name, INITIAL_VERSION, project, topic, topicProject, description, grouped, endpoint, retryPolicy,
-                consumptionPolicy
+                consumptionPolicy, filterPolicy
         );
     }
 
@@ -93,7 +99,8 @@ public static SubscriptionResource from(VaradhiSubscription subscription) {
                 subscription.isGrouped(),
                 subscription.getEndpoint(),
                 subscription.getRetryPolicy(),
-                subscription.getConsumptionPolicy()
+                subscription.getConsumptionPolicy(),
+                subscription.getFilterPolicy()
         );
         subResource.setVersion(subscription.getVersion());
         return subResource;

entities/src/main/java/com/flipkart/varadhi/entities/VaradhiSubscription.java

@@ -2,6 +2,7 @@
 
 import com.fasterxml.jackson.annotation.JsonCreator;
 import com.fasterxml.jackson.annotation.JsonIgnore;
+import com.flipkart.varadhi.entities.filter.FilterPolicy;
 import lombok.*;
 
 @Getter
@@ -17,6 +18,7 @@ public class VaradhiSubscription extends MetaStoreEntity {
     private ConsumptionPolicy consumptionPolicy;
     private SubscriptionShards shards;
     private Status status;
+    private FilterPolicy filterPolicy;
 
 
     private VaradhiSubscription(
@@ -30,7 +32,8 @@ private VaradhiSubscription(
             RetryPolicy retryPolicy,
             ConsumptionPolicy consumptionPolicy,
             SubscriptionShards shards,
-            Status status
+            Status status,
+            FilterPolicy filterPolicy
     ) {
         super(name, version);
         this.project = project;
@@ -45,6 +48,9 @@ private VaradhiSubscription(
         }
         this.shards = shards;
         this.status = status;
+
+        // TODO(aayush): parse and validate filter policy
+        this.filterPolicy = filterPolicy;
     }
 
     public static VaradhiSubscription of(
@@ -56,11 +62,12 @@ public static VaradhiSubscription of(
             Endpoint endpoint,
             RetryPolicy retryPolicy,
             ConsumptionPolicy consumptionPolicy,
-            SubscriptionShards shards
+            SubscriptionShards shards,
+            FilterPolicy filterPolicy
     ) {
         return new VaradhiSubscription(
                 name, INITIAL_VERSION, project, topic, description, grouped, endpoint, retryPolicy, consumptionPolicy,
-                shards, new Status(State.Creating)
+                shards, new Status(State.Creating), filterPolicy
         );
     }
 
@@ -100,6 +107,7 @@ public enum State {
     public static class Status {
         String message;
         State state;
+
         public Status(State state) {
             this.state = state;
         }

entities/src/main/java/com/flipkart/varadhi/entities/filter/FilterExpression.java

@@ -0,0 +1,11 @@
+package com.flipkart.varadhi.entities.filter;
+
+import lombok.Data;
+
+import java.util.List;
+
+@Data
+public class FilterExpression {
+    private final FilterOperation.BooleanOps op;
+    private final List<FilterExpression> exp;
+}

entities/src/main/java/com/flipkart/varadhi/entities/filter/FilterOperation.java

@@ -0,0 +1,13 @@
+package com.flipkart.varadhi.entities.filter;
+
+public enum FilterOperation {
+    startsWith,
+    endsWith,
+    contains,
+    in,
+    exists;
+
+    public enum BooleanOps {
+        AND, OR, NOT;
+    }
+}

entities/src/main/java/com/flipkart/varadhi/entities/filter/FilterPolicy.java

@@ -0,0 +1,12 @@
+package com.flipkart.varadhi.entities.filter;
+
+import lombok.Data;
+
+@Data
+public class FilterPolicy {
+
+    /**
+     * Expression struct that describes the various conditions and subexpressions joined by binary or unary operators.
+     */
+    private final FilterExpression filterExpression;
+}

entities/src/main/java/module-info.java

@@ -7,4 +7,5 @@
     exports com.flipkart.varadhi.entities;
     exports com.flipkart.varadhi.entities.cluster;
     exports com.flipkart.varadhi.entities.auth;
+    exports com.flipkart.varadhi.entities.filter;
 }

entities/src/testFixtures/java/com/flipkart/varadhi/entities/SubProvider.java

@@ -125,7 +125,8 @@ public VaradhiSubscription build(String name, String subProject, String subscrib
                     endpoint == null ? getHttpEndpoint() : endpoint,
                     retryPolicy == null ? getRetryPolicy() : retryPolicy,
                     consumptionPolicy == null ? getConsumptionPolicy() : consumptionPolicy,
-                    shards
+                    shards,
+                    null //TODO(aayush): provide actual filter policy
             );
         }
     }

server/src/main/java/com/flipkart/varadhi/utils/VaradhiSubscriptionFactory.java

@@ -1,6 +1,5 @@
 package com.flipkart.varadhi.utils;
 
-import com.flipkart.varadhi.entities.InternalQueueCategory;
 import com.flipkart.varadhi.entities.*;
 import com.flipkart.varadhi.spi.services.StorageSubscriptionFactory;
 import com.flipkart.varadhi.spi.services.StorageTopicFactory;
@@ -49,7 +48,8 @@ public VaradhiSubscription get(SubscriptionResource subscriptionResource, Projec
                 subscriptionResource.getEndpoint(),
                 subscriptionResource.getRetryPolicy(),
                 subscriptionResource.getConsumptionPolicy(),
-                shards
+                shards,
+                subscriptionResource.getFilterPolicy()
         );
     }
 

server/src/test/java/com/flipkart/varadhi/services/SubscriptionServiceTest.java

@@ -39,7 +39,6 @@
 import org.junit.jupiter.api.Test;
 import org.junit.jupiter.api.extension.ExtendWith;
 
-import java.net.MalformedURLException;
 import java.net.URI;
 import java.util.List;
 import java.util.concurrent.CompletableFuture;
@@ -141,7 +140,8 @@ void testSubscriptionEntitiesSerDe() {
                 false,
                 endpoint,
                 retryPolicy,
-                consumptionPolicy
+                consumptionPolicy,
+                null
         );
 
         VaradhiSubscriptionFactory factory = new VaradhiSubscriptionFactory(pts, psf, ptf, region);
@@ -171,7 +171,9 @@ void getSubscriptionListReturnsCorrectSubscriptions() {
         subscriptionService.createSubscription(unGroupedTopic, sub1, o1t1p1);
         subscriptionService.createSubscription(unGroupedTopic, sub2, o1t1p1);
         subscriptionService.createSubscription(
-                unGroupedTopic, SubscriptionHandlersTest.getUngroupedSubscription("sub3", o1t1p2, unGroupedTopic), o1t1p2);
+                unGroupedTopic, SubscriptionHandlersTest.getUngroupedSubscription("sub3", o1t1p2, unGroupedTopic),
+                o1t1p2
+        );
 
         List<String> actualSubscriptions = subscriptionService.getSubscriptionList(o1t1p1.getName());
 
@@ -263,7 +265,8 @@ void updateSubscriptionUpdatesCorrectly(VertxTestContext ctx) {
         doReturn(unGroupedTopic).when(varadhiMetaStore).getTopic(unGroupedTopic.getName());
         subscriptionService.createSubscription(unGroupedTopic, sub1, o1t1p1);
         VaradhiSubscription update =
-                SubscriptionHandlersTest.getUngroupedSubscription(sub1.getName().split(NAME_SEPARATOR_REGEX)[1], o1t1p1, unGroupedTopic);
+                SubscriptionHandlersTest.getUngroupedSubscription(
+                        sub1.getName().split(NAME_SEPARATOR_REGEX)[1], o1t1p1, unGroupedTopic);
         update.setVersion(1);
         CompletableFuture<SubscriptionStatus> status =
                 CompletableFuture.completedFuture(new SubscriptionStatus(update.getName(), SubscriptionState.STOPPED));
@@ -284,7 +287,8 @@ void updateSubscriptionWithVersionConflictThrows(VertxTestContext ctx) {
         doReturn(unGroupedTopic).when(varadhiMetaStore).getTopic(unGroupedTopic.getName());
         subscriptionService.createSubscription(unGroupedTopic, sub1, o1t1p1);
         VaradhiSubscription update =
-                SubscriptionHandlersTest.getUngroupedSubscription(sub1.getName().split(NAME_SEPARATOR_REGEX)[1], o1t1p1, unGroupedTopic);
+                SubscriptionHandlersTest.getUngroupedSubscription(
+                        sub1.getName().split(NAME_SEPARATOR_REGEX)[1], o1t1p1, unGroupedTopic);
         update.setVersion(2);
         CompletableFuture<SubscriptionStatus> status =
                 CompletableFuture.completedFuture(new SubscriptionStatus(update.getName(), SubscriptionState.STOPPED));

server/src/test/java/com/flipkart/varadhi/web/admin/SubscriptionHandlersTest.java

@@ -1,6 +1,9 @@
 package com.flipkart.varadhi.web.admin;
 
 import com.flipkart.varadhi.entities.*;
+import com.flipkart.varadhi.entities.filter.FilterExpression;
+import com.flipkart.varadhi.entities.filter.FilterOperation;
+import com.flipkart.varadhi.entities.filter.FilterPolicy;
 import com.flipkart.varadhi.exceptions.ResourceNotFoundException;
 import com.flipkart.varadhi.services.ProjectService;
 import com.flipkart.varadhi.services.SubscriptionService;
@@ -38,6 +41,8 @@ public class SubscriptionHandlersTest extends WebTestBase {
     private static final ConsumptionPolicy consumptionPolicy = new ConsumptionPolicy(1, 1, false, 1, null);
     private static final TopicCapacityPolicy capacityPolicy = new TopicCapacityPolicy(1, 10, 1);
     private static final SubscriptionShards shards = new SubscriptionUnitShard(0, capacityPolicy, null, null, null);
+    private static final FilterPolicy filterPolicy =
+            new FilterPolicy(new FilterExpression(FilterOperation.BooleanOps.AND, List.of()));
 
     private final Project project = Project.of("project1", "", "team1", "org1");
     private final TopicResource topicResource = TopicResource.unGrouped("topic1", "project2", null);
@@ -71,7 +76,8 @@ private static VaradhiSubscription getSubscription(
                 endpoint,
                 retryPolicy,
                 consumptionPolicy,
-                shards
+                shards,
+                null
         );
     }
 
@@ -268,7 +274,8 @@ private SubscriptionResource getSubscriptionResource(
                 false,
                 endpoint,
                 retryPolicy,
-                consumptionPolicy
+                consumptionPolicy,
+                null
         );
     }