-
Notifications
You must be signed in to change notification settings - Fork 91
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
[server] Pass ingestion role change for store version from StoreInges…
…tionTask to AggKafkaConsumerService. (#1034) This PR introduced a new class TopicPartitionReplicaRole to bundle information about particular topic partition: 1. Version role: future/current/backup 2. Workload type: Active Active or Write Compute due to its nature of heavy processing needed. And a new config SERVER_RESUBSCRIPTION_TRIGGERED_BY_VERSION_INGESTION_CONTEXT_CHANGE_ENABLED to turn on/off the resubscription. StoreIngestionTask thread will periodically check the store's version role changed and store version's workload type changed. If changed, it will trigger resubscription by graceful un-subscription/ re-subscription for all partitions to let bottom AggKafkaConsumerService to find corresponding consumer thread pool for them. In the future KafkaConsumerService aware of store version role during subscription, there will be more information with TopicPartitionReplicaRole passed with aggKafkaConsumerService#subscribe. In the future, we could also bundle more information to this object to pass into KafkaConsumerService to let it do better consumer pool allocation. With proper allocation of consumer thread pool, we could achieve prioritization on certain type of traffic (e.g. we could allocate larger consumer thread for current hybrid leader ingestion). A unit test was added to verify the subscription/subscription happened for local and remote ingestion, by intentionally triggering version role change during ingestion. Co-authored-by: Hao Xu <[email protected]>
- Loading branch information
Showing
18 changed files
with
500 additions
and
110 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
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
102 changes: 102 additions & 0 deletions
102
...t/src/main/java/com/linkedin/davinci/kafka/consumer/PartitionReplicaIngestionContext.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 |
|---|---|---|
| @@ -0,0 +1,102 @@ | ||
| package com.linkedin.davinci.kafka.consumer; | ||
|
|
||
| import com.linkedin.venice.exceptions.VeniceException; | ||
| import com.linkedin.venice.exceptions.VeniceNoStoreException; | ||
| import com.linkedin.venice.meta.Store; | ||
| import com.linkedin.venice.meta.Version; | ||
| import com.linkedin.venice.pubsub.api.PubSubTopic; | ||
| import com.linkedin.venice.pubsub.api.PubSubTopicPartition; | ||
| import org.apache.logging.log4j.LogManager; | ||
| import org.apache.logging.log4j.Logger; | ||
|
|
||
|
|
||
| /** | ||
| * This class is for wrapping the information about the role of the partition replica on that host to | ||
| * {@link AggKafkaConsumerService} to achieve finer granularity of consumer assignment. Those information should be | ||
| * triggered by store version role (future, current and backup), workload type and leader or follower state. Version role | ||
| * and workload type information are properly managed by {@link StoreIngestionTask} to be sent to | ||
| * {@link AggKafkaConsumerService}. We could add more information regarding this partition replica if needed in the future. | ||
| */ | ||
|
|
||
| public class PartitionReplicaIngestionContext { | ||
| private static final Logger LOGGER = LogManager.getLogger(PartitionReplicaIngestionContext.class); | ||
| private final PubSubTopicPartition pubSubTopicPartition; | ||
| private final PubSubTopic versionTopic; | ||
| private final VersionRole versionRole; | ||
| private final WorkloadType workloadType; | ||
|
|
||
| public enum VersionRole { | ||
| CURRENT, BACKUP, FUTURE | ||
| } | ||
|
|
||
| // TODO: Add more workload types if needed, here we only care about active active or write compute workload. | ||
| public enum WorkloadType { | ||
| AA_OR_WRITE_COMPUTE, NON_AA_OR_WRITE_COMPUTE | ||
| } | ||
|
|
||
| public PartitionReplicaIngestionContext( | ||
| PubSubTopic versionTopic, | ||
| PubSubTopicPartition pubSubTopicPartition, | ||
| VersionRole versionRole, | ||
| WorkloadType workloadType) { | ||
| this.versionTopic = versionTopic; | ||
| this.pubSubTopicPartition = pubSubTopicPartition; | ||
| this.versionRole = versionRole; | ||
| this.workloadType = workloadType; | ||
| } | ||
|
|
||
| public VersionRole getVersionRole() { | ||
| return versionRole; | ||
| } | ||
|
|
||
| public PubSubTopicPartition getPubSubTopicPartition() { | ||
| return pubSubTopicPartition; | ||
| } | ||
|
|
||
| public PubSubTopic getVersionTopic() { | ||
| return versionTopic; | ||
| } | ||
|
|
||
| public WorkloadType getWorkloadType() { | ||
| return workloadType; | ||
| } | ||
|
|
||
| public static WorkloadType getWorkloadType(PubSubTopic versionTopic, Store store) { | ||
| checkVersionTopicStoreOrThrow(versionTopic, store); | ||
| int versionNumber = Version.parseVersionFromKafkaTopicName(versionTopic.getName()); | ||
| Version version = store.getVersion(versionNumber); | ||
| if (version == null) { | ||
| return WorkloadType.NON_AA_OR_WRITE_COMPUTE; | ||
| } | ||
| if (store.isWriteComputationEnabled() || version.isActiveActiveReplicationEnabled()) { | ||
| return WorkloadType.AA_OR_WRITE_COMPUTE; | ||
| } | ||
| return WorkloadType.NON_AA_OR_WRITE_COMPUTE; | ||
| } | ||
|
|
||
| public static VersionRole getStoreVersionRole(PubSubTopic versionTopic, Store store) { | ||
| checkVersionTopicStoreOrThrow(versionTopic, store); | ||
| int versionNumber = Version.parseVersionFromKafkaTopicName(versionTopic.getName()); | ||
| int currentVersionNumber = store.getCurrentVersion(); | ||
| if (currentVersionNumber < versionNumber) { | ||
| return VersionRole.FUTURE; | ||
| } else if (currentVersionNumber > versionNumber) { | ||
| return VersionRole.BACKUP; | ||
| } else { | ||
| return VersionRole.CURRENT; | ||
| } | ||
| } | ||
|
|
||
| private static void checkVersionTopicStoreOrThrow(PubSubTopic versionTopic, Store store) { | ||
| if (store == null) { | ||
| LOGGER.error("Invalid store meta-data for {}", versionTopic); | ||
| throw new VeniceNoStoreException(versionTopic.getStoreName()); | ||
| } | ||
|
|
||
| if (!store.getName().equals(versionTopic.getStoreName())) { | ||
| throw new VeniceException( | ||
| "Store name mismatch for store " + store.getName() + " and version topic " + versionTopic); | ||
| } | ||
| } | ||
|
|
||
| } |
Oops, something went wrong.