add largestUsedRTVersionNumber in store config

Author: arjun4084346

Diff for: internal/venice-common/src/main/java/com/linkedin/venice/controllerapi/ControllerApiConstants.java

@@ -71,6 +71,7 @@ public class ControllerApiConstants {
   public static final String BATCH_GET_ROUTER_CACHE_ENABLED = "batch_get_router_cache_enabled";
   public static final String BATCH_GET_LIMIT = "batch_get_limit";
   public static final String LARGEST_USED_VERSION_NUMBER = "largest_used_version_number";
+  public static final String LARGEST_USED_RT_VERSION_NUMBER = "largest_used_rt_version_number";
   public static final String NUM_VERSIONS_TO_PRESERVE = "num_versions_to_preserve";
   public static final String DISABLE_META_STORE = "disable_meta_store";
   public static final String DISABLE_DAVINCI_PUSH_STATUS_STORE = "disable_davinci_push_status_store";

Diff for: internal/venice-common/src/main/java/com/linkedin/venice/helix/HelixStoreGraveyard.java

@@ -77,6 +77,31 @@ public int getLargestUsedVersionNumber(String storeName) {
     return largestUsedVersionNumber;
   }
 
+  @Override
+  public int getLargestUsedRTVersionNumber(String storeName) {
+    if (VeniceSystemStoreUtils.isSystemStore(storeName)) {
+      VeniceSystemStoreType systemStoreType = VeniceSystemStoreType.getSystemStoreType(storeName);
+      if (systemStoreType != null && systemStoreType.isStoreZkShared()) {
+        String userStoreName = systemStoreType.extractRegularStoreName(storeName);
+        return getPerUserStoreSystemStoreLargestUsedRTVersionNumber(userStoreName, systemStoreType);
+      }
+    }
+
+    List<Store> stores = getStoreFromAllClusters(storeName);
+    if (stores.isEmpty()) {
+      // If store does NOT existing in graveyard, it means store has never been deleted, return 0 which is the default
+      // value of largestUsedRTVersionNumber for a new store.
+      return Store.NON_EXISTING_VERSION;
+    }
+    int largestUsedRTVersionNumber = Store.NON_EXISTING_VERSION;
+    for (Store deletedStore: stores) {
+      if (deletedStore.getLargestUsedVersionNumber() > largestUsedRTVersionNumber) {
+        largestUsedRTVersionNumber = deletedStore.getLargestUsedRTVersionNumber();
+      }
+    }
+    return largestUsedRTVersionNumber;
+  }
+
   @Override
   public void putStoreIntoGraveyard(String clusterName, Store store) {
     int largestUsedVersionNumber = getLargestUsedVersionNumber(store.getName());
@@ -196,6 +221,35 @@ private int getPerUserStoreSystemStoreLargestUsedVersionNumber(
     return largestUsedVersionNumber;
   }
 
+  private int getPerUserStoreSystemStoreLargestUsedRTVersionNumber(
+      String userStoreName,
+      VeniceSystemStoreType systemStoreType) {
+    String systemStoreName = systemStoreType.getSystemStoreName(userStoreName);
+    List<Store> deletedStores = getStoreFromAllClusters(userStoreName);
+    if (deletedStores.isEmpty()) {
+      LOGGER.info(
+          "User store: {} does NOT exist in the store graveyard. Hence, no largest used RT version for its system store: {}",
+          userStoreName,
+          systemStoreName);
+      return Store.NON_EXISTING_VERSION;
+    }
+    int largestUsedRTVersionNumber = Store.NON_EXISTING_VERSION;
+    for (Store deletedStore: deletedStores) {
+      Map<String, SystemStoreAttributes> systemStoreNamesToAttributes = deletedStore.getSystemStores();
+      SystemStoreAttributes systemStoreAttributes =
+          systemStoreNamesToAttributes.get(VeniceSystemStoreType.getSystemStoreType(systemStoreName).getPrefix());
+      if (systemStoreAttributes != null) {
+        largestUsedRTVersionNumber =
+            Math.max(largestUsedRTVersionNumber, systemStoreAttributes.getLargestUsedRTVersionNumber());
+      }
+    }
+
+    if (largestUsedRTVersionNumber == Store.NON_EXISTING_VERSION) {
+      LOGGER.info("Can not find largest used RT version number for {}.", systemStoreName);
+    }
+    return largestUsedRTVersionNumber;
+  }
+
   private String getGeneralStoreGraveyardPath() {
     return getStoreGraveyardPath(PathResourceRegistry.WILDCARD_MATCH_ANY, PathResourceRegistry.WILDCARD_MATCH_ANY);
   }

Diff for: internal/venice-common/src/main/java/com/linkedin/venice/meta/ReadOnlyStore.java

@@ -715,6 +715,16 @@ public void setLargestUsedVersionNumber(int largestUsedVersionNumber) {
       throw new UnsupportedOperationException();
     }
 
+    @Override
+    public int getLargestUsedRTVersionNumber() {
+      return this.delegate.getLargestUsedRTVersionNumber();
+    }
+
+    @Override
+    public void setLargestUsedRTVersionNumber(int largestUsedRTVersionNumber) {
+      throw new UnsupportedOperationException();
+    }
+
     @Override
     public int getCurrentVersion() {
       return this.delegate.getCurrentVersion();
@@ -855,6 +865,16 @@ public void setLargestUsedVersionNumber(int largestUsedVersionNumber) {
     throw new UnsupportedOperationException();
   }
 
+  @Override
+  public int getLargestUsedRTVersionNumber() {
+    return this.delegate.getLargestUsedRTVersionNumber();
+  }
+
+  @Override
+  public void setLargestUsedRTVersionNumber(int largestUsedRTVersionNumber) {
+    throw new UnsupportedOperationException();
+  }
+
   @Override
   public long getStorageQuotaInByte() {
     return this.delegate.getStorageQuotaInByte();

Diff for: internal/venice-common/src/main/java/com/linkedin/venice/meta/Store.java

@@ -80,6 +80,10 @@ static boolean isSystemStore(String storeName) {
 
   void setLargestUsedVersionNumber(int largestUsedVersionNumber);
 
+  int getLargestUsedRTVersionNumber();
+
+  void setLargestUsedRTVersionNumber(int largestUsedRTVersionNumber);
+
   long getStorageQuotaInByte();
 
   void setStorageQuotaInByte(long storageQuotaInByte);

Diff for: internal/venice-common/src/main/java/com/linkedin/venice/meta/StoreGraveyard.java

@@ -12,11 +12,17 @@
  */
 public interface StoreGraveyard {
   /**
-   * Retrieve the largest used version number by the given store name from graveyard. Return 0 if the store dose not
+   * Retrieve the largest used version number by the given store name from graveyard. Return 0 if the store does not
    * exist in the graveyard, which is the default value we used for the new store.
    */
   int getLargestUsedVersionNumber(String storeName);
 
+  /**
+   * Retrieve the largest used version number for the real time topic by the given store name from graveyard.
+   * Return 0 if the store does not exist in the graveyard, which is the default value we used for the new store.
+   */
+  int getLargestUsedRTVersionNumber(String storeName);
+
   /**
    * Put the given store into graveyard. If the store has already existed in the graveyard, update it by this given
    * store.

Diff for: internal/venice-common/src/main/java/com/linkedin/venice/meta/SystemStore.java

@@ -186,6 +186,18 @@ public void setLargestUsedVersionNumber(int largestUsedVersionNumber) {
     systemStoreAttributes.setLargestUsedVersionNumber(largestUsedVersionNumber);
   }
 
+  @Override
+  public int getLargestUsedRTVersionNumber() {
+    SystemStoreAttributes systemStoreAttributes = fetchAndBackfillSystemStoreAttributes(true);
+    return systemStoreAttributes.getLargestUsedRTVersionNumber();
+  }
+
+  @Override
+  public void setLargestUsedRTVersionNumber(int largestUsedRTVersionNumber) {
+    SystemStoreAttributes systemStoreAttributes = fetchAndBackfillSystemStoreAttributes(false);
+    systemStoreAttributes.setLargestUsedRTVersionNumber(largestUsedRTVersionNumber);
+  }
+
   @Override
   public long getStorageQuotaInByte() {
     return zkSharedStore.getStorageQuotaInByte();

Diff for: internal/venice-common/src/main/java/com/linkedin/venice/meta/SystemStoreAttributes.java

@@ -11,6 +11,10 @@ public interface SystemStoreAttributes extends DataModelBackedStructure<SystemSt
 
   void setLargestUsedVersionNumber(int largestUsedVersionNumber);
 
+  int getLargestUsedRTVersionNumber();
+
+  void setLargestUsedRTVersionNumber(int largestUsedRTVersionNumber);
+
   int getCurrentVersion();
 
   void setCurrentVersion(int currentVersion);

Diff for: internal/venice-common/src/main/java/com/linkedin/venice/meta/SystemStoreAttributesImpl.java

@@ -50,6 +50,16 @@ public void setLargestUsedVersionNumber(int largestUsedVersionNumber) {
     this.dataModel.largestUsedVersionNumber = largestUsedVersionNumber;
   }
 
+  @Override
+  public int getLargestUsedRTVersionNumber() {
+    return this.dataModel.largestUsedRTVersionNumber;
+  }
+
+  @Override
+  public void setLargestUsedRTVersionNumber(int largestUsedRTVersionNumber) {
+    this.dataModel.largestUsedRTVersionNumber = largestUsedRTVersionNumber;
+  }
+
   @Override
   public int getCurrentVersion() {
     return this.dataModel.currentVersion;

Diff for: internal/venice-common/src/main/java/com/linkedin/venice/meta/ZKStore.java

@@ -365,6 +365,18 @@ public void setLargestUsedVersionNumber(int largestUsedVersionNumber) {
     this.storeProperties.largestUsedVersionNumber = largestUsedVersionNumber;
   }
 
+  @SuppressWarnings("unused") // Used by Serializer/De-serializer for storing to Zoo Keeper
+  @Override
+  public int getLargestUsedRTVersionNumber() {
+    return this.storeProperties.largestUsedRTVersionNumber;
+  }
+
+  @SuppressWarnings("unused") // Used by Serializer/De-serializer for storing to Zoo Keeper
+  @Override
+  public void setLargestUsedRTVersionNumber(int largestUsedRTVersionNumber) {
+    this.storeProperties.largestUsedRTVersionNumber = largestUsedRTVersionNumber;
+  }
+
   @SuppressWarnings("unused") // Used by Serializer/De-serializer for storing to Zoo Keeper
   @Override
   public long getStorageQuotaInByte() {