Adding consumer spi, varadhi consumer, consumer manager interfaces

consumer/build.gradle

@@ -3,5 +3,9 @@ plugins {
 }
 
 dependencies {
-
+ api(project(':entities'))
+ api(project(':spi'))
+
+ // async interface are using vertx.Future
+ api("io.vertx:vertx-core")
 }

consumer/src/main/java/com/flipkart/varadhi/consumer/ConsumerState.java

@@ -0,0 +1,29 @@
+package com.flipkart.varadhi.consumer;
+
+/**
+ * Status for a consumer that has been started.
+ */
+public enum ConsumerState {
+
+ /**
+ * varadhi consumer will try to make delivery based on the subscription details.
+ */
+ CONSUMING,
+
+ /**
+ * varadhi consumer has trapped an uncaught exception and is haulted.
+ */
+ ERRORED,
+
+ /**
+ * varadhi consumer is delivering at throttled rate, lost likely due to destination errors and subscription's
+ * consumption policy.
+ * Throttled is a wide bucket. TODO: evaluate splitting based on the requirements later on.
+ */
+ THROTTLED,
+
+ /**
+ * varadhi consumer is paused and not making message delivery. But it <i>may</i> be processing failed groups.
+ */
+ PAUSED
+}

consumer/src/main/java/com/flipkart/varadhi/consumer/ConsumersManager.java

@@ -0,0 +1,41 @@
+package com.flipkart.varadhi.consumer;
+
+import com.flipkart.varadhi.entities.ConsumptionPolicy;
+import com.flipkart.varadhi.entities.Endpoint;
+import com.flipkart.varadhi.entities.Project;
+import com.flipkart.varadhi.spi.services.TopicPartitions;
+import io.vertx.core.Future;
+
+/**
+ * Management layer that will manage consumers for multiple subscriptions.
+ */
+public interface ConsumersManager {
+
+ /**
+ * Start consumption for a shard of a subscription. If subscription failed, the future will contain the exception
+ * details.
+ * `shardName` identifies the different shards within the subscription.
+ *
+ * @return
+ */
+ Future<Void> startSubscription(
+ Project project,
+ String subscription,
+ String shardName,
+ TopicPartitions<?> topic,
+ boolean grouped,
+ Endpoint endpoint,
+ ConsumptionPolicy consumptionPolicy,
+ ConsumptionFailurePolicy failurePolicy
+ );
+
+ Future<Void> stopSubscription(String subscription, String shardName);
+
+ Future<Void> pauseSubscription(String subscription, String shardName);
+
+ Future<Void> resumeSubscription(String subscription, String shardName);
+
+ ConsumerState getConsumerState(String subscription, String shardName);
+
+ // TODO likely need status on the starting / stopping as well; as the above status is for a running consumer..
+}

consumer/src/main/java/com/flipkart/varadhi/consumer/ConsumptionFailurePolicy.java

@@ -0,0 +1,17 @@
+package com.flipkart.varadhi.consumer;
+
+import com.flipkart.varadhi.entities.RetryPolicy;
+import com.flipkart.varadhi.entities.StorageTopic;
+import lombok.Getter;
+import lombok.RequiredArgsConstructor;
+
+@Getter
+@RequiredArgsConstructor
+public class ConsumptionFailurePolicy {
+
+ private final RetryPolicy retryPolicy;
+
+ private final StorageRetryTopic retryTopic;
+
+ private final StorageTopic deadLetterTopic;
+}

consumer/src/main/java/com/flipkart/varadhi/consumer/StorageRetryTopic.java

@@ -0,0 +1,28 @@
+package com.flipkart.varadhi.consumer;
+
+import com.fasterxml.jackson.annotation.JsonIgnore;
+import com.flipkart.varadhi.entities.StorageTopic;
+
+public class StorageRetryTopic {
+
+ private final StorageTopic[] retryTopics;
+
+ public StorageRetryTopic(StorageTopic[] retryTopics) {
+ this.retryTopics = retryTopics;
+ }
+
+ /**
+ * @param retryCount 1-based retry count
+ *
+ * @return the storage topic for the given retry count
+ */
+ @JsonIgnore
+ public StorageTopic getTopicForRetry(int retryCount) {
+ return retryTopics[retryCount - 1];
+ }
+
+ @JsonIgnore
+ public int getMaxRetryCount() {
+ return retryTopics.length;
+ }
+}

consumer/src/main/java/com/flipkart/varadhi/consumer/VaradhiConsumer.java

@@ -0,0 +1,40 @@
+package com.flipkart.varadhi.consumer;
+
+import com.flipkart.varadhi.entities.VaradhiSubscription;
+
+/**
+ * Base varadhi consumer to consume messages from a varadhi topic. It corresponds to a single shard of some
+ * subscription.
+ */
+public interface VaradhiConsumer {
+
+ String getSubscriptionName();
+
+ int getShardId();
+
+ ConsumerState getState();
+
+ /**
+ * Start the consumer. It will start message delivery from the last committed offset. If no offset is committed, it
+ * can start from the earliest or latest offset based on the implementation.
+ *
+ * @throws Exception bad configuration can result in exception in which case the consumer will not start.
+ */
+ void start() throws Exception;
+
+ /**
+ * Pause the consumer. It just means that message delivery will stop happening immediately.
+ */
+ void pause();
+
+ /**
+ * Resume from paused state. It means that message delivery will start happening immediately.
+ */
+ void resume();
+
+ /**
+ * Close the consumer. It will stop message delivery and release all resources. After calling this calling other
+ * methods on this consumer will result in an exception.
+ */
+ void close();
+}

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

@@ -0,0 +1,17 @@
+package com.flipkart.varadhi.entities;
+
+
+import com.fasterxml.jackson.annotation.JsonSubTypes;
+import com.fasterxml.jackson.annotation.JsonTypeInfo;
+import lombok.Getter;
+import lombok.RequiredArgsConstructor;
+
+@Getter
+@RequiredArgsConstructor
+@JsonTypeInfo(use = JsonTypeInfo.Id.NAME, include = JsonTypeInfo.As.PROPERTY, property = "@shardsStrategy")
+@JsonSubTypes({
+ @JsonSubTypes.Type(value = UnitSubscriptionShard.class, name = "unit"),
+})
+public abstract class SubscriptionShards {
+ private final int shardCount;
+}

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

@@ -0,0 +1,20 @@
+package com.flipkart.varadhi.entities;
+
+import lombok.Getter;
+
+@Getter
+public class UnitSubscriptionShard extends SubscriptionShards {
+
+ private final int shardId;
+
+ private final RetryTopic retryTopic;
+
+ private final InternalCompositeTopic deadLetterTopic;
+
+ public UnitSubscriptionShard(int shardId, RetryTopic retryTopic, InternalCompositeTopic deadLetterTopic) {
+ super(1);
+ this.shardId = shardId;
+ this.retryTopic = retryTopic;
+ this.deadLetterTopic = deadLetterTopic;
+ }
+}

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

@@ -13,7 +13,7 @@ public class VaradhiSubscription extends MetaStoreEntity {
  Endpoint endpoint;
  RetryPolicy retryPolicy;
  ConsumptionPolicy consumptionPolicy;
- SubscriptionShard[] shards;
+ SubscriptionShards shards;
 
  public VaradhiSubscription(
  String name,
@@ -25,7 +25,7 @@ public VaradhiSubscription(
  Endpoint endpoint,
  RetryPolicy retryPolicy,
  ConsumptionPolicy consumptionPolicy,
- SubscriptionShard[] shards
+ SubscriptionShards shards
  ) {
  super(name, version);
  this.project = project;
@@ -35,7 +35,7 @@ public VaradhiSubscription(
  this.endpoint = endpoint;
  this.retryPolicy = retryPolicy;
  this.consumptionPolicy = consumptionPolicy;
- if (shards == null || shards.length == 0) {
+ if (shards == null || shards.getShardCount() <= 0) {
  throw new IllegalArgumentException("shards cannot be null or empty");
  }
  this.shards = shards;

server/build.gradle

@@ -14,6 +14,7 @@ dependencies {
  implementation(project(":messaging"))
  implementation(project(":entities"))
  implementation(project(":authz"))
+ implementation(project(":consumer"))
 
  implementation("org.apache.logging.log4j:log4j-slf4j2-impl")
  implementation("org.apache.logging.log4j:log4j-core")

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

@@ -1,7 +1,8 @@
 package com.flipkart.varadhi.utils;
 
+import com.flipkart.varadhi.entities.UnitSubscriptionShard;
 import com.flipkart.varadhi.entities.SubscriptionResource;
-import com.flipkart.varadhi.entities.SubscriptionShard;
+import com.flipkart.varadhi.entities.SubscriptionShards;
 import com.flipkart.varadhi.entities.VaradhiSubscription;
 
 import static com.flipkart.varadhi.entities.VersionedEntity.NAME_SEPARATOR;
@@ -11,8 +12,7 @@ public final class SubscriptionHelper {
  private SubscriptionHelper() {
  }
 
- private static final SubscriptionShard[] dummyShards =
- new SubscriptionShard[]{new SubscriptionShard(0, null, null, null)};
+ private static final SubscriptionShards dummyShards = new UnitSubscriptionShard(0, null, null);
 
  public static VaradhiSubscription fromResource(SubscriptionResource subscriptionResource, int version) {
  return new VaradhiSubscription(

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

@@ -35,8 +35,7 @@ public class SubscriptionHandlersTest extends WebTestBase {
  1, 1, 1, 1
  );
  private static final ConsumptionPolicy consumptionPolicy = new ConsumptionPolicy(1, 1, false, 1, null);
- private static final SubscriptionShard[] shards =
- new SubscriptionShard[]{new SubscriptionShard(0, null, null, null)};
+ private static final SubscriptionShards shards = new UnitSubscriptionShard(0, null, null);
 
  static {
  try {

settings.gradle

@@ -8,4 +8,4 @@
  */
 
 rootProject.name = 'varadhi'
-include('entities', 'spi', 'common', 'core', 'messaging', 'pulsar', 'server', 'authz', 'consumer')
+include('entities', 'spi', 'common', 'core', 'messaging', 'pulsar', 'consumer', 'authz', 'server')

spi/src/main/java/com/flipkart/varadhi/spi/services/Consumer.java

@@ -0,0 +1,34 @@
+package com.flipkart.varadhi.spi.services;
+
+import com.flipkart.varadhi.entities.Offset;
+
+import java.util.concurrent.CompletableFuture;
+
+/**
+ * Consumer interface to receive messages from a topic.
+ */
+public interface Consumer<O extends Offset> {
+
+ /**
+ * Receive a batch of messages from the subscribed topics.
+ *
+ * @return
+ */
+ CompletableFuture<PolledMessages<O>> receiveAsync();
+
+ /**
+ * Commit upto the offset, signifying that all messages upto (& including) the offset have been processed.
+ *
+ * @param topic
+ * @param partition
+ * @param offset
+ *
+ * @return
+ */
+ CompletableFuture<Void> commitCumulative(String topic, int partition, O offset);
+
+ /**
+ * Unsubscribes the topics and closes the consumer object and any associated resources.
+ */
+ void close();
+}

spi/src/main/java/com/flipkart/varadhi/spi/services/ConsumerFactory.java

@@ -0,0 +1,21 @@
+package com.flipkart.varadhi.spi.services;
+
+import com.flipkart.varadhi.entities.Offset;
+import com.flipkart.varadhi.entities.StorageTopic;
+
+import java.util.Collection;
+
+public interface ConsumerFactory<T extends StorageTopic, O extends Offset> {
+
+ // TODO: need to see if we need property bag param or other params
+
+ /**
+ * @param topics to be consumed
+ * @param name name of the consumer. synonymous to consumer group / subscription name
+ *
+ * @return Consumer instance
+ *
+ * @throws MessagingException
+ */
+ Consumer<O> getConsumer(Collection<TopicPartitions<T>> topics, String name) throws MessagingException;
+}

spi/src/main/java/com/flipkart/varadhi/spi/services/PartitionMessages.java

@@ -0,0 +1,17 @@
+package com.flipkart.varadhi.spi.services;
+
+import com.flipkart.varadhi.entities.Offset;
+import lombok.AllArgsConstructor;
+
+/**
+ * Represent a batch of messages polled from a particular partition.
+ */
+@AllArgsConstructor
+public class PartitionMessages<O extends Offset> {
+
+ private final String topic;
+
+ private final int partition;
+
+ private final PolledMessage<O> messages;
+}