[common][compat] Add previousCurrentVersion field to track version swap history (linkedin#2433)

Author: misyel

build.gradle

@@ -327,8 +327,6 @@ subprojects {
       // when actually using the new protocol. Example to pin KME to v12 when introducing v13:
       // project(':internal:venice-common').file('src/main/resources/avro/KafkaMessageEnvelope/v12', PathValidation.DIRECTORY)
       def versionOverrides = [
-          project(':internal:venice-common').file('src/main/resources/avro/StoreMetaValue/v39', PathValidation.DIRECTORY),
-          project(':services:venice-controller').file('src/main/resources/avro/AdminOperation/v94', PathValidation.DIRECTORY)
       ]
 
       def schemaDirs = [sourceDir]

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

@@ -281,6 +281,7 @@ public class ControllerApiConstants {
   public static final String STORE_LIFECYCLE_HOOKS_LIST = "store_lifecycle_hooks_list";
   public static final String KEY_URN_COMPRESSION_ENABLED = "key_urn_compression_enabled";
   public static final String KEY_URN_FIELDS = "key_urn_fields";
+  public static final String PREVIOUS_CURRENT_VERSION = "previous_current_version";
 
   /**
    * Params for repush job

internal/venice-common/src/main/java/com/linkedin/venice/controllerapi/UpdateStoreQueryParams.java

@@ -54,6 +54,7 @@
 import static com.linkedin.venice.controllerapi.ControllerApiConstants.PARTITIONER_PARAMS;
 import static com.linkedin.venice.controllerapi.ControllerApiConstants.PARTITION_COUNT;
 import static com.linkedin.venice.controllerapi.ControllerApiConstants.PERSONA_NAME;
+import static com.linkedin.venice.controllerapi.ControllerApiConstants.PREVIOUS_CURRENT_VERSION;
 import static com.linkedin.venice.controllerapi.ControllerApiConstants.PUSH_STREAM_SOURCE_ADDRESS;
 import static com.linkedin.venice.controllerapi.ControllerApiConstants.READ_COMPUTATION_ENABLED;
 import static com.linkedin.venice.controllerapi.ControllerApiConstants.READ_QUOTA_IN_CU;
@@ -918,6 +919,14 @@ public Optional<List<String>> getKeyUrnFields() {
     return getStringList(KEY_URN_FIELDS);
   }
 
+  public UpdateStoreQueryParams setPreviousCurrentVersion(int previousCurrentVersion) {
+    return putInteger(PREVIOUS_CURRENT_VERSION, previousCurrentVersion);
+  }
+
+  public Optional<Integer> getPreviousCurrentVersion() {
+    return getInteger(PREVIOUS_CURRENT_VERSION);
+  }
+
   // ***************** above this line are getters and setters *****************
   private UpdateStoreQueryParams putInteger(String name, int value) {
     return (UpdateStoreQueryParams) add(name, value);

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

@@ -709,6 +709,16 @@ public void setRmdVersionId(int replicationMetadataVersionId) {
       throw new UnsupportedOperationException();
     }
 
+    @Override
+    public int getPreviousCurrentVersion() {
+      return this.delegate.getPreviousCurrentVersion();
+    }
+
+    @Override
+    public void setPreviousCurrentVersion(int previousCurrentVersion) {
+      throw new UnsupportedOperationException();
+    }
+
     @Override
     public Version cloneVersion() {
       return this.delegate.cloneVersion();
@@ -1045,6 +1055,9 @@ public StoreProperties cloneStoreProperties() {
     storeProperties.setStoreLifecycleHooks(convertStoreLifecycleHooks(getStoreLifecycleHooks()));
     storeProperties.setKeyUrnCompressionEnabled(isKeyUrnCompressionEnabled());
     storeProperties.setKeyUrnFields(getKeyUrnFields().stream().map(String::toString).collect(Collectors.toList()));
+    storeProperties.setPreviousCurrentVersion(getPreviousCurrentVersion());
+    // Set blobDbEnabled to default value - field exists in schema but not yet exposed via Store interface
+    storeProperties.setBlobDbEnabled("NOT_SPECIFIED");
 
     return storeProperties;
   }
@@ -1765,6 +1778,16 @@ public List<String> getKeyUrnFields() {
     return delegate.getKeyUrnFields();
   }
 
+  @Override
+  public int getPreviousCurrentVersion() {
+    return this.delegate.getPreviousCurrentVersion();
+  }
+
+  @Override
+  public void setPreviousCurrentVersion(int previousCurrentVersion) {
+    throw new UnsupportedOperationException();
+  }
+
   @Override
   public String toString() {
     return this.delegate.toString();
@@ -1920,6 +1943,9 @@ private static StoreVersion convertVersion(Version version) {
     storeVersion.setKeyUrnCompressionEnabled(version.isKeyUrnCompressionEnabled());
     storeVersion.setKeyUrnFields(version.getKeyUrnFields().stream().map(String::toString).collect(Collectors.toList()));
     storeVersion.setRepushTtlSeconds(version.getRepushTtlSeconds());
+    storeVersion.setPreviousCurrentVersion(version.getPreviousCurrentVersion());
+    // Set blobDbEnabled to default value - field exists in schema but not yet exposed via Version interface
+    storeVersion.setBlobDbEnabled("NOT_SPECIFIED");
 
     return storeVersion;
   }

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

@@ -385,4 +385,8 @@ default IntSet getVersionNumbers() {
   void setKeyUrnFields(List<String> keyUrnFields);
 
   List<String> getKeyUrnFields();
+
+  int getPreviousCurrentVersion();
+
+  void setPreviousCurrentVersion(int previousCurrentVersion);
 }

internal/venice-common/src/main/java/com/linkedin/venice/meta/StoreInfo.java

@@ -94,6 +94,7 @@ public static StoreInfo fromStore(Store store) {
     storeInfo.setKeyUrnCompressionEnabled(store.isKeyUrnCompressionEnabled());
     storeInfo.setKeyUrnFields(store.getKeyUrnFields());
     storeInfo.setFlinkVeniceViewsEnabled(store.isFlinkVeniceViewsEnabled());
+    storeInfo.setPreviousCurrentVersion(store.getPreviousCurrentVersion());
     return storeInfo;
   }
 
@@ -378,6 +379,7 @@ public static StoreInfo fromStore(Store store) {
   private boolean keyUrnCompressionEnabled = false;
   private List<String> keyUrnFields = new ArrayList<>();
   private boolean flinkVeniceViewsEnabled = false;
+  private int previousCurrentVersion = -1;
 
   public StoreInfo() {
   }
@@ -1019,4 +1021,12 @@ public List<String> getKeyUrnFields() {
   public void setKeyUrnFields(List<String> keyUrnFields) {
     this.keyUrnFields = keyUrnFields;
   }
+
+  public int getPreviousCurrentVersion() {
+    return previousCurrentVersion;
+  }
+
+  public void setPreviousCurrentVersion(int previousCurrentVersion) {
+    this.previousCurrentVersion = previousCurrentVersion;
+  }
 }

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

@@ -837,6 +837,16 @@ public List<String> getKeyUrnFields() {
     return Collections.emptyList();
   }
 
+  @Override
+  public int getPreviousCurrentVersion() {
+    return zkSharedStore.getPreviousCurrentVersion();
+  }
+
+  @Override
+  public void setPreviousCurrentVersion(int previousCurrentVersion) {
+    throwUnsupportedOperationException("setPreviousCurrentVersion");
+  }
+
   @Override
   public boolean isGlobalRtDivEnabled() {
     return zkSharedStore.isGlobalRtDivEnabled();

internal/venice-common/src/main/java/com/linkedin/venice/meta/Version.java

@@ -314,6 +314,10 @@ default void setTimestampMetadataVersionId(int replicationMetadataVersionId) {
 
   List<String> getKeyUrnFields();
 
+  int getPreviousCurrentVersion();
+
+  void setPreviousCurrentVersion(int previousCurrentVersion);
+
   /**
    * Kafka topic name is composed by store name and version.
    * <p>

internal/venice-common/src/main/java/com/linkedin/venice/meta/VersionImpl.java

@@ -480,6 +480,16 @@ public List<String> getKeyUrnFields() {
     return this.storeVersion.keyUrnFields.stream().map(Objects::toString).collect(Collectors.toList());
   }
 
+  @Override
+  public int getPreviousCurrentVersion() {
+    return this.storeVersion.previousCurrentVersion;
+  }
+
+  @Override
+  public void setPreviousCurrentVersion(int previousCurrentVersion) {
+    this.storeVersion.previousCurrentVersion = previousCurrentVersion;
+  }
+
   @Override
   public StoreVersion dataModel() {
     return this.storeVersion;
@@ -568,6 +578,7 @@ public Version cloneVersion() {
     clonedVersion.setKeyUrnCompressionEnabled(isKeyUrnCompressionEnabled());
     clonedVersion.setKeyUrnFields(getKeyUrnFields());
     clonedVersion.setRepushTtlSeconds(getRepushTtlSeconds());
+    clonedVersion.setPreviousCurrentVersion(getPreviousCurrentVersion());
     return clonedVersion;
   }
 

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

@@ -251,6 +251,7 @@ public ZKStore(Store store) {
     setKeyUrnCompressionEnabled(store.isKeyUrnCompressionEnabled());
     setKeyUrnFields(store.getKeyUrnFields());
     setFlinkVeniceViewsEnabled(store.isFlinkVeniceViewsEnabled());
+    setPreviousCurrentVersion(store.getPreviousCurrentVersion());
 
     for (Version storeVersion: store.getVersions()) {
       forceAddVersion(storeVersion.cloneVersion(), true);
@@ -308,13 +309,35 @@ public int getCurrentVersion() {
   /**
    * Set current serving version number of this store. If store is disabled to write, thrown {@link
    * StoreDisabledException}.
+   *
+   * When a new version is promoted to current, this method also sets the previousCurrentVersion
+   * field on the NEW current version to track which version was current before.
    */
   @Override
   public void setCurrentVersion(int currentVersion) {
     checkDisableStoreWrite("setStoreCurrentVersion", currentVersion);
     // Update the latest version promotion to current timestamp, which is useful for backup version retention.
     setLatestVersionPromoteToCurrentTimestamp(System.currentTimeMillis());
+
+    // Capture old current version before updating
+    int oldCurrentVersion = getCurrentVersion();
+
     setCurrentVersionWithoutCheck(currentVersion);
+
+    // Set previousCurrentVersion on the NEW current version
+    if (oldCurrentVersion != Store.NON_EXISTING_VERSION && currentVersion != Store.NON_EXISTING_VERSION) {
+      updateVersionPreviousCurrentVersion(currentVersion, oldCurrentVersion);
+    }
+  }
+
+  /** Updates the previousCurrentVersion field on a version. */
+  private void updateVersionPreviousCurrentVersion(int versionNumber, int previousCurrentVersion) {
+    for (StoreVersion storeVersion: storeProperties.versions) {
+      if (storeVersion.number == versionNumber) {
+        storeVersion.previousCurrentVersion = previousCurrentVersion;
+        return;
+      }
+    }
   }
 
   @SuppressWarnings("unused") // Used by Serializer/De-serializer for storing to ZooKeeper
@@ -1105,6 +1128,16 @@ public List<String> getKeyUrnFields() {
     return this.storeProperties.keyUrnFields.stream().map(Objects::toString).collect(Collectors.toList());
   }
 
+  @Override
+  public int getPreviousCurrentVersion() {
+    return this.storeProperties.previousCurrentVersion;
+  }
+
+  @Override
+  public void setPreviousCurrentVersion(int previousCurrentVersion) {
+    this.storeProperties.previousCurrentVersion = previousCurrentVersion;
+  }
+
   @Override
   public boolean isGlobalRtDivEnabled() {
     return this.storeProperties.globalRtDivEnabled;