[vpj][controller] Fix minor bugs in validation flow for DeferredVersionSwapService (linkedin#2133)

- Use the version config's target region as the target region to validate for validation flow
- Use getControllerClientMap() as this will use d2 map and url map
- Add check for if all child regions are online then mark parent as online to stop vpj polling to account for roll forward controller client timeouts
- Refactor DeferredVersionSwapService to separate out sequential roll forward logic and parallel roll forward logic
- Add vpj config to modify wait time

Author: misyel

clients/venice-push-job/src/main/java/com/linkedin/venice/hadoop/PushJobSetting.java

@@ -78,6 +78,7 @@ public class PushJobSetting implements Serializable {
   public String targetedRegions;
   public boolean isTargetedRegionPushEnabled;
   public boolean isTargetRegionPushWithDeferredSwapEnabled;
+  public int targetRegionPushWithDeferredSwapWaitTime;
   public boolean isSystemSchemaReaderEnabled;
   public boolean isZstdDictCreationRequired;
   public boolean isZstdDictCreationSuccess;

clients/venice-push-job/src/main/java/com/linkedin/venice/hadoop/VenicePushJob.java

@@ -74,6 +74,7 @@
 import static com.linkedin.venice.vpj.VenicePushJobConstants.TARGETED_REGION_PUSH_ENABLED;
 import static com.linkedin.venice.vpj.VenicePushJobConstants.TARGETED_REGION_PUSH_LIST;
 import static com.linkedin.venice.vpj.VenicePushJobConstants.TARGETED_REGION_PUSH_WITH_DEFERRED_SWAP;
+import static com.linkedin.venice.vpj.VenicePushJobConstants.TARGETED_REGION_PUSH_WITH_DEFERRED_SWAP_WAIT_TIME_MINUTES;
 import static com.linkedin.venice.vpj.VenicePushJobConstants.TEMP_DIR_PREFIX;
 import static com.linkedin.venice.vpj.VenicePushJobConstants.UNCREATED_VERSION_NUMBER;
 import static com.linkedin.venice.vpj.VenicePushJobConstants.VALUE_FIELD_PROP;
@@ -96,6 +97,7 @@
 import com.linkedin.venice.controllerapi.RepushInfoResponse;
 import com.linkedin.venice.controllerapi.SchemaResponse;
 import com.linkedin.venice.controllerapi.StoreResponse;
+import com.linkedin.venice.controllerapi.UpdateStoreQueryParams;
 import com.linkedin.venice.controllerapi.VersionCreationResponse;
 import com.linkedin.venice.d2.D2ClientFactory;
 import com.linkedin.venice.etl.ETLValueSchemaTransformation;
@@ -431,6 +433,9 @@ private PushJobSetting getPushJobSetting(VeniceProperties props) {
               + " at the same time");
     }
 
+    pushJobSettingToReturn.targetRegionPushWithDeferredSwapWaitTime =
+        props.getInt(TARGETED_REGION_PUSH_WITH_DEFERRED_SWAP_WAIT_TIME_MINUTES, -1);
+
     if (props.containsKey(TARGETED_REGION_PUSH_LIST)) {
       if (pushJobSettingToReturn.isTargetedRegionPushEnabled
           || pushJobSettingToReturn.isTargetRegionPushWithDeferredSwapEnabled) {
@@ -686,6 +691,13 @@ public void run() {
         initKIFRepushDetails();
       }
 
+      if (pushJobSetting.targetRegionPushWithDeferredSwapWaitTime > -1) {
+        controllerClient.updateStore(
+            pushJobSetting.storeName,
+            new UpdateStoreQueryParams()
+                .setTargetRegionSwapWaitTime(pushJobSetting.targetRegionPushWithDeferredSwapWaitTime));
+      }
+
       setupJobTimeoutMonitor();
       initPushJobDetails();
       logGreeting();

clients/venice-push-job/src/main/java/com/linkedin/venice/vpj/VenicePushJobConstants.java

@@ -393,6 +393,12 @@ private VenicePushJobConstants() {
    */
   public static final String TARGETED_REGION_PUSH_WITH_DEFERRED_SWAP = "targeted.region.push.with.deferred.swap";
 
+  /**
+   * Config to update the wait time in minutes for target region push with deferred version swap
+   */
+  public static final String TARGETED_REGION_PUSH_WITH_DEFERRED_SWAP_WAIT_TIME_MINUTES =
+      "targeted.region.push.with.deferred.swap.wait.time.minutes";
+
   public static final boolean DEFAULT_IS_DUPLICATED_KEY_ALLOWED = false;
 
   /**

internal/venice-test-common/src/integrationTest/java/com/linkedin/venice/endToEnd/TestDeferredVersionSwap.java

@@ -10,6 +10,7 @@
 import static com.linkedin.venice.utils.TestWriteUtils.getTempDataDirectory;
 import static com.linkedin.venice.vpj.VenicePushJobConstants.TARGETED_REGION_PUSH_LIST;
 import static com.linkedin.venice.vpj.VenicePushJobConstants.TARGETED_REGION_PUSH_WITH_DEFERRED_SWAP;
+import static com.linkedin.venice.vpj.VenicePushJobConstants.TARGETED_REGION_PUSH_WITH_DEFERRED_SWAP_WAIT_TIME_MINUTES;
 import static org.testng.Assert.assertEquals;
 import static org.testng.Assert.assertFalse;
 import static org.testng.Assert.assertNotNull;
@@ -142,7 +143,6 @@ public void testDeferredVersionSwap(String targetRegions) throws IOException {
         IntegrationTestPushUtils.defaultVPJProps(multiRegionMultiClusterWrapper, inputDirPath, storeName);
     String keySchemaStr = "\"string\"";
     UpdateStoreQueryParams storeParms = new UpdateStoreQueryParams().setUnusedSchemaDeletionEnabled(true);
-    storeParms.setTargetRegionSwapWaitTime(1);
     String parentControllerURLs = multiRegionMultiClusterWrapper.getControllerConnectString();
 
     try (ControllerClient parentControllerClient = new ControllerClient(CLUSTER_NAMES[0], parentControllerURLs)) {
@@ -151,6 +151,7 @@ public void testDeferredVersionSwap(String targetRegions) throws IOException {
       // Start push job with target region push enabled
       props.put(TARGETED_REGION_PUSH_WITH_DEFERRED_SWAP, true);
       props.put(TARGETED_REGION_PUSH_LIST, targetRegions);
+      props.put(TARGETED_REGION_PUSH_WITH_DEFERRED_SWAP_WAIT_TIME_MINUTES, 1);
       IntegrationTestPushUtils.runVPJ(props);
       TestUtils.waitForNonDeterministicPushCompletion(
           Version.composeKafkaTopic(storeName, 1),