Remove MARK_VERSION_ROLLED_BACK admin operation

Replace cross-region protocol message with direct deleteOldVersion calls per region via controllerClientMap. Non-current PUSHED/ONLINE child copies of a terminal deferred version are deleted immediately rather than transitioned through ROLLED_BACK; the double-guard in VeniceHelixAdmin and controllerClientMap ensures current versions are never touched.

Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com>

Author: KaiSernLim

docs/operations/data-management/version-lifecycle.md

@@ -8,9 +8,9 @@ backup, within configured retention, or part of a store migration.
 ## Deletion decision table
 
 The count-based sweep is implemented by `Store.retrieveVersionsToDelete`. The time-based sweep is
-implemented by `StoreBackupVersionCleanupService`. Deferred version swap terminal transitions first
-reconcile bootstrap-complete, non-current child copies to `ROLLED_BACK`, so they cannot remain
-invisible to both sweeps.
+implemented by `StoreBackupVersionCleanupService`. Deferred version swap terminal transitions
+directly delete bootstrap-complete, non-current child copies via `ControllerClient.deleteOldVersion`,
+so they cannot remain invisible to both sweeps.
 
 | Initial status | Role / initial condition | Count retention | Time / safety gate | Trigger | Deletion decision |
 | --- | --- | --- | --- | --- | --- |
@@ -26,7 +26,7 @@ invisible to both sweeps.
 | `PUSHED` | Non-current version without a terminal parent decision | Excluded because it may be an active deferred or concurrent swap candidate | Not eligible | Count or time cleanup | **KEEP** |
 | `PUSHED` or `ONLINE` | Still current in a child after parent `ERROR` or `ROLLED_BACK` | Protected while serving; reconciliation remains incomplete | Not eligible | Parent terminal reconciliation | **DEFER: keep and retry** |
 | `PUSHED` or `ONLINE` | Still current in a child after parent `PARTIALLY_ONLINE` | Serving the intentional partial state | Not eligible | Parent terminal reconciliation | **KEEP** |
-| `PUSHED` or non-current `ONLINE` | Parent deferred swap becomes `ERROR`, `PARTIALLY_ONLINE`, or `ROLLED_BACK` | Removed from count sweep | Rolled-back retention gate applies | Parent terminal transition | **DEFER: mark `ROLLED_BACK`** |
+| `PUSHED` or non-current `ONLINE` | Parent deferred swap becomes `ERROR`, `PARTIALLY_ONLINE`, or `ROLLED_BACK` | Removed from count sweep | None — immediately deleted | Parent terminal transition | **DELETE** |
 | `ONLINE` | Backup within the configured preserved count | Within limit | Not considered | Count sweep | **KEEP** |
 | `ONLINE` | Backup beyond the configured preserved count | Exceeds limit | Not considered | Count sweep | **DELETE** |
 | `ONLINE` | Backup considered by retention cleanup | Not considered | Before minimum retention | Time sweep | **DEFER** |
@@ -39,15 +39,11 @@ invisible to both sweeps.
 Count-based and time-based cleanup are independent triggers. The first applicable trigger may delete
 an eligible backup, but neither trigger may delete the current version. `PUSHED` versions are never
 deleted from status and version number alone because multiple deferred or concurrent swaps can be
-active. A terminal parent decision explicitly converts bootstrap-complete non-current child copies
-to `ROLLED_BACK`. The controller scans every deferred terminal parent version, including versions
-superseded by a newer push, and retries while a child is unreachable, missing the target metadata,
-in progress, or unexpectedly still current after parent `ERROR` or `ROLLED_BACK`.
-
-When a child copy becomes `ROLLED_BACK`, the controller resets the store-level latest-promotion
-timestamp to start the rollback retention window. A subsequent promotion can reset that shared clock
-again, so a rolled-back version may be retained longer than the configured duration; it cannot be
-deleted immediately because the current version was promoted long before the rollback.
+active. A terminal parent decision explicitly deletes bootstrap-complete non-current child copies
+via per-region `ControllerClient.deleteOldVersion`. The controller scans every deferred terminal
+parent version, including versions superseded by a newer push, and retries while a child is
+unreachable, missing the target metadata, in progress, or unexpectedly still current after parent
+`ERROR` or `ROLLED_BACK`.
 
 ## State machine
 
@@ -60,7 +56,7 @@ stateDiagram-v2
     STARTED --> ERROR: push fails
     STARTED --> KILLED: push is killed
     PUSHED --> ONLINE: version swap succeeds
-    PUSHED --> ROLLED_BACK: deferred swap terminates while non-current
+    PUSHED --> DELETED: deferred swap terminates while non-current (parent terminal sweep)
     ONLINE --> ROLLED_BACK: rollback or abandoned non-current copy
 
     NOT_CREATED --> DELETED: stale below-current metadata after time gate
@@ -81,7 +77,7 @@ stateDiagram-v2
       Before terminal parent decision: KEEP
       Current after ERROR/ROLLED_BACK: retry
       Current after PARTIALLY_ONLINE: KEEP
-      Parent terminal + non-current: ROLLED_BACK
+      Parent terminal + non-current: DELETE immediately
     end note
 
     note right of ONLINE

internal/venice-common/src/main/java/com/linkedin/venice/serialization/avro/AvroProtocolDefinition.java

@@ -71,7 +71,7 @@ public enum AvroProtocolDefinition {
    *
    * TODO: Move AdminOperation to venice-common module so that we can properly reference it here.
    */
-  ADMIN_OPERATION(102, SpecificData.get().getSchema(ByteBuffer.class), "AdminOperation"),
+  ADMIN_OPERATION(101, SpecificData.get().getSchema(ByteBuffer.class), "AdminOperation"),
 
   /**
    * Single chunk of a large multi-chunk value. Just a bunch of bytes.

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

@@ -1418,14 +1418,26 @@ private boolean reconcileAbandonedVersionInChildRegions(
     }
 
     if (!regionsToReconcile.isEmpty()) {
-      String regionsFilter = RegionUtils.composeRegionList(regionsToReconcile);
-      veniceParentHelixAdmin.markVersionRolledBack(clusterName, storeName, targetVersionNum, regionsFilter);
-      LOGGER.info(
-          "Reconciled version: {} for store: {} as ROLLED_BACK in child regions: {} after parent status: {}",
-          targetVersionNum,
-          storeName,
-          regionsFilter,
-          parentStatus);
+      for (String region: regionsToReconcile) {
+        try {
+          controllerClientMap.get(region).deleteOldVersion(storeName, targetVersionNum);
+          LOGGER.info(
+              "Deleted abandoned deferred version {} for store {} in region {} (parent status: {})",
+              targetVersionNum,
+              storeName,
+              region,
+              parentStatus);
+        } catch (Exception e) {
+          LOGGER.warn(
+              "Failed to delete abandoned deferred version {} for store {} in region {} (parent status: {})",
+              targetVersionNum,
+              storeName,
+              region,
+              parentStatus,
+              e);
+          allRegionsTerminal = false;
+        }
+      }
     }
     return allRegionsTerminal;
   }

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

@@ -5124,76 +5124,6 @@ public void rollbackToBackupVersion(String clusterName, String storeName, String
     }
   }
 
-  /**
-   * Mark {@code versionNumber} as {@link VersionStatus#ROLLED_BACK} in the regions selected by
-   * {@code regionFilter}, WITHOUT changing the store's current version.
-   *
-   * <p>Deferred swaps can end globally while a bootstrap-complete copy is still non-current in some
-   * regions. Marking that copy ROLLED_BACK makes it eligible for time-based cleanup. In-progress
-   * versions and the current serving version remain untouched.
-   */
-  public void markVersionRolledBack(String clusterName, String storeName, int versionNumber, String regionFilter) {
-    String currentRegion = getRegionName();
-    if (!StringUtils.isEmpty(regionFilter)) {
-      Set<String> regionsFilter = parseRegionsFilterList(regionFilter);
-      if (!regionsFilter.contains(currentRegion)) {
-        LOGGER.info(
-            "markVersionRolledBack will be skipped for store: {} version: {} in cluster: {}, because the region filter"
-                + " is {} which doesn't include the current region: {}",
-            storeName,
-            versionNumber,
-            clusterName,
-            regionsFilter,
-            currentRegion);
-        return;
-      }
-    }
-
-    storeMetadataUpdate(clusterName, storeName, (store, resources) -> {
-      if (!store.containsVersion(versionNumber)) {
-        LOGGER.info(
-            "markVersionRolledBack skipped: version {} not found in store {} in cluster {}",
-            versionNumber,
-            storeName,
-            clusterName);
-        return store;
-      }
-      // Never override the current version's status; that version serves reads.
-      if (store.getCurrentVersion() == versionNumber) {
-        LOGGER.warn(
-            "markVersionRolledBack skipped: version {} is the current version of store {} in cluster {}",
-            versionNumber,
-            storeName,
-            clusterName);
-        return store;
-      }
-      VersionStatus currentStatus = store.getVersionStatus(versionNumber);
-      if (VersionStatus.isVersionRolledBack(currentStatus) || VersionStatus.canDelete(currentStatus)) {
-        // Already ROLLED_BACK, or already in a terminal cleanable status (KILLED/ERROR) — nothing to do.
-        return store;
-      }
-      if (!VersionStatus.isBootstrapCompleted(currentStatus)) {
-        LOGGER.info(
-            "markVersionRolledBack skipped: version {} of store {} in cluster {} has not completed bootstrap (status {})",
-            versionNumber,
-            storeName,
-            clusterName,
-            currentStatus);
-        return store;
-      }
-      LOGGER.info(
-          "Marking version {} of store {} in cluster {} as ROLLED_BACK (was {})",
-          versionNumber,
-          storeName,
-          clusterName,
-          currentStatus);
-      store.updateVersionStatus(versionNumber, ROLLED_BACK);
-      store.setLatestVersionPromoteToCurrentTimestamp(
-          Math.max(store.getLatestVersionPromoteToCurrentTimestamp(), System.currentTimeMillis()));
-      return store;
-    });
-  }
-
   /**
    * Update the largest used version number of a specified store.
    */

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

@@ -63,7 +63,6 @@
 import com.linkedin.venice.controller.kafka.protocol.admin.PushStatusSystemStoreAutoCreationValidation;
 import com.linkedin.venice.controller.kafka.protocol.admin.ResumeStore;
 import com.linkedin.venice.controller.kafka.protocol.admin.RollbackCurrentVersion;
-import com.linkedin.venice.controller.kafka.protocol.admin.MarkVersionRolledBack;
 import com.linkedin.venice.controller.kafka.protocol.admin.SchemaMeta;
 import com.linkedin.venice.controller.kafka.protocol.admin.SetStoreOwner;
 import com.linkedin.venice.controller.kafka.protocol.admin.SetStorePartitionCount;
@@ -2520,34 +2519,6 @@ void checkNewPushCapacityFromChildren(String clusterName, String storeName) {
     }
   }
 
-  /**
-   * Mark {@code versionNum} as ROLLED_BACK in the child regions selected by {@code regionFilter},
-   * WITHOUT changing the current version. Used to reconcile the non-target regions after a deferred
-   * version swap rollback: those regions bootstrapped the target version but never swapped to it, so
-   * their copy is stranded in PUSHED status above their unchanged current version and leaks disk.
-   * Marking it ROLLED_BACK makes it visible to {@code StoreBackupVersionCleanupService}.
-   */
-  public void markVersionRolledBack(String clusterName, String storeName, int versionNum, String regionFilter) {
-    acquireAdminMessageLock(clusterName, storeName);
-    try {
-      getVeniceHelixAdmin().checkPreConditionForUpdateStoreMetadata(clusterName, storeName);
-
-      MarkVersionRolledBack markVersionRolledBack =
-          (MarkVersionRolledBack) AdminMessageType.MARK_VERSION_ROLLED_BACK.getNewInstance();
-      markVersionRolledBack.clusterName = clusterName;
-      markVersionRolledBack.storeName = storeName;
-      markVersionRolledBack.versionNum = versionNum;
-      markVersionRolledBack.regionsFilter = regionFilter;
-      AdminOperation message = new AdminOperation();
-      message.operationType = AdminMessageType.MARK_VERSION_ROLLED_BACK.getValue();
-      message.payloadUnion = markVersionRolledBack;
-
-      sendAdminMessageAndWaitForConsumed(clusterName, storeName, message);
-    } finally {
-      releaseAdminMessageLock(clusterName, storeName);
-    }
-  }
-
   private void updateParentVersionStatusAfterRollback(
       String clusterName,
       String storeName,

services/venice-controller/src/main/java/com/linkedin/venice/controller/kafka/consumer/AdminExecutionTask.java

@@ -30,7 +30,6 @@
 import com.linkedin.venice.controller.kafka.protocol.admin.ResumeStore;
 import com.linkedin.venice.controller.kafka.protocol.admin.RollForwardCurrentVersion;
 import com.linkedin.venice.controller.kafka.protocol.admin.RollbackCurrentVersion;
-import com.linkedin.venice.controller.kafka.protocol.admin.MarkVersionRolledBack;
 import com.linkedin.venice.controller.kafka.protocol.admin.SetStoreCurrentVersion;
 import com.linkedin.venice.controller.kafka.protocol.admin.SetStoreOwner;
 import com.linkedin.venice.controller.kafka.protocol.admin.SetStorePartitionCount;
@@ -345,9 +344,6 @@ private void processMessage(AdminOperation adminOperation) {
         case ROLLFORWARD_CURRENT_VERSION:
           handleRollForwardToFutureVersion((RollForwardCurrentVersion) adminOperation.payloadUnion);
           break;
-        case MARK_VERSION_ROLLED_BACK:
-          handleMarkVersionRolledBack((MarkVersionRolledBack) adminOperation.payloadUnion);
-          break;
         default:
           throw new VeniceException("Unknown admin operation type: " + adminOperation.operationType);
       }
@@ -939,15 +935,6 @@ private void handleRollbackCurrentVersion(RollbackCurrentVersion message) {
     admin.rollbackToBackupVersion(clusterName, storeName, regionFilter);
   }
 
-  private void handleMarkVersionRolledBack(MarkVersionRolledBack message) {
-    String clusterName = message.getClusterName().toString();
-    String storeName = message.getStoreName().toString();
-    int versionNum = message.getVersionNum();
-    CharSequence regionsFilter = message.getRegionsFilter();
-    String regionFilter = regionsFilter == null ? null : regionsFilter.toString();
-    admin.markVersionRolledBack(clusterName, storeName, versionNum, regionFilter);
-  }
-
   private void handleDeleteUnusedValueSchema(DeleteUnusedValueSchemas message) {
     String clusterName = message.getClusterName().toString();
     String storeName = message.getStoreName().toString();

services/venice-controller/src/main/java/com/linkedin/venice/controller/kafka/protocol/enums/AdminMessageType.java

@@ -15,7 +15,6 @@
 import com.linkedin.venice.controller.kafka.protocol.admin.DisableStoreRead;
 import com.linkedin.venice.controller.kafka.protocol.admin.EnableStoreRead;
 import com.linkedin.venice.controller.kafka.protocol.admin.KillOfflinePushJob;
-import com.linkedin.venice.controller.kafka.protocol.admin.MarkVersionRolledBack;
 import com.linkedin.venice.controller.kafka.protocol.admin.MetaSystemStoreAutoCreationValidation;
 import com.linkedin.venice.controller.kafka.protocol.admin.MetadataSchemaCreation;
 import com.linkedin.venice.controller.kafka.protocol.admin.MigrateStore;
@@ -66,8 +65,7 @@ public enum AdminMessageType implements VeniceDimensionInterface {
   CONFIGURE_INCREMENTAL_PUSH_FOR_CLUSTER(22, true), META_SYSTEM_STORE_AUTO_CREATION_VALIDATION(23, false),
   PUSH_STATUS_SYSTEM_STORE_AUTO_CREATION_VALIDATION(24, false), CREATE_STORAGE_PERSONA(25, false),
   DELETE_STORAGE_PERSONA(26, false), UPDATE_STORAGE_PERSONA(27, false), DELETE_UNUSED_VALUE_SCHEMA(28, false),
-  ROLLBACK_CURRENT_VERSION(29, false), ROLLFORWARD_CURRENT_VERSION(30, false),
-  MARK_VERSION_ROLLED_BACK(31, false);
+  ROLLBACK_CURRENT_VERSION(29, false), ROLLFORWARD_CURRENT_VERSION(30, false);
 
   private final int value;
   private final boolean batchUpdate;
@@ -140,8 +138,6 @@ public Object getNewInstance() {
         return new RollForwardCurrentVersion();
       case ROLLBACK_CURRENT_VERSION:
         return new RollbackCurrentVersion();
-      case MARK_VERSION_ROLLED_BACK:
-        return new MarkVersionRolledBack();
       default:
         throw new VeniceException("Unsupported " + getClass().getSimpleName() + " value: " + value);
     }