[controller][config][adminTool] Implement automated store-migration tasks and add pause/resume migration APIs to StoreMigrationManage (#1927)

* Implementation of store migration tasks, make the delay (in seconds) configurable in migration scheduler to ensure the integration test completes on time, add an isMigrating check to avoid scheduling duplicate migrations, and introduce a pauseAfterStep argument.

Author: ranwang2024

clients/venice-admin-tool/README

@@ -559,7 +559,7 @@ java -jar venice-admin-tool-all.jar --list-storage-nodes --url <url> --cluster <
 java -jar venice-admin-tool-all.jar --list-store-push-info --url <url> --store <store> [--cluster <cluster>] [--partition-detail-enabled <partition-detail-enabled>]
 java -jar venice-admin-tool-all.jar --list-stores --url <url> --cluster <cluster> [--include-system-stores <include-system-stores>]
 java -jar venice-admin-tool-all.jar --migrate-store --url <url> --store <store> --cluster-src <cluster-src> --cluster-dest <cluster-dest>
-java -jar venice-admin-tool-all.jar --auto-migrate-store --url <url> --store <store> --cluster-src <cluster-src> --cluster-dest <cluster-dest> [--initial-step <initial-step>] [--abort-on-failure <abort-on-failure>]
+java -jar venice-admin-tool-all.jar --auto-migrate-store --url <url> --store <store> --cluster-src <cluster-src> --cluster-dest <cluster-dest> [--initial-step <initial-step>] [--paused-after-step <paused-after-step>] [--abort-on-failure <abort-on-failure>]
 java -jar venice-admin-tool-all.jar --migration-status --url <url> --store <store> --cluster-src <cluster-src> --cluster-dest <cluster-dest>
 java -jar venice-admin-tool-all.jar --monitor-data-recovery --url <url> --dest-fabric <dest-fabric> --datetime <datetime> [--stores <stores>] [--cluster <cluster>] [--interval <interval>]
 java -jar venice-admin-tool-all.jar --new-storage-persona --url <url> --cluster <cluster> --storage-persona <storage-persona> --storage-quota <storage-quota> --store <store> --owner <owner>

clients/venice-admin-tool/src/main/java/com/linkedin/venice/AdminTool.java

@@ -2495,6 +2495,8 @@ public static void autoMigrateStore(CommandLine cmd) {
     Optional<Boolean> abortOnFailure =
         Optional.ofNullable(getOptionalArgument(cmd, Arg.ABORT_ON_FAILURE)).map(Boolean::parseBoolean);
     Optional<Integer> currStep = Optional.ofNullable(getOptionalArgument(cmd, Arg.INITIAL_STEP)).map(Integer::parseInt);
+    Optional<Integer> pauseAfterStep =
+        Optional.ofNullable(getOptionalArgument(cmd, Arg.PAUSE_AFTER_STEP)).map(Integer::parseInt);
 
     if (srcClusterName.equals(destClusterName)) {
       throw new VeniceException("Source and destination cluster cannot be the same!");
@@ -2508,7 +2510,7 @@ public static void autoMigrateStore(CommandLine cmd) {
     assertStoreNotMigrating(srcControllerClient, storeName);
 
     StoreMigrationResponse storeMigrationResponse =
-        srcControllerClient.autoMigrateStore(storeName, destClusterName, currStep, abortOnFailure);
+        srcControllerClient.autoMigrateStore(storeName, destClusterName, currStep, pauseAfterStep, abortOnFailure);
     printObject(storeMigrationResponse);
 
     if (storeMigrationResponse.isError()) {

clients/venice-admin-tool/src/main/java/com/linkedin/venice/Arg.java

@@ -331,6 +331,7 @@ public enum Arg {
       "enum-schema-evolution-allowed", "esea", true, "Allow enum schema evolution for a store"
   ), INITIAL_STEP("initial-step", "is", true, "Initial step of the auto store migration"),
   ABORT_ON_FAILURE("abort-on-failure", "aof", true, "Abort the auto store migration if any step fails"),
+  PAUSE_AFTER_STEP("pause-after-step", "pas", true, "Pause the auto store migration after this step"),
   STORE_LIFECYCLE_HOOKS_LIST("store-lifecycle-hooks-list", "slhl", true, "List of store lifecycle hooks"),
   KEY_URN_COMPRESSION_EANBLED(
       "key-urn-compression-enabled", "kuce", true, "Enable/Disable key urn compression for a store."

clients/venice-admin-tool/src/test/java/com/linkedin/venice/TestAdminTool.java

@@ -228,7 +228,7 @@ public void testAutoMigrateStore() throws ParseException, IOException {
       storeAutoMigrationResponse.setSrcClusterName(srcCluster);
       storeAutoMigrationResponse.setCluster(dstCluster);
       storeAutoMigrationResponse.setName(storeName);
-      when(srcControllerClient.autoMigrateStore(eq(storeName), eq(dstCluster), any(), any()))
+      when(srcControllerClient.autoMigrateStore(eq(storeName), eq(dstCluster), any(), any(), any()))
           .thenReturn(storeAutoMigrationResponse);
 
       StoreResponse storeResponse = new StoreResponse();
@@ -246,9 +246,11 @@ public void testAutoMigrateStore() throws ParseException, IOException {
           .thenReturn(destControllerClient);
 
       AdminTool.autoMigrateStore(fullCmd);
-      Mockito.verify(srcControllerClient).autoMigrateStore(storeName, dstCluster, Optional.of(0), Optional.of(true));
+      Mockito.verify(srcControllerClient)
+          .autoMigrateStore(storeName, dstCluster, Optional.of(0), Optional.empty(), Optional.of(true));
       AdminTool.autoMigrateStore(BasicCmd);
-      Mockito.verify(srcControllerClient).autoMigrateStore(storeName, dstCluster, Optional.empty(), Optional.empty());
+      Mockito.verify(srcControllerClient)
+          .autoMigrateStore(storeName, dstCluster, Optional.empty(), Optional.empty(), Optional.empty());
 
       srcStoreInfo.setMigrating(true);
       storeResponse.setStore(srcStoreInfo);

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

@@ -14,7 +14,8 @@ public class RedundantExceptionFilter {
   private static RedundantExceptionFilter singleton;
 
   private final int bitSetSize;
-  private final ScheduledExecutorService cleanerExecutor = Executors.newScheduledThreadPool(1);
+  private final ScheduledExecutorService cleanerExecutor = Executors.newSingleThreadScheduledExecutor(
+      new DaemonThreadFactory(this.getClass().getName() + "-RedundantExceptionFilterExecutor"));
 
   private BitSet activeBitset;
   private BitSet oldBitSet;

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

@@ -2939,6 +2939,17 @@ private ConfigKeys() {
    */
   public static final String STORE_MIGRATION_MAX_RETRY_ATTEMPTS = "store.migration.max.retry.attempts";
 
+  /**
+   * (Only matters if MULTITASK_SCHEDULER_SERVICE_ENABLED true). Class name of {@link com.linkedin.venice.controller.multitaskscheduler.MultiTaskSchedulerService} implementation
+   */
+  public static final String STORE_MIGRATION_FABRIC_LIST = "store.migration.fabric.list";
+
+  /**
+   * (Only matters if MULTITASK_SCHEDULER_SERVICE_ENABLED true). Class name of {@link com.linkedin.venice.controller.multitaskscheduler.MultiTaskSchedulerService} implementation
+   */
+  public static final String STORE_MIGRATION_TASK_SCHEDULING_INTERVAL_SECONDS =
+      "store.migration.task.scheduling.interval.seconds";
+
   /**
    * The strategy for how to share memory-heavy objects used in the ingestion hot path.
    */

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

@@ -12,6 +12,7 @@ public class ControllerApiConstants {
 
   public static final String AUTO_STORE_MIGRATION_CURRENT_STEP = "auto_store_migration_current_step";
   public static final String AUTO_STORE_MIGRATION_ABORT_ON_FAILURE = "auto_store_migration_abort_on_failure";
+  public static final String AUTO_STORE_MIGRATION_PAUSE_AFTER_STEP = "auto_store_migration_pause_after_step";
   /**
    * @deprecated Use {@link #STORE_NAME} instead.
    */

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

@@ -5,6 +5,7 @@
 import static com.linkedin.venice.controllerapi.ControllerApiConstants.AMPLIFICATION_FACTOR;
 import static com.linkedin.venice.controllerapi.ControllerApiConstants.AUTO_STORE_MIGRATION_ABORT_ON_FAILURE;
 import static com.linkedin.venice.controllerapi.ControllerApiConstants.AUTO_STORE_MIGRATION_CURRENT_STEP;
+import static com.linkedin.venice.controllerapi.ControllerApiConstants.AUTO_STORE_MIGRATION_PAUSE_AFTER_STEP;
 import static com.linkedin.venice.controllerapi.ControllerApiConstants.BATCH_JOB_HEARTBEAT_ENABLED;
 import static com.linkedin.venice.controllerapi.ControllerApiConstants.CLUSTER;
 import static com.linkedin.venice.controllerapi.ControllerApiConstants.CLUSTER_DEST;
@@ -627,9 +628,11 @@ public StoreMigrationResponse autoMigrateStore(
       String storeName,
       String destClusterName,
       Optional<Integer> currStep,
+      Optional<Integer> pauseAfterStep,
       Optional<Boolean> abortOnFailure) {
     QueryParams params = newParams().add(NAME, storeName).add(CLUSTER_DEST, destClusterName);
     currStep.ifPresent(cs -> params.add(AUTO_STORE_MIGRATION_CURRENT_STEP, cs));
+    pauseAfterStep.ifPresent(pas -> params.add(AUTO_STORE_MIGRATION_PAUSE_AFTER_STEP, pas));
     abortOnFailure.ifPresent(aof -> params.add(AUTO_STORE_MIGRATION_ABORT_ON_FAILURE, aof));
     params.add(AUTO_STORE_MIGRATION_ABORT_ON_FAILURE, abortOnFailure);
     return request(ControllerRoute.AUTO_MIGRATE_STORE, params, StoreMigrationResponse.class);