[controller] Extract VersionLifecyclePolicy from VeniceHelixAdmin / VeniceParentHelixAdmin (linkedin#2827)

[controller] Extract VersionLifecyclePolicy from VeniceHelixAdmin / VeniceParentHelixAdmin                                                                                                                                                                                                                       
                                                                                                                                                                                                                                                                                                                   
  VeniceHelixAdmin is a ~10k-line god class. Carve a new com.linkedin.venice.controller.versionlifecycle sub-package and land its first helper, VersionLifecyclePolicy — a final pure-statics class (no I/O against controller state, all dependencies passed in) that owns the lifecycle decisions previously     
  buried in the admins. Mirrors HybridStoreConfigPolicy's shape and seeds the planned per-concern decomposition (TopicLifecycleManager, VersionRetirementManager, VersionProvisioner, CurrentVersionManager — named in package-info.java).                                                                         
                                                                                                                                                                                                                                                                                                                   
  Lifted, with call-site rewires in VeniceHelixAdmin, VeniceParentHelixAdmin, and DataRecoveryManager: checkBackupVersionCleanupCapacityForNewPush, checkRollbackOriginVersionCapacityForNewPush, getBackupVersionNumber, getFinalReturnStatus, getOverallPushStatus, hasFatalDataValidationError,                 
  updateStoreTTLRepushFlag, isRealTimeTopicRequired (with isParent() inlined as a boolean arg), checkPreConditionForSingleVersionDeletion (4-arg pure variant; the 3-arg admin wrapper that does the I/O lookup is retained for its two callers), and the STATUS_PRIORITIES constant (package-private to match its 
  original visibility and satisfy spotbugs MS_MUTABLE_COLLECTION_PKGPROTECT). Method bodies are byte-identical to the originals; admin call sites inline the (config, time, isParent) arguments the deleted instance wrappers used to resolve. The stale admin-side javadoc on isRealTimeTopicRequired — which     
  described parent-region conditions the body never actually checked — is replaced with one that matches the code.                                                                                                                                                                                                 
                                                                                                                                                                                                                                                                                                                   
  Tests: consolidate the two standalone capacity tests (TestBackupVersionCleanupCapacityCheck, TestRollbackOriginVersionCapacityCheck) plus in-admin coverage for getFinalReturnStatus / getOverallPushStatus / updateStoreTTLRepushFlag / isRealTimeTopicRequired into a new versionlifecycle/ test package,      
  organized along source-class section boundaries so they travel with their owners on the planned per-helper extractions: NewPushCapacityGuardsTest (19), PushStatusAggregationTest (11), LifecyclePredicatesTest (17), VersionSelectionTest (4) = 51 tests. Shared mock factories and constants live in           
  package-private VersionLifecycleTestSupport. The two existing integration tests (TestPushBlocked{ByRollbackOriginGuard,WithinMinCleanupDelay}) only had their javadoc {@link}s retargeted.                                                                                                                       
                                                                                                                                                                                                                                                                                                                   
  Net: VeniceHelixAdmin -300, VeniceParentHelixAdmin -59, new policy +311, new test files +917, deleted in-admin/standalone tests -662. No protocol or wire-format change; getBackupVersionNumber was not on the Admin interface, so the lift is internal-only.                                                    
                                                                       

Co-authored-by: Ali Poursamadi <apoursamadi@linkedin.com>
Co-authored-by: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

Author: 3 people

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

@@ -433,7 +433,7 @@ private PushJobSetting getPushJobSetting(VeniceProperties props) {
     }
 
     // Compliance push and TTL repush settings are mutually exclusive because the controller uses push ID prefix
-    // to manage TTL settings. See VeniceHelixAdmin#updateStoreTTLRepushFlag for details.
+    // to manage TTL settings. See VersionLifecyclePolicy#updateStoreTTLRepushFlag for details.
     if (pushJobSettingToReturn.isCompliancePush
         && (pushJobSettingToReturn.repushTTLEnabled || pushJobSettingToReturn.allowRegularPushWithTTLRepush)) {
       throw new VeniceException("Compliance push cannot be combined with TTL repush settings");

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

@@ -36,7 +36,7 @@
 
 
 /**
- * Integration test for {@link com.linkedin.venice.controller.VeniceHelixAdmin#checkRollbackOriginVersionCapacityForNewPush}.
+ * Integration test for {@link com.linkedin.venice.controller.versionlifecycle.VersionLifecyclePolicy#checkRollbackOriginVersionCapacityForNewPush}.
  *
  * <p>After a rollback, v2 becomes ROLLED_BACK (on the parent, once propagated) and v1 becomes
  * current. The test exercises both sides of the rollback-origin retention window:

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

@@ -36,7 +36,7 @@
 
 
 /**
- * Integration test for {@link com.linkedin.venice.controller.VeniceHelixAdmin#checkBackupVersionCleanupCapacityForNewPush}.
+ * Integration test for {@link com.linkedin.venice.controller.versionlifecycle.VersionLifecyclePolicy#checkBackupVersionCleanupCapacityForNewPush}.
  *
  * <p>The capacity guard rejects a new push when one or more backup versions are in a deletable
  * state (e.g., {@code KILLED} / {@code ERROR}) and the store is still within the configured

services/venice-controller/src/main/java/com/linkedin/venice/controller/VeniceHelixAdmin.java

@@ -163,6 +163,7 @@
 import com.linkedin.venice.controller.supersetschema.DefaultSupersetSchemaGenerator;
 import com.linkedin.venice.controller.supersetschema.SupersetSchemaGenerator;
 import com.linkedin.venice.controller.util.ParentControllerConfigUpdateUtils;
+import com.linkedin.venice.controller.versionlifecycle.VersionLifecyclePolicy;
 import com.linkedin.venice.controllerapi.AdminCommandExecution;
 import com.linkedin.venice.controllerapi.ControllerClient;
 import com.linkedin.venice.controllerapi.ControllerResponse;
@@ -1885,7 +1886,7 @@ public Version incrementVersionIdempotent(
       // the child via VeniceHelixAdmin.addVersion; running it here surfaces the rejection to the push
       // job synchronously instead of only failing the child admin-consumption asynchronously.
       if (VeniceSystemStoreType.getSystemStoreType(storeName) == null) {
-        VeniceHelixAdmin.checkRollbackOriginVersionCapacityForNewPush(
+        VersionLifecyclePolicy.checkRollbackOriginVersionCapacityForNewPush(
             clusterName,
             storeName,
             store,
@@ -2745,7 +2746,7 @@ private void updateParentVersionStatusAfterRollback(
       // checkRollbackOriginVersionCapacityForNewPush (which requires rollback-origin versions to have
       // number > currentVersion so stale ROLLED_BACK entries lingering in parent metadata age out).
       int backupVersionNum =
-          getVeniceHelixAdmin().getBackupVersionNumber(store.getVersions(), store.getCurrentVersion());
+          VersionLifecyclePolicy.getBackupVersionNumber(store.getVersions(), store.getCurrentVersion());
       if (backupVersionNum != NON_EXISTING_VERSION) {
         store.setCurrentVersion(backupVersionNum);
       } else {
@@ -4505,8 +4506,8 @@ private OfflinePushStatusInfo getOffLineJobStatus(
 
     StringBuilder currentReturnStatusDetails = new StringBuilder();
 
-    ExecutionStatus currentReturnStatus =
-        getFinalReturnStatus(statuses, childRegions, numChildRegionsFailedToFetchStatus, currentReturnStatusDetails);
+    ExecutionStatus currentReturnStatus = VersionLifecyclePolicy
+        .getFinalReturnStatus(statuses, childRegions, numChildRegionsFailedToFetchStatus, currentReturnStatusDetails);
 
     String storeName = Version.parseStoreFromKafkaTopicName(kafkaTopic);
     int versionNum = Version.parseVersionFromKafkaTopicName(kafkaTopic);
@@ -4710,56 +4711,6 @@ private void handleTerminalJobStatus(
     }
   }
 
-  /**
-   * Based on the global information, start determining the final status to return
-   * @param statuses
-   * @param childRegions
-   * @param numChildRegionsFailedToFetchStatus
-   * @param currentReturnStatusDetails
-   * @return
-   */
-  protected static ExecutionStatus getFinalReturnStatus(
-      Map<String, ExecutionStatus> statuses,
-      Set<String> childRegions,
-      int numChildRegionsFailedToFetchStatus,
-      StringBuilder currentReturnStatusDetails) {
-    ExecutionStatus currentReturnStatus = ExecutionStatus.NEW;
-
-    // Sort the per-datacenter status in this order, and return the first one in the list
-    // Edge case example: if one cluster is stuck in NOT_CREATED, then
-    // as another cluster goes from PROGRESS to COMPLETED
-    // the aggregate status will go from PROGRESS back down to NOT_CREATED.
-    List<ExecutionStatus> sortedStatuses = statuses.values()
-        .stream()
-        .sorted(Comparator.comparingInt(VeniceHelixAdmin.STATUS_PRIORITIES::indexOf))
-        .collect(Collectors.toList());
-
-    if (!sortedStatuses.isEmpty()) {
-      currentReturnStatus = sortedStatuses.get(0);
-    }
-
-    int successCount = childRegions.size() - numChildRegionsFailedToFetchStatus;
-    if (successCount < (childRegions.size() / 2) + 1) {
-      // Strict majority must be reachable, otherwise keep polling
-      currentReturnStatus = ExecutionStatus.PROGRESS;
-    }
-
-    if (currentReturnStatus.isTerminal()) {
-      // If there is a temporary datacenter connection failure, we want VPJ to report failure while allowing the push
-      // to succeed in remaining datacenters. If we want to allow the push to succeed in async in the remaining
-      // datacenter, then put the topic delete into an else block under `if (numChildRegionsFailedToFetchStatus > 0)`
-      if (numChildRegionsFailedToFetchStatus > 0) {
-        currentReturnStatus = ExecutionStatus.ERROR;
-        currentReturnStatusDetails.append(numChildRegionsFailedToFetchStatus)
-            .append("/")
-            .append(childRegions.size())
-            .append(" DCs unreachable. ");
-      }
-    }
-
-    return currentReturnStatus;
-  }
-
   /**
    * Based on the control configs and push information to decide whether to truncate the Kafka topic or not.
    * @param clusterName

services/venice-controller/src/main/java/com/linkedin/venice/controller/VeniceParentHelixAdmin.java

@@ -3,6 +3,7 @@
 import com.linkedin.venice.controller.Admin;
 import com.linkedin.venice.controller.ParticipantStoreClientsManager;
 import com.linkedin.venice.controller.VeniceHelixAdmin;
+import com.linkedin.venice.controller.versionlifecycle.VersionLifecyclePolicy;
 import com.linkedin.venice.exceptions.VeniceException;
 import com.linkedin.venice.exceptions.VeniceNoStoreException;
 import com.linkedin.venice.meta.DataRecoveryVersionConfigImpl;
@@ -135,7 +136,7 @@ public void prepareStoreVersionForDataRecovery(
        * We need to set the store's current version to the backup version or {@link Store#NON_EXISTING_VERSION} to
        * perform data recovery on the current version.
        */
-      int backupVersion = veniceAdmin.getBackupVersionNumber(store.getVersions(), store.getCurrentVersion());
+      int backupVersion = VersionLifecyclePolicy.getBackupVersionNumber(store.getVersions(), store.getCurrentVersion());
       veniceAdmin.setStoreCurrentVersion(clusterName, storeName, backupVersion);
       veniceAdmin.wipeCluster(clusterName, destinationFabric, Optional.of(storeName), Optional.of(versionNumber));
     } else {