[test][controller] Fix multi-region tests to use VeniceTwoLayerMultiR…

…egionMultiClusterWrapper (#1067)

There are several tests that create multi-region clusters through manual, non-standard setups. An example of this is that many tests create a multi-region cluster that shares the same Kafka cluster between them. While this might be valuable for some tests, it is a maintenance overhead for most of them. Our test suite already has "VeniceTwoLayerMultiRegionMultiClusterWrapper" that can be used directly for all current use-cases.

This commit replaces all tests that setup a multi-region test cluster through non-standard means to use "VeniceTwoLayerMultiRegionMultiClusterWrapper" instead.

This commit also fixes an issue where if the config to enable active active by default for batch-only stores is enabled, new store creation for system stores makes them also active-active. This was masked previously by different configs set on child and parent controllers. So, running this config in parent controller would have been problematic.

This change exposed another bug that storage persona only runs validations when running in a multi-region setups while there is nothing conceptually blocking it from running in a single-region setup. This will be addressed in a follow up PR.

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

@@ -3,108 +3,101 @@
 import static com.linkedin.venice.ConfigKeys.ENABLE_NATIVE_REPLICATION_AS_DEFAULT_FOR_BATCH_ONLY;
 import static com.linkedin.venice.ConfigKeys.NATIVE_REPLICATION_SOURCE_FABRIC_AS_DEFAULT_FOR_BATCH_ONLY_STORES;
 import static com.linkedin.venice.ConfigKeys.NATIVE_REPLICATION_SOURCE_FABRIC_AS_DEFAULT_FOR_HYBRID_STORES;
-import static com.linkedin.venice.controller.VeniceHelixAdmin.VERSION_ID_UNSET;
-import static org.mockito.ArgumentMatchers.any;
-import static org.mockito.ArgumentMatchers.anyString;
-import static org.mockito.Mockito.doReturn;
-import static org.mockito.Mockito.mock;
+import static com.linkedin.venice.utils.TestUtils.assertCommand;
+import static org.testng.Assert.assertEquals;
+import static org.testng.Assert.assertTrue;
 
+import com.linkedin.venice.controllerapi.ControllerClient;
 import com.linkedin.venice.controllerapi.UpdateStoreQueryParams;
-import com.linkedin.venice.meta.Version;
-import com.linkedin.venice.pubsub.manager.TopicManager;
-import com.linkedin.venice.pubsub.manager.TopicManagerRepository;
+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.StoreInfo;
+import com.linkedin.venice.utils.TestUtils;
+import com.linkedin.venice.utils.Time;
 import com.linkedin.venice.utils.Utils;
-import java.io.IOException;
-import java.util.Optional;
 import java.util.Properties;
+import java.util.concurrent.TimeUnit;
 import org.testng.Assert;
 import org.testng.annotations.AfterClass;
 import org.testng.annotations.BeforeClass;
 import org.testng.annotations.Test;
 
 
-public class TestClusterLevelConfigForNativeReplication extends AbstractTestVeniceHelixAdmin {
+public class TestClusterLevelConfigForNativeReplication {
+  private static final long TEST_TIMEOUT = 60 * Time.MS_PER_SECOND;
+
+  private VeniceTwoLayerMultiRegionMultiClusterWrapper multiRegionMultiClusterWrapper;
+  private ControllerClient parentControllerClient;
+
   @BeforeClass(alwaysRun = true)
-  public void setUp() throws Exception {
-    setupCluster();
+  public void setUp() {
+    Utils.thisIsLocalhost();
+    Properties parentControllerProps = new Properties();
+    // enable native replication for batch-only stores through cluster-level config
+    parentControllerProps.setProperty(ENABLE_NATIVE_REPLICATION_AS_DEFAULT_FOR_BATCH_ONLY, "true");
+    parentControllerProps.setProperty(NATIVE_REPLICATION_SOURCE_FABRIC_AS_DEFAULT_FOR_BATCH_ONLY_STORES, "dc-batch");
+    parentControllerProps.setProperty(NATIVE_REPLICATION_SOURCE_FABRIC_AS_DEFAULT_FOR_HYBRID_STORES, "dc-hybrid");
+    multiRegionMultiClusterWrapper = ServiceFactory.getVeniceTwoLayerMultiRegionMultiClusterWrapper(
+        new VeniceMultiRegionClusterCreateOptions.Builder().numberOfRegions(1)
+            .numberOfParentControllers(1)
+            .numberOfChildControllers(1)
+            .numberOfRouters(1)
+            .numberOfServers(1)
+            .parentControllerProperties(parentControllerProps)
+            .build());
+    parentControllerClient = new ControllerClient(
+        multiRegionMultiClusterWrapper.getClusterNames()[0],
+        multiRegionMultiClusterWrapper.getControllerConnectString());
   }
 
   @AfterClass(alwaysRun = true)
-  public void cleanUp() {
-    cleanupCluster();
-  }
-
-  @Override
-  Properties getControllerProperties(String clusterName) throws IOException {
-    Properties props = super.getControllerProperties(clusterName);
-    // enable native replication for batch-only stores through cluster-level config
-    props.setProperty(ENABLE_NATIVE_REPLICATION_AS_DEFAULT_FOR_BATCH_ONLY, "true");
-    props.setProperty(NATIVE_REPLICATION_SOURCE_FABRIC_AS_DEFAULT_FOR_BATCH_ONLY_STORES, "dc-batch");
-    props.setProperty(NATIVE_REPLICATION_SOURCE_FABRIC_AS_DEFAULT_FOR_HYBRID_STORES, "dc-hybrid");
-    return props;
+  public void tearDown() {
+    Utils.closeQuietlyWithErrorLogged(parentControllerClient);
+    Utils.closeQuietlyWithErrorLogged(multiRegionMultiClusterWrapper);
   }
 
-  @Test
+  @Test(timeOut = TEST_TIMEOUT)
   public void testClusterLevelNativeReplicationConfigForNewStores() {
-    TopicManagerRepository originalTopicManagerRepository = veniceAdmin.getTopicManagerRepository();
-
-    TopicManager mockedTopicManager = mock(TopicManager.class);
-    TopicManagerRepository mockedTopicManageRepository = mock(TopicManagerRepository.class);
-    doReturn(mockedTopicManager).when(mockedTopicManageRepository).getLocalTopicManager();
-    doReturn(mockedTopicManager).when(mockedTopicManageRepository).getTopicManager(any(String.class));
-    doReturn(mockedTopicManager).when(mockedTopicManageRepository).getTopicManager(anyString());
-    veniceAdmin.setTopicManagerRepository(mockedTopicManageRepository);
     String storeName = Utils.getUniqueString("test-store");
     String pushJobId1 = "test-push-job-id-1";
-    /**
-     * Do not enable any store-level config for leader/follower mode or native replication feature.
-     */
-    veniceAdmin.createStore(clusterName, storeName, "test-owner", KEY_SCHEMA, VALUE_SCHEMA);
+    parentControllerClient.createNewStore(storeName, "test-owner", "\"string\"", "\"string\"");
+    parentControllerClient.emptyPush(storeName, pushJobId1, 1);
 
-    /**
-     * Add a version
-     */
-    veniceAdmin.addVersionAndTopicOnly(
-        clusterName,
-        storeName,
-        pushJobId1,
-        VERSION_ID_UNSET,
-        1,
-        1,
-        false,
-        true,
-        Version.PushType.BATCH,
-        null,
-        null,
-        Optional.empty(),
-        -1,
-        1,
-        Optional.empty(),
-        false);
     // Version 1 should exist.
-    Assert.assertEquals(veniceAdmin.getStore(clusterName, storeName).getVersions().size(), 1);
+    StoreInfo store = assertCommand(parentControllerClient.getStore(storeName)).getStore();
+    assertEquals(store.getVersions().size(), 1);
     // native replication should be enabled by cluster-level config
-    Assert.assertEquals(veniceAdmin.getStore(clusterName, storeName).isNativeReplicationEnabled(), true);
-    Assert.assertEquals(veniceAdmin.getStore(clusterName, storeName).getNativeReplicationSourceFabric(), "dc-batch");
-    veniceAdmin.updateStore(
-        clusterName,
-        storeName,
-        new UpdateStoreQueryParams().setHybridRewindSeconds(1L).setHybridOffsetLagThreshold(1L));
-    Assert.assertEquals(veniceAdmin.getStore(clusterName, storeName).getNativeReplicationSourceFabric(), "dc-hybrid");
-    veniceAdmin.updateStore(
-        clusterName,
-        storeName,
-        new UpdateStoreQueryParams().setHybridRewindSeconds(-1L).setHybridOffsetLagThreshold(-1L));
-    Assert.assertEquals(veniceAdmin.getStore(clusterName, storeName).getNativeReplicationSourceFabric(), "dc-batch");
-    veniceAdmin.updateStore(
-        clusterName,
-        storeName,
-        new UpdateStoreQueryParams().setIncrementalPushEnabled(true)
-            .setHybridRewindSeconds(1L)
-            .setHybridOffsetLagThreshold(10));
-    Assert.assertEquals(veniceAdmin.getStore(clusterName, storeName).getNativeReplicationSourceFabric(), "dc-hybrid");
-
-    // Set topic original topic manager back
-    veniceAdmin.setTopicManagerRepository(originalTopicManagerRepository);
+    assertTrue(store.isNativeReplicationEnabled());
+    assertEquals(store.getNativeReplicationSourceFabric(), "dc-batch");
+    assertCommand(
+        parentControllerClient.updateStore(
+            storeName,
+            new UpdateStoreQueryParams().setHybridRewindSeconds(1L).setHybridOffsetLagThreshold(1L)));
+    TestUtils.waitForNonDeterministicAssertion(TEST_TIMEOUT, TimeUnit.MILLISECONDS, () -> {
+      Assert.assertEquals(
+          parentControllerClient.getStore(storeName).getStore().getNativeReplicationSourceFabric(),
+          "dc-hybrid");
+    });
+    assertCommand(
+        parentControllerClient.updateStore(
+            storeName,
+            new UpdateStoreQueryParams().setHybridRewindSeconds(-1L).setHybridOffsetLagThreshold(-1L)));
+    TestUtils.waitForNonDeterministicAssertion(TEST_TIMEOUT, TimeUnit.MILLISECONDS, () -> {
+      Assert.assertEquals(
+          parentControllerClient.getStore(storeName).getStore().getNativeReplicationSourceFabric(),
+          "dc-batch");
+    });
+    assertCommand(
+        parentControllerClient.updateStore(
+            storeName,
+            new UpdateStoreQueryParams().setIncrementalPushEnabled(true)
+                .setHybridRewindSeconds(1L)
+                .setHybridOffsetLagThreshold(10)));
+    TestUtils.waitForNonDeterministicAssertion(TEST_TIMEOUT, TimeUnit.MILLISECONDS, () -> {
+      Assert.assertEquals(
+          parentControllerClient.getStore(storeName).getStore().getNativeReplicationSourceFabric(),
+          "dc-hybrid");
+    });
   }
 }

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

@@ -10,8 +10,6 @@
 import static com.linkedin.venice.controller.SchemaConstants.VALUE_SCHEMA_FOR_WRITE_COMPUTE_V3;
 import static com.linkedin.venice.controller.SchemaConstants.VALUE_SCHEMA_FOR_WRITE_COMPUTE_V4;
 import static com.linkedin.venice.controller.SchemaConstants.VALUE_SCHEMA_FOR_WRITE_COMPUTE_V5;
-import static com.linkedin.venice.integration.utils.VeniceClusterWrapperConstants.CHILD_REGION_NAME_PREFIX;
-import static com.linkedin.venice.integration.utils.VeniceClusterWrapperConstants.DEFAULT_PARENT_DATA_CENTER_REGION_NAME;
 import static com.linkedin.venice.utils.TestUtils.assertCommand;
 import static com.linkedin.venice.utils.TestUtils.waitForNonDeterministicAssertion;
 import static com.linkedin.venice.utils.TestUtils.waitForNonDeterministicPushCompletion;
@@ -32,14 +30,11 @@
 import com.linkedin.venice.controllerapi.StoreResponse;
 import com.linkedin.venice.controllerapi.UpdateStoreQueryParams;
 import com.linkedin.venice.controllerapi.VersionCreationResponse;
-import com.linkedin.venice.integration.utils.PubSubBrokerConfigs;
-import com.linkedin.venice.integration.utils.PubSubBrokerWrapper;
 import com.linkedin.venice.integration.utils.ServiceFactory;
 import com.linkedin.venice.integration.utils.VeniceClusterWrapper;
-import com.linkedin.venice.integration.utils.VeniceControllerCreateOptions;
 import com.linkedin.venice.integration.utils.VeniceControllerWrapper;
+import com.linkedin.venice.integration.utils.VeniceMultiRegionClusterCreateOptions;
 import com.linkedin.venice.integration.utils.VeniceTwoLayerMultiRegionMultiClusterWrapper;
-import com.linkedin.venice.integration.utils.ZkServerWrapper;
 import com.linkedin.venice.meta.ETLStoreConfig;
 import com.linkedin.venice.meta.HybridStoreConfig;
 import com.linkedin.venice.meta.StoreInfo;
@@ -560,40 +555,33 @@ public static Object[][] controllerSSLAndSupersetSchemaGenerator() {
   public void testStoreMetaDataUpdateFromParentToChildController(
       boolean isControllerSslEnabled,
       boolean isSupersetSchemaGeneratorEnabled) throws IOException {
-    Properties properties = new Properties();
+    Properties parentControllerProps = new Properties();
     // This cluster setup don't have server, we cannot perform push here.
-    properties.setProperty(CONTROLLER_AUTO_MATERIALIZE_META_SYSTEM_STORE, String.valueOf(false));
-    properties.setProperty(CONTROLLER_AUTO_MATERIALIZE_DAVINCI_PUSH_STATUS_SYSTEM_STORE, String.valueOf(false));
+    parentControllerProps.setProperty(CONTROLLER_AUTO_MATERIALIZE_META_SYSTEM_STORE, String.valueOf(false));
+    parentControllerProps
+        .setProperty(CONTROLLER_AUTO_MATERIALIZE_DAVINCI_PUSH_STATUS_SYSTEM_STORE, String.valueOf(false));
     if (isSupersetSchemaGeneratorEnabled) {
-      properties.setProperty(CONTROLLER_PARENT_EXTERNAL_SUPERSET_SCHEMA_GENERATION_ENABLED, String.valueOf(true));
-      properties.put(
+      parentControllerProps
+          .setProperty(CONTROLLER_PARENT_EXTERNAL_SUPERSET_SCHEMA_GENERATION_ENABLED, String.valueOf(true));
+      parentControllerProps.put(
           VeniceControllerWrapper.SUPERSET_SCHEMA_GENERATOR,
           new SupersetSchemaGeneratorWithCustomProp("test_prop"));
     }
 
-    try (ZkServerWrapper zkServer = ServiceFactory.getZkServer();
-        PubSubBrokerWrapper pubSubBrokerWrapper = ServiceFactory.getPubSubBroker(
-            new PubSubBrokerConfigs.Builder().setZkWrapper(zkServer)
-                .setRegionName(DEFAULT_PARENT_DATA_CENTER_REGION_NAME)
-                .build());
-        VeniceControllerWrapper childControllerWrapper = ServiceFactory.getVeniceController(
-            new VeniceControllerCreateOptions.Builder(clusterName, zkServer, pubSubBrokerWrapper)
-                .sslToKafka(isControllerSslEnabled)
-                .regionName(CHILD_REGION_NAME_PREFIX + "0")
-                .build());
-        ZkServerWrapper parentZk = ServiceFactory.getZkServer();
-        VeniceControllerWrapper parentControllerWrapper = ServiceFactory.getVeniceController(
-            new VeniceControllerCreateOptions.Builder(clusterName, parentZk, pubSubBrokerWrapper)
-                .childControllers(new VeniceControllerWrapper[] { childControllerWrapper })
-                .extraProperties(properties)
+    try (VeniceTwoLayerMultiRegionMultiClusterWrapper venice =
+        ServiceFactory.getVeniceTwoLayerMultiRegionMultiClusterWrapper(
+            new VeniceMultiRegionClusterCreateOptions.Builder().numberOfRegions(1)
+                .numberOfClusters(1)
+                .numberOfParentControllers(1)
+                .numberOfChildControllers(1)
+                .numberOfServers(0)
+                .numberOfRouters(0)
+                .replicationFactor(1)
+                .parentControllerProperties(parentControllerProps)
                 .sslToKafka(isControllerSslEnabled)
                 .build())) {
-      String childControllerUrl = isControllerSslEnabled
-          ? childControllerWrapper.getSecureControllerUrl()
-          : childControllerWrapper.getControllerUrl();
-      String parentControllerUrl = isControllerSslEnabled
-          ? parentControllerWrapper.getSecureControllerUrl()
-          : parentControllerWrapper.getControllerUrl();
+      String childControllerUrl = venice.getChildRegions().get(0).getControllerConnectString();
+      String parentControllerUrl = venice.getControllerConnectString();
       Optional<SSLFactory> sslFactory =
           isControllerSslEnabled ? Optional.of(SslUtils.getVeniceLocalSslFactory()) : Optional.empty();
       try (ControllerClient parentControllerClient = new ControllerClient(clusterName, parentControllerUrl, sslFactory);