-
Notifications
You must be signed in to change notification settings - Fork 4
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Refactoring and start consumer integration. (#154)
* Refactoring for - Add active produce/consume topic/subscription to composite entities - Move default topic/node capacity to config from hard coded values. - moved varadhitopic factory/service to server module. - refactoring related to start subscription in consumer
- Loading branch information
Showing
51 changed files
with
397 additions
and
370 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
28 changes: 0 additions & 28 deletions
28
consumer/src/main/java/com/flipkart/varadhi/consumer/StorageRetryTopic.java
This file was deleted.
Oops, something went wrong.
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
42 changes: 35 additions & 7 deletions
42
entities/src/main/java/com/flipkart/varadhi/entities/InternalCompositeSubscription.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -1,15 +1,43 @@ | ||
| package com.flipkart.varadhi.entities; | ||
|
|
||
| import com.fasterxml.jackson.annotation.JsonIgnore; | ||
| import lombok.Data; | ||
| import lombok.AllArgsConstructor; | ||
| import lombok.Getter; | ||
|
|
||
| @Data | ||
| import java.util.ArrayList; | ||
| import java.util.Arrays; | ||
| import java.util.List; | ||
|
|
||
| @Getter | ||
| @AllArgsConstructor | ||
| public class InternalCompositeSubscription { | ||
| private final String subRegion; | ||
| private final InternalQueueType queueType; | ||
| private StorageSubscription<StorageTopic>[] storageSubscriptions; | ||
| private int produceIndex; | ||
| private int consumeIndex; | ||
|
|
||
|
|
||
| public static InternalCompositeSubscription of( | ||
| StorageSubscription<StorageTopic> storageSubscription, InternalQueueType queueType | ||
| ) { | ||
| return new InternalCompositeSubscription(queueType, new StorageSubscription[]{storageSubscription}, 0, 0); | ||
| } | ||
|
|
||
| @JsonIgnore | ||
| public StorageTopic getTopicToProduce() { | ||
| if (queueType.getCategory() == InternalQueueCategory.MAIN) { | ||
| throw new IllegalArgumentException("Main Subscription does not have a topic to produce"); | ||
| } | ||
| return storageSubscriptions[produceIndex].getTopicPartitions().getTopic(); | ||
| } | ||
|
|
||
| @JsonIgnore | ||
| public StorageSubscription<StorageTopic> getSubscriptionToConsume() { | ||
| return storageSubscriptions[consumeIndex]; | ||
| } | ||
|
|
||
| /** | ||
| * As of now only 1 is supported, but in future this can be an array where we can add more storage topics. | ||
| */ | ||
| private final StorageSubscription<StorageTopic> storageSubscription; | ||
| @JsonIgnore | ||
| public List<StorageSubscription<StorageTopic>> getActiveSubscriptions() { | ||
| return new ArrayList<>(Arrays.asList(storageSubscriptions)); | ||
| } | ||
| } |
33 changes: 25 additions & 8 deletions
33
entities/src/main/java/com/flipkart/varadhi/entities/InternalCompositeTopic.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -1,22 +1,39 @@ | ||
| package com.flipkart.varadhi.entities; | ||
|
|
||
|
|
||
| import lombok.Data; | ||
| import com.fasterxml.jackson.annotation.JsonIgnore; | ||
| import lombok.AllArgsConstructor; | ||
| import lombok.Getter; | ||
| import lombok.Setter; | ||
|
|
||
| import java.util.ArrayList; | ||
| import java.util.Arrays; | ||
| import java.util.List; | ||
|
|
||
| /** | ||
| * A wrapper on the storage topic. In future this class will handle the adding additional storage topics for the purpose | ||
| * of increasing partition count without affecting ordering. | ||
| * This concept is internal and is never exposed to the user. | ||
| */ | ||
| @Data | ||
| @Getter | ||
| @AllArgsConstructor | ||
| public class InternalCompositeTopic { | ||
| private StorageTopic[] storageTopics; | ||
| private int produceIndex; | ||
| @Setter | ||
| private TopicState topicState; | ||
|
|
||
| private final String topicRegion; | ||
| public static InternalCompositeTopic of(StorageTopic storageTopic) { | ||
| return new InternalCompositeTopic(new StorageTopic[]{storageTopic}, 0, TopicState.Producing); | ||
| } | ||
|
|
||
| private final TopicState topicState; | ||
| @JsonIgnore | ||
| public StorageTopic getTopicToProduce() { | ||
| return storageTopics[produceIndex]; | ||
| } | ||
|
|
||
| /** | ||
| * As of now only 1 is supported, but in future this can be an array where we can add more storage topics. | ||
| */ | ||
| private final StorageTopic storageTopic; | ||
| @JsonIgnore | ||
| public List<StorageTopic> getActiveTopics() { | ||
| return new ArrayList<>(Arrays.asList(storageTopics)); | ||
| } | ||
| } |
Oops, something went wrong.