[controller] Harden update-store workflow

clients/venice-push-job/src/test/java/com/linkedin/venice/hadoop/VenicePushJobTest.java

@@ -440,7 +440,6 @@ private StoreInfo getStoreInfo(Consumer<StoreInfo> info, boolean applyFirst) {
     storeInfo.setChunkingEnabled(false);
     storeInfo.setCompressionStrategy(CompressionStrategy.NO_OP);
     storeInfo.setWriteComputationEnabled(false);
-    storeInfo.setIncrementalPushEnabled(false);
     storeInfo.setNativeReplicationSourceFabric("dc-0");
     Map<String, Integer> coloMaps = new HashMap<String, Integer>() {
       {

internal/venice-common/src/main/java/com/linkedin/venice/ConfigKeys.java

@@ -375,8 +375,16 @@ private ConfigKeys() {
       "controller.store.graveyard.cleanup.sleep.interval.between.list.fetch.minutes";
 
   /**
-   * Whether the superset schema generation in Parent Controller should be done via passed callback or not.
+   * Whether the superset schema generation in Primary Controller should be done via passed callback or not.
    */
+  public static final String CONTROLLER_EXTERNAL_SUPERSET_SCHEMA_GENERATION_ENABLED =
+      "controller.external.superset.schema.generation.enabled";
+
+  /**
+   * Whether the superset schema generation in Primary Controller should be done via passed callback or not.
+   * @deprecated Use {@link #CONTROLLER_EXTERNAL_SUPERSET_SCHEMA_GENERATION_ENABLED}
+   */
+  @Deprecated
   public static final String CONTROLLER_PARENT_EXTERNAL_SUPERSET_SCHEMA_GENERATION_ENABLED =
       "controller.parent.external.superset.schema.generation.enabled";
 

internal/venice-common/src/main/java/com/linkedin/venice/helix/StoragePersonaRepository.java

@@ -179,15 +179,15 @@ private void deleteStores(List<String> storeNames) {
 
   public StoragePersona getPersonaContainingStore(String storeName) {
     String personaName = storeNamePersonaMap.get(storeName);
-    if (personaName == null)
+    if (personaName == null) {
       return null;
+    }
     return getPersona(personaName);
   }
 
   private boolean isStoreSetValid(StoragePersona persona, Optional<Store> additionalStore) {
     Set<String> setToValidate = new HashSet<>();
-    if (additionalStore.isPresent())
-      setToValidate.add(additionalStore.get().getName());
+    additionalStore.ifPresent(store -> setToValidate.add(store.getName()));
     setToValidate.addAll(persona.getStoresToEnforce());
     return setToValidate.stream()
         .allMatch(

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

@@ -24,7 +24,7 @@ public enum BackupStrategy {
   // KEEP_IN_KAFKA_ONLY,
   /** Keep in user-specified store eg HDD, other DB */
   // KEEP_IN_USER_STORE;
-  private int value;
+  private final int value;
 
   BackupStrategy(int v) {
     this.value = v;
@@ -35,6 +35,10 @@ public enum BackupStrategy {
     Arrays.stream(values()).forEach(s -> idMapping.put(s.value, s));
   }
 
+  public int getValue() {
+    return value;
+  }
+
   public static BackupStrategy fromInt(int i) {
     BackupStrategy strategy = idMapping.get(i);
     if (strategy == null) {

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

@@ -27,4 +27,9 @@ public interface HybridStoreConfig extends DataModelBackedStructure<StoreHybridC
   BufferReplayPolicy getBufferReplayPolicy();
 
   HybridStoreConfig clone();
+
+  default boolean isHybrid() {
+    return getRewindTimeInSeconds() >= 0
+        && (getOffsetLagThresholdToGoOnline() >= 0 || getProducerTimestampLagThresholdToGoOnlineInSeconds() >= 0);
+  }
 }

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

@@ -92,6 +92,9 @@ public int hashCode() {
 
   @JsonIgnore
   public PartitionerConfig clone() {
-    return new PartitionerConfigImpl(getPartitionerClass(), getPartitionerParams(), getAmplificationFactor());
+    return new PartitionerConfigImpl(
+        getPartitionerClass(),
+        new HashMap<>(getPartitionerParams()),
+        getAmplificationFactor());
   }
 }

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

@@ -209,7 +209,7 @@ public ZKStore(Store store) {
     setSchemaAutoRegisterFromPushJobEnabled(store.isSchemaAutoRegisterFromPushJobEnabled());
     setLatestSuperSetValueSchemaId(store.getLatestSuperSetValueSchemaId());
     setHybridStoreDiskQuotaEnabled(store.isHybridStoreDiskQuotaEnabled());
-    setEtlStoreConfig(store.getEtlStoreConfig());
+    setEtlStoreConfig(store.getEtlStoreConfig().clone());
     setStoreMetadataSystemStoreEnabled(store.isStoreMetadataSystemStoreEnabled());
     setLatestVersionPromoteToCurrentTimestamp(store.getLatestVersionPromoteToCurrentTimestamp());
     setBackupVersionRetentionMs(store.getBackupVersionRetentionMs());
@@ -220,7 +220,7 @@ public ZKStore(Store store) {
     setStoreMetaSystemStoreEnabled(store.isStoreMetaSystemStoreEnabled());
     setActiveActiveReplicationEnabled(store.isActiveActiveReplicationEnabled());
     setRmdVersion(store.getRmdVersion());
-    setViewConfigs(store.getViewConfigs());
+    setViewConfigs(new HashMap<>(store.getViewConfigs()));
     setStorageNodeReadQuotaEnabled(store.isStorageNodeReadQuotaEnabled());
     setUnusedSchemaDeletionEnabled(store.isUnusedSchemaDeletionEnabled());
     setMinCompactionLagSeconds(store.getMinCompactionLagSeconds());
@@ -365,11 +365,7 @@ public void setLargestUsedVersionNumber(int largestUsedVersionNumber) {
   @SuppressWarnings("unused") // Used by Serializer/De-serializer for storing to Zoo Keeper
   @Override
   public long getStorageQuotaInByte() {
-    // This is a safeguard in case that some old stores do not have storage quota field
-    return (this.storeProperties.storageQuotaInByte <= 0
-        && this.storeProperties.storageQuotaInByte != UNLIMITED_STORAGE_QUOTA)
-            ? DEFAULT_STORAGE_QUOTA
-            : this.storeProperties.storageQuotaInByte;
+    return this.storeProperties.storageQuotaInByte;
   }
 
   @SuppressWarnings("unused") // Used by Serializer/De-serializer for storing to Zoo Keeper

internal/venice-common/src/main/java/com/linkedin/venice/utils/PartitionUtils.java

@@ -63,12 +63,16 @@ public static int calculatePartitionCount(
     } else if (partitionCount < minPartitionCount) {
       partitionCount = minPartitionCount;
     }
+
+    int returnPartitionCount = partitionCount <= 0 ? 1 : (int) partitionCount;
+
     LOGGER.info(
         "Assign partition count: {} calculated by storage quota: {} to the new version of store: {}",
-        partitionCount,
+        returnPartitionCount,
         storageQuota,
         storeName);
-    return (int) partitionCount;
+
+    return returnPartitionCount;
   }
 
   public static VenicePartitioner getVenicePartitioner(PartitionerConfig config) {

internal/venice-common/src/test/java/com/linkedin/venice/meta/BackupStrategyTest.java

@@ -0,0 +1,23 @@
+package com.linkedin.venice.meta;
+
+import static org.testng.Assert.assertEquals;
+import static org.testng.Assert.assertThrows;
+
+import com.linkedin.venice.exceptions.VeniceException;
+import org.testng.annotations.Test;
+
+
+public class BackupStrategyTest {
+  @Test
+  public void testFromInt() {
+    assertEquals(BackupStrategy.fromInt(0), BackupStrategy.KEEP_MIN_VERSIONS);
+    assertEquals(BackupStrategy.fromInt(1), BackupStrategy.DELETE_ON_NEW_PUSH_START);
+    assertThrows(VeniceException.class, () -> BackupStrategy.fromInt(2));
+  }
+
+  @Test
+  public void testGetValue() {
+    assertEquals(BackupStrategy.KEEP_MIN_VERSIONS.getValue(), 0);
+    assertEquals(BackupStrategy.DELETE_ON_NEW_PUSH_START.getValue(), 1);
+  }
+}

internal/venice-common/src/test/java/com/linkedin/venice/meta/TestHybridStoreConfig.java

@@ -22,4 +22,26 @@ public void deserializes() throws IOException {
     Assert.assertEquals(fasterXml.getRewindTimeInSeconds(), 123L);
     Assert.assertEquals(fasterXml.getDataReplicationPolicy(), DataReplicationPolicy.NON_AGGREGATE);
   }
+
+  @Test
+  public void testIsHybrid() {
+    HybridStoreConfig hybridStoreConfig;
+    hybridStoreConfig = new HybridStoreConfigImpl(-1, -1, -1, null, null);
+    Assert.assertFalse(hybridStoreConfig.isHybrid());
+
+    hybridStoreConfig = new HybridStoreConfigImpl(100, -1, -1, null, null);
+    Assert.assertFalse(hybridStoreConfig.isHybrid());
+
+    hybridStoreConfig = new HybridStoreConfigImpl(100, 100, -1, null, null);
+    Assert.assertTrue(hybridStoreConfig.isHybrid());
+
+    hybridStoreConfig = new HybridStoreConfigImpl(100, 100, 100, null, null);
+    Assert.assertTrue(hybridStoreConfig.isHybrid());
+
+    hybridStoreConfig = new HybridStoreConfigImpl(100, -1, 100, null, null);
+    Assert.assertTrue(hybridStoreConfig.isHybrid());
+
+    hybridStoreConfig = new HybridStoreConfigImpl(-1, -1, 100, null, null);
+    Assert.assertFalse(hybridStoreConfig.isHybrid());
+  }
 }

internal/venice-test-common/src/integrationTest/java/com/linkedin/venice/controller/TestClusterLevelConfigForActiveActiveReplication.java

@@ -98,4 +98,40 @@ public void testClusterLevelActiveActiveReplicationConfigForNewHybridStores() th
       assertFalse(parentControllerClient.getStore(storeName).getStore().isActiveActiveReplicationEnabled());
     });
   }
+
+  @Test(timeOut = TEST_TIMEOUT)
+  public void testClusterLevelActiveActiveReplicationConfigForNewIncrementalPushStores() throws IOException {
+    String storeName = Utils.getUniqueString("test-store-incremental");
+    String pushJobId1 = "test-push-job-id-1";
+    parentControllerClient.createNewStore(storeName, "test-owner", "\"string\"", "\"string\"");
+    parentControllerClient.emptyPush(storeName, pushJobId1, 1);
+
+    // Version 1 should exist.
+    StoreInfo store = assertCommand(parentControllerClient.getStore(storeName)).getStore();
+    assertEquals(store.getVersions().size(), 1);
+
+    // Check store level Active/Active is enabled or not
+    assertFalse(store.isActiveActiveReplicationEnabled());
+    assertFalse(store.isIncrementalPushEnabled());
+    assertFalse(store.isActiveActiveReplicationEnabled());
+
+    // Convert to incremental push store
+    assertCommand(
+        parentControllerClient.updateStore(storeName, new UpdateStoreQueryParams().setIncrementalPushEnabled(true)));
+    TestUtils.waitForNonDeterministicAssertion(TEST_TIMEOUT, TimeUnit.MILLISECONDS, () -> {
+      StoreInfo storeToTest = parentControllerClient.getStore(storeName).getStore();
+      assertTrue(storeToTest.isIncrementalPushEnabled());
+      assertTrue(storeToTest.isActiveActiveReplicationEnabled());
+    });
+
+    // After inc push is disabled, even default A/A config for pure hybrid store is false,
+    // original store A/A config is enabled.
+    assertCommand(
+        parentControllerClient.updateStore(storeName, new UpdateStoreQueryParams().setIncrementalPushEnabled(false)));
+    TestUtils.waitForNonDeterministicAssertion(TEST_TIMEOUT, TimeUnit.MILLISECONDS, () -> {
+      StoreInfo storeToTest = parentControllerClient.getStore(storeName).getStore();
+      assertFalse(storeToTest.isIncrementalPushEnabled());
+      assertTrue(storeToTest.isActiveActiveReplicationEnabled());
+    });
+  }
 }

internal/venice-test-common/src/integrationTest/java/com/linkedin/venice/controller/TestClusterLevelConfigForNativeReplication.java

@@ -13,6 +13,7 @@
 import com.linkedin.venice.integration.utils.ServiceFactory;
 import com.linkedin.venice.integration.utils.VeniceMultiRegionClusterCreateOptions;
 import com.linkedin.venice.integration.utils.VeniceTwoLayerMultiRegionMultiClusterWrapper;
+import com.linkedin.venice.meta.DataReplicationPolicy;
 import com.linkedin.venice.meta.StoreInfo;
 import com.linkedin.venice.meta.Version;
 import com.linkedin.venice.utils.TestUtils;
@@ -97,6 +98,7 @@ public void testClusterLevelNativeReplicationConfigForNewStores() {
         parentControllerClient.updateStore(
             storeName,
             new UpdateStoreQueryParams().setIncrementalPushEnabled(true)
+                .setHybridDataReplicationPolicy(DataReplicationPolicy.NONE)
                 .setHybridRewindSeconds(1L)
                 .setHybridOffsetLagThreshold(10)));
     TestUtils.waitForNonDeterministicAssertion(TEST_TIMEOUT, TimeUnit.MILLISECONDS, () -> {
@@ -107,7 +109,7 @@ public void testClusterLevelNativeReplicationConfigForNewStores() {
   }
 
   @Test(timeOut = TEST_TIMEOUT)
-  public void testConvertHybridDuringPushjob() {
+  public void testConvertHybridDuringPushJob() {
     String storeName = Utils.getUniqueString("test-store");
     parentControllerClient.createNewStore(storeName, "test-owner", "\"string\"", "\"string\"");
     parentControllerClient.requestTopicForWrites(
@@ -128,7 +130,7 @@ public void testConvertHybridDuringPushjob() {
         storeName,
         new UpdateStoreQueryParams().setHybridRewindSeconds(1L).setHybridOffsetLagThreshold(1L));
     Assert.assertTrue(response.isError());
-    Assert.assertTrue(response.getError().contains("Cannot convert to hybrid as there is already a pushjob running"));
+    Assert.assertTrue(response.getError().contains("Cannot convert to hybrid as there is already a push job running"));
     parentControllerClient.killOfflinePushJob(Version.composeKafkaTopic(storeName, 1));
   }
 

internal/venice-test-common/src/integrationTest/java/com/linkedin/venice/controller/TestParentControllerWithMultiDataCenter.java

@@ -470,8 +470,10 @@ public void testEnableActiveActiveReplicationSchema() {
       Assert.assertFalse(schemaResponse2.isError(), "addValeSchema returned error: " + schemaResponse2.getError());
 
       // Enable AA on store
-      UpdateStoreQueryParams updateStoreToEnableAARepl =
-          new UpdateStoreQueryParams().setNativeReplicationEnabled(true).setActiveActiveReplicationEnabled(true);
+      UpdateStoreQueryParams updateStoreToEnableAARepl = new UpdateStoreQueryParams().setNativeReplicationEnabled(true)
+          .setActiveActiveReplicationEnabled(true)
+          .setHybridOffsetLagThreshold(1000)
+          .setHybridRewindSeconds(1000);
       TestWriteUtils.updateStore(storeName, parentControllerClient, updateStoreToEnableAARepl);
       /**
        * Test Active/Active replication config enablement generates the active active metadata schema.