Merge pull request #5971 from sseago/biav2-follow-on

Follow-on fixes for BIAv2 controller work
2026-05-29 18:40:43 +00:00 · 2023-03-14 09:47:32 +08:00
parent 36163c9a0e dd63e8182c
commit eeee4e06d2
32 changed files with 630 additions and 570 deletions
--- a/changelogs/unreleased/5971-sseago
+++ b/changelogs/unreleased/5971-sseago
@@ -0,0 +1 @@
+Follow-on fixes for BIAv2 controller work
--- a/config/crd/v1/bases/velero.io_backups.yaml
+++ b/config/crd/v1/bases/velero.io_backups.yaml
@@ -420,19 +420,19 @@ spec:
          status:
            description: BackupStatus captures the current status of a Velero backup.
            properties:
-              asyncBackupItemOperationsAttempted:
-                description: AsyncBackupItemOperationsAttempted is the total number
-                  of attempted async BackupItemAction operations for this backup.
+              backupItemOperationsAttempted:
+                description: BackupItemOperationsAttempted is the total number of
+                  attempted async BackupItemAction operations for this backup.
                type: integer
-              asyncBackupItemOperationsCompleted:
-                description: AsyncBackupItemOperationsCompleted is the total number
-                  of successfully completed async BackupItemAction operations for
-                  this backup.
+              backupItemOperationsCompleted:
+                description: BackupItemOperationsCompleted is the total number of
+                  successfully completed async BackupItemAction operations for this
+                  backup.
                type: integer
-              asyncBackupItemOperationsFailed:
-                description: AsyncBackupItemOperationsFailed is the total number of
-                  async BackupItemAction operations for this backup which ended with
-                  an error.
+              backupItemOperationsFailed:
+                description: BackupItemOperationsFailed is the total number of async
+                  BackupItemAction operations for this backup which ended with an
+                  error.
                type: integer
              completionTimestamp:
                description: CompletionTimestamp records the time a backup was completed.
@@ -476,8 +476,8 @@ spec:
                - InProgress
                - WaitingForPluginOperations
                - WaitingForPluginOperationsPartiallyFailed
-                - FinalizingAfterPluginOperations
-                - FinalizingAfterPluginOperationsPartiallyFailed
+                - Finalizing
+                - FinalizingPartiallyFailed
                - Completed
                - PartiallyFailed
                - Failed
--- a/config/crd/v1/crds/crds.go
+++ b/config/crd/v1/crds/crds.go
--- a/design/general-progress-monitoring.md
+++ b/design/general-progress-monitoring.md
@@ -123,11 +123,11 @@ from a plugin could cause a backup or restore to move to "PartiallyFailed".  If
 deleted (cancelled), the plug-ins will attempt to delete the snapshots and stop the data movement -
 this may not be possible with all storage systems.

-In addition, for backups (but not restores), there will also be two additional phases,
-"FinalizingAfterPluginOperations" and "FinalizingAfterPluginOperationsPartiallyFailed", which will
-handle any steps required after plugin operations have all completed. Initially, this will just
-include adding any required resources to the backup that might have changed during asynchronous
-operation execution, although eventually other cleanup actions could be added to this phase.
+In addition, for backups (but not restores), there will also be two additional phases, "Finalizing"
+and "FinalizingPartiallyFailed", which will handle any steps required after plugin operations have
+all completed. Initially, this will just include adding any required resources to the backup that
+might have changed during asynchronous operation execution, although eventually other cleanup
+actions could be added to this phase.

 ### State progression

@@ -156,24 +156,24 @@ asynchronous plugin operations and no errors so far, "WaitingForPluginOperations
 backups or restores which have unfinished asynchronous plugin operations at least one error,
 "Completed" for restores with no unfinished asynchronous plugin operations and no errors,
 "PartiallyFailed" for restores with no unfinished asynchronous plugin operations and at least one
-error, "FinalizingAfterPluginOperations" for backups with no unfinished asynchronous plugin
-operations and no errors, "FinalizingAfterPluginOperationsPartiallyFailed" for backups with no
-unfinished asynchronous plugin operations and at least one error, or "PartiallyFailed".
-Backups/restores which would have a final phase of "Completed" or "PartiallyFailed" may move to the
-"WaitingForPluginOperations" or "WaitingForPluginOperationsPartiallyFailed" state.  A backup/restore
-which will be marked "Failed" will go directly to the "Failed" phase.  Uploads may continue in the
-background for snapshots that were taken by a "Failed" backup/restore, but no progress will not be
-monitored or updated. If there are any operations in progress when a backup is moved to the "Failed"
-phase (although with the current workflow, that shouldn't happen), Cancel() should be called on
-these operations. When a "Failed" backup is deleted, all snapshots will be deleted and at that point
-any uploads still in progress should be aborted.
+error, "Finalizing" for backups with no unfinished asynchronous plugin operations and no errors,
+"FinalizingPartiallyFailed" for backups with no unfinished asynchronous plugin operations and at
+least one error, or "PartiallyFailed".  Backups/restores which would have a final phase of
+"Completed" or "PartiallyFailed" may move to the "WaitingForPluginOperations" or
+"WaitingForPluginOperationsPartiallyFailed" state.  A backup/restore which will be marked "Failed"
+will go directly to the "Failed" phase.  Uploads may continue in the background for snapshots that
+were taken by a "Failed" backup/restore, but no progress will not be monitored or updated. If there
+are any operations in progress when a backup is moved to the "Failed" phase (although with the
+current workflow, that shouldn't happen), Cancel() should be called on these operations. When a
+"Failed" backup is deleted, all snapshots will be deleted and at that point any uploads still in
+progress should be aborted.

 ### WaitingForPluginOperations (new)
 The "WaitingForPluginOperations" phase signifies that the main part of the backup/restore, including
 snapshotting has completed successfully and uploading and any other asynchronous BIA/RIA plugin
 operations are continuing.  In the event of an error during this phase, the phase will change to
 WaitingForPluginOperationsPartiallyFailed.  On success, the phase changes to
-"FinalizingAfterPluginOperations" for backups and "Completed" for restores.  Backups cannot be
+"Finalizing" for backups and "Completed" for restores.  Backups cannot be
 restored from when they are in the WaitingForPluginOperations state.

 ### WaitingForPluginOperationsPartiallyFailed (new)
@@ -182,21 +182,19 @@ backup/restore, including snapshotting has completed, but there were partial fai
 the main part or during any async operations, including snapshot uploads.  Backups cannot be
 restored from when they are in the WaitingForPluginOperationsPartiallyFailed state.

-### FinalizingAfterPluginOperations (new)
-The "FinalizingAfterPluginOperations" phase signifies that asynchronous backup operations have all
-completed successfully and Velero is currently backing up any resources indicated by asynchronous
-plugins as items to back up after operations complete. Once this is done, the phase changes to
-Completed.  Backups cannot be restored from when they are in the FinalizingAfterPluginOperations
-state.
+### Finalizing (new)
+The "Finalizing" phase signifies that asynchronous backup operations have all completed successfully
+and Velero is currently backing up any resources indicated by asynchronous plugins as items to back
+up after operations complete. Once this is done, the phase changes to Completed.  Backups cannot be
+restored from when they are in the Finalizing state.

-### FinalizingAfterPluginOperationsPartiallyFailed (new)
+### FinalizingPartiallyFailed (new)

-The "FinalizingAfterPluginOperationsPartiallyFailed" phase signifies that, for a backup which had
-errors during initial processing or asynchronous plugin operation, asynchronous backup operations
-have all completed and Velero is currently backing up any resources indicated by asynchronous
-plugins as items to back up after operations complete. Once this is done, the phase changes to
-PartiallyFailed.  Backups cannot be restored from when they are in the
-FinalizingAfterPluginOperationsPartiallyFailed state.
+The "FinalizingPartiallyFailed" phase signifies that, for a backup which had errors during initial
+processing or asynchronous plugin operation, asynchronous backup operations have all completed and
+Velero is currently backing up any resources indicated by asynchronous plugins as items to back up
+after operations complete. Once this is done, the phase changes to PartiallyFailed.  Backups cannot
+be restored from when they are in the FinalizingPartiallyFailed state.

 ### Failed
 When a backup/restore has had fatal errors it is marked as "Failed" Backups in this state cannot be
@@ -244,14 +242,14 @@ WaitingForPluginOperationsPartiallyFailed phase, another backup/restore may be s
 While in the WaitingForPluginOperations or WaitingForPluginOperationsPartiallyFailed phase, the
 snapshots and item actions will be periodically polled.  When all of the snapshots and item actions
 have reported success, restores will move directly to the Completed or PartiallyFailed phase, and
-backups will move to the FinalizingAfterPluginOperations or
-FinalizingAfterPluginOperationsPartiallyFailed phase, depending on whether the backup/restore was in
-the WaitingForPluginOperations or WaitingForPluginOperationsPartiallyFailed phase.
+backups will move to the Finalizing or FinalizingPartiallyFailed phase, depending on whether the
+backup/restore was in the WaitingForPluginOperations or WaitingForPluginOperationsPartiallyFailed
+phase.

-While in the FinalizingAfterPluginOperations or FinalizingAfterPluginOperationsPartiallyFailed
-phase, Velero will update the backup with any resources indicated by plugins that they must be added
-to the backup after operations are completed, and then the backup will move to the Completed or
-PartiallyFailed phase, depending on whether there are any backup errors.
+While in the Finalizing or FinalizingPartiallyFailed phase, Velero will update the backup with any
+resources indicated by plugins that they must be added to the backup after operations are completed,
+and then the backup will move to the Completed or PartiallyFailed phase, depending on whether there
+are any backup errors.

 The Backup resources will be written to object storage at the time the backup leaves the InProgress
 phase, but it will not be synced to other clusters (or usable for restores in the current cluster)
@@ -272,7 +270,7 @@ ignored.
    type OperationProgress struct {
        Completed bool                          // True when the operation has completed, either successfully or with a failure
        Err string                              // Set when the operation has failed
-        NCompleted, NTotal int64                // Quantity completed so far and the total quantity associated with the operaation in operationUnits
+        NCompleted, NTotal int64                // Quantity completed so far and the total quantity associated with the operation in operationUnits
                                                // For data mover and volume snapshotter use cases, this would be in bytes
                                                // On successful completion, completed and total should be the same.
        OperationUnits string                   // Units represented by completed and total -- for data mover and item
--- a/pkg/apis/velero/v1/backup_types.go
+++ b/pkg/apis/velero/v1/backup_types.go
@@ -226,7 +226,7 @@ const (

 // BackupPhase is a string representation of the lifecycle phase
 // of a Velero backup.
-// +kubebuilder:validation:Enum=New;FailedValidation;InProgress;WaitingForPluginOperations;WaitingForPluginOperationsPartiallyFailed;FinalizingAfterPluginOperations;FinalizingAfterPluginOperationsPartiallyFailed;Completed;PartiallyFailed;Failed;Deleting
+// +kubebuilder:validation:Enum=New;FailedValidation;InProgress;WaitingForPluginOperations;WaitingForPluginOperationsPartiallyFailed;Finalizing;FinalizingPartiallyFailed;Completed;PartiallyFailed;Failed;Deleting
 type BackupPhase string

 const (
@@ -256,22 +256,22 @@ const (
 	// ongoing.  The backup is not usable yet.
 	BackupPhaseWaitingForPluginOperationsPartiallyFailed BackupPhase = "WaitingForPluginOperationsPartiallyFailed"

-	// BackupPhaseFinalizingAfterPluginOperations means the backup of
+	// BackupPhaseFinalizing means the backup of
 	// Kubernetes resources, creation of snapshots, and other
 	// async plugin operations were successful and snapshot upload and
 	// other plugin operations are now complete, but the Backup is awaiting
 	// final update of resources modified during async operations.
 	// The backup is not usable yet.
-	BackupPhaseFinalizingAfterPluginOperations BackupPhase = "FinalizingAfterPluginOperations"
+	BackupPhaseFinalizing BackupPhase = "Finalizing"

-	// BackupPhaseFinalizingAfterPluginOperationsPartiallyFailed means the backup of
+	// BackupPhaseFinalizingPartiallyFailed means the backup of
 	// Kubernetes resources, creation of snapshots, and other
 	// async plugin operations were successful and snapshot upload and
 	// other plugin operations are now complete, but one or more errors
 	// occurred during backup or async operation processing, and the
 	// Backup is awaiting final update of resources modified during async
 	// operations. The backup is not usable yet.
-	BackupPhaseFinalizingAfterPluginOperationsPartiallyFailed BackupPhase = "FinalizingAfterPluginOperationsPartiallyFailed"
+	BackupPhaseFinalizingPartiallyFailed BackupPhase = "FinalizingPartiallyFailed"

 	// BackupPhaseCompleted means the backup has run successfully without
 	// errors.
@@ -374,20 +374,20 @@ type BackupStatus struct {
 	// +optional
 	CSIVolumeSnapshotsCompleted int `json:"csiVolumeSnapshotsCompleted,omitempty"`

-	// AsyncBackupItemOperationsAttempted is the total number of attempted
+	// BackupItemOperationsAttempted is the total number of attempted
 	// async BackupItemAction operations for this backup.
 	// +optional
-	AsyncBackupItemOperationsAttempted int `json:"asyncBackupItemOperationsAttempted,omitempty"`
+	BackupItemOperationsAttempted int `json:"backupItemOperationsAttempted,omitempty"`

-	// AsyncBackupItemOperationsCompleted is the total number of successfully completed
+	// BackupItemOperationsCompleted is the total number of successfully completed
 	// async BackupItemAction operations for this backup.
 	// +optional
-	AsyncBackupItemOperationsCompleted int `json:"asyncBackupItemOperationsCompleted,omitempty"`
+	BackupItemOperationsCompleted int `json:"backupItemOperationsCompleted,omitempty"`

-	// AsyncBackupItemOperationsFailed is the total number of async
+	// BackupItemOperationsFailed is the total number of async
 	// BackupItemAction operations for this backup which ended with an error.
 	// +optional
-	AsyncBackupItemOperationsFailed int `json:"asyncBackupItemOperationsFailed,omitempty"`
+	BackupItemOperationsFailed int `json:"backupItemOperationsFailed,omitempty"`
 }

 // BackupProgress stores information about the progress of a Backup's execution.
--- a/pkg/backup/backup.go
+++ b/pkg/backup/backup.go
@@ -422,7 +422,7 @@ func (kb *kubernetesBackupper) BackupWithResolvers(log logrus.FieldLogger,
 }

 func (kb *kubernetesBackupper) backupItem(log logrus.FieldLogger, gr schema.GroupResource, itemBackupper *itemBackupper, unstructured *unstructured.Unstructured, preferredGVR schema.GroupVersionResource) bool {
-	backedUpItem, err := itemBackupper.backupItem(log, unstructured, gr, preferredGVR, false)
+	backedUpItem, _, err := itemBackupper.backupItem(log, unstructured, gr, preferredGVR, false, false)
 	if aggregate, ok := err.(kubeerrs.Aggregate); ok {
 		log.WithField("name", unstructured.GetName()).Infof("%d errors encountered backup up item", len(aggregate.Errors()))
 		// log each error separately so we get error location info in the log, and an
@@ -441,7 +441,7 @@ func (kb *kubernetesBackupper) backupItem(log logrus.FieldLogger, gr schema.Grou
 }

 func (kb *kubernetesBackupper) finalizeItem(log logrus.FieldLogger, gr schema.GroupResource, itemBackupper *itemBackupper, unstructured *unstructured.Unstructured, preferredGVR schema.GroupVersionResource) (bool, []FileForArchive) {
-	backedUpItem, updateFiles, err := itemBackupper.finalizeItem(log, unstructured, gr, preferredGVR)
+	backedUpItem, updateFiles, err := itemBackupper.backupItem(log, unstructured, gr, preferredGVR, true, true)
 	if aggregate, ok := err.(kubeerrs.Aggregate); ok {
 		log.WithField("name", unstructured.GetName()).Infof("%d errors encountered backup up item", len(aggregate.Errors()))
 		// log each error separately so we get error location info in the log, and an
@@ -563,15 +563,15 @@ func (kb *kubernetesBackupper) FinalizeBackup(log logrus.FieldLogger,
 		pageSize:              kb.clientPageSize,
 	}

-	// Get item list from itemoperation.BackupOperation.Spec.ItemsToUpdate
+	// Get item list from itemoperation.BackupOperation.Spec.PostOperationItems
 	var resourceIDs []velero.ResourceIdentifier
 	for _, operation := range asyncBIAOperations {
-		if len(operation.Spec.ItemsToUpdate) != 0 {
-			resourceIDs = append(resourceIDs, operation.Spec.ItemsToUpdate...)
+		if len(operation.Spec.PostOperationItems) != 0 {
+			resourceIDs = append(resourceIDs, operation.Spec.PostOperationItems...)
 		}
 	}
 	items := collector.getItemsFromResourceIdentifiers(resourceIDs)
-	log.WithField("progress", "").Infof("Collected %d items from the async BIA operations ItemsToUpdate list", len(items))
+	log.WithField("progress", "").Infof("Collected %d items from the async BIA operations PostOperationItems list", len(items))

 	itemBackupper := &itemBackupper{
 		backupRequest:   backupRequest,
--- a/pkg/backup/backup_test.go
+++ b/pkg/backup/backup_test.go
@@ -1136,23 +1136,23 @@ func TestBackupResourceOrdering(t *testing.T) {
 // to run for specific resources/namespaces and simply records the items
 // that it is executed for.
 type recordResourcesAction struct {
-	selector        velero.ResourceSelector
-	ids             []string
-	backups         []velerov1.Backup
-	additionalItems []velero.ResourceIdentifier
-	operationID     string
-	itemsToUpdate   []velero.ResourceIdentifier
+	selector           velero.ResourceSelector
+	ids                []string
+	backups            []velerov1.Backup
+	additionalItems    []velero.ResourceIdentifier
+	operationID        string
+	postOperationItems []velero.ResourceIdentifier
 }

 func (a *recordResourcesAction) Execute(item runtime.Unstructured, backup *velerov1.Backup) (runtime.Unstructured, []velero.ResourceIdentifier, string, []velero.ResourceIdentifier, error) {
 	metadata, err := meta.Accessor(item)
 	if err != nil {
-		return item, a.additionalItems, a.operationID, a.itemsToUpdate, err
+		return item, a.additionalItems, a.operationID, a.postOperationItems, err
 	}
 	a.ids = append(a.ids, kubeutil.NamespaceAndName(metadata))
 	a.backups = append(a.backups, *backup)

-	return item, a.additionalItems, a.operationID, a.itemsToUpdate, nil
+	return item, a.additionalItems, a.operationID, a.postOperationItems, nil
 }

 func (a *recordResourcesAction) AppliesTo() (velero.ResourceSelector, error) {
--- a/pkg/backup/item_backupper.go
+++ b/pkg/backup/item_backupper.go
@@ -76,34 +76,27 @@ type FileForArchive struct {
 	FileBytes []byte
 }

-// finalizeItem backs up an individual item and returns its content to replace previous content
-// in the backup tarball
-// In addition to the error return, backupItem also returns a bool indicating whether the item
-// was actually backed up and a slice of filepaths and filecontent to replace the data in the original tarball.
-func (ib *itemBackupper) finalizeItem(logger logrus.FieldLogger, obj runtime.Unstructured, groupResource schema.GroupResource, preferredGVR schema.GroupVersionResource) (bool, []FileForArchive, error) {
-	return ib.backupItemInternal(logger, obj, groupResource, preferredGVR, true, true)
-}
-
 // backupItem backs up an individual item to tarWriter. The item may be excluded based on the
 // namespaces IncludesExcludes list.
+// If finalize is true, then it returns the bytes instead of writing them to the tarWriter
 // In addition to the error return, backupItem also returns a bool indicating whether the item
 // was actually backed up.
-func (ib *itemBackupper) backupItem(logger logrus.FieldLogger, obj runtime.Unstructured, groupResource schema.GroupResource, preferredGVR schema.GroupVersionResource, mustInclude bool) (bool, error) {
-	selectedForBackup, files, err := ib.backupItemInternal(logger, obj, groupResource, preferredGVR, mustInclude, false)
-	// return if not selected, an error occurred, or there are no files to add
-	if selectedForBackup == false || err != nil || len(files) == 0 {
-		return selectedForBackup, err
+func (ib *itemBackupper) backupItem(logger logrus.FieldLogger, obj runtime.Unstructured, groupResource schema.GroupResource, preferredGVR schema.GroupVersionResource, mustInclude, finalize bool) (bool, []FileForArchive, error) {
+	selectedForBackup, files, err := ib.backupItemInternal(logger, obj, groupResource, preferredGVR, mustInclude, finalize)
+	// return if not selected, an error occurred, there are no files to add, or for finalize
+	if selectedForBackup == false || err != nil || len(files) == 0 || finalize {
+		return selectedForBackup, files, err
 	}
 	for _, file := range files {
 		if err := ib.tarWriter.WriteHeader(file.Header); err != nil {
-			return false, errors.WithStack(err)
+			return false, []FileForArchive{}, errors.WithStack(err)
 		}

 		if _, err := ib.tarWriter.Write(file.FileBytes); err != nil {
-			return false, errors.WithStack(err)
+			return false, []FileForArchive{}, errors.WithStack(err)
 		}
 	}
-	return true, nil
+	return true, []FileForArchive{}, nil
 }

 func (ib *itemBackupper) backupItemInternal(logger logrus.FieldLogger, obj runtime.Unstructured, groupResource schema.GroupResource, preferredGVR schema.GroupVersionResource, mustInclude, finalize bool) (bool, []FileForArchive, error) {
@@ -178,7 +171,7 @@ func (ib *itemBackupper) backupItemInternal(logger logrus.FieldLogger, obj runti
 		return false, itemFiles, err
 	}

-	if !finalize && groupResource == kuberesource.Pods {
+	if groupResource == kuberesource.Pods {
 		// pod needs to be initialized for the unstructured converter
 		pod = new(corev1api.Pod)
 		if err := runtime.DefaultUnstructuredConverter.FromUnstructured(obj.UnstructuredContent(), pod); err != nil {
@@ -211,7 +204,7 @@ func (ib *itemBackupper) backupItemInternal(logger logrus.FieldLogger, obj runti
 	// the group version of the object.
 	versionPath := resourceVersion(obj)

-	updatedObj, err := ib.executeActions(log, obj, groupResource, name, namespace, metadata, finalize)
+	updatedObj, additionalItemFiles, err := ib.executeActions(log, obj, groupResource, name, namespace, metadata, finalize)
 	if err != nil {
 		backupErrs = append(backupErrs, err)

@@ -222,6 +215,7 @@ func (ib *itemBackupper) backupItemInternal(logger logrus.FieldLogger, obj runti
 		}
 		return false, itemFiles, kubeerrs.NewAggregate(backupErrs)
 	}
+	itemFiles = append(itemFiles, additionalItemFiles...)
 	obj = updatedObj
 	if metadata, err = meta.Accessor(obj); err != nil {
 		return false, itemFiles, errors.WithStack(err)
@@ -230,13 +224,13 @@ func (ib *itemBackupper) backupItemInternal(logger logrus.FieldLogger, obj runti
 	name = metadata.GetName()
 	namespace = metadata.GetNamespace()

-	if !finalize && groupResource == kuberesource.PersistentVolumes {
+	if groupResource == kuberesource.PersistentVolumes {
 		if err := ib.takePVSnapshot(obj, log); err != nil {
 			backupErrs = append(backupErrs, err)
 		}
 	}

-	if !finalize && groupResource == kuberesource.Pods && pod != nil {
+	if groupResource == kuberesource.Pods && pod != nil {
 		// this function will return partial results, so process podVolumeBackups
 		// even if there are errors.
 		podVolumeBackups, errs := ib.backupPodVolumes(log, pod, pvbVolumes)
@@ -309,33 +303,34 @@ func (ib *itemBackupper) executeActions(
 	name, namespace string,
 	metadata metav1.Object,
 	finalize bool,
-) (runtime.Unstructured, error) {
+) (runtime.Unstructured, []FileForArchive, error) {
+	var itemFiles []FileForArchive
 	for _, action := range ib.backupRequest.ResolvedActions {
 		if !action.ShouldUse(groupResource, namespace, metadata, log) {
 			continue
 		}
 		log.Info("Executing custom action")

-		updatedItem, additionalItemIdentifiers, operationID, itemsToUpdate, err := action.Execute(obj, ib.backupRequest.Backup)
+		updatedItem, additionalItemIdentifiers, operationID, postOperationItems, err := action.Execute(obj, ib.backupRequest.Backup)
 		if err != nil {
-			return nil, errors.Wrapf(err, "error executing custom action (groupResource=%s, namespace=%s, name=%s)", groupResource.String(), namespace, name)
+			return nil, itemFiles, errors.Wrapf(err, "error executing custom action (groupResource=%s, namespace=%s, name=%s)", groupResource.String(), namespace, name)
 		}

 		u := &unstructured.Unstructured{Object: updatedItem.UnstructuredContent()}
-		mustInclude := u.GetAnnotations()[mustIncludeAdditionalItemAnnotation] == "true"
+		mustInclude := u.GetAnnotations()[mustIncludeAdditionalItemAnnotation] == "true" || finalize
 		// remove the annotation as it's for communication between BIA and velero server,
 		// we don't want the resource be restored with this annotation.
 		if _, ok := u.GetAnnotations()[mustIncludeAdditionalItemAnnotation]; ok {
 			delete(u.GetAnnotations(), mustIncludeAdditionalItemAnnotation)
 		}
 		obj = u
-		if finalize {
-			continue
-		}

 		// If async plugin started async operation, add it to the ItemOperations list
 		// ignore during finalize phase
 		if operationID != "" {
+			if finalize {
+				return nil, itemFiles, errors.New(fmt.Sprintf("Backup Item Action created operation during finalize (groupResource=%s, namespace=%s, name=%s)", groupResource.String(), namespace, name))
+			}
 			resourceIdentifier := velero.ResourceIdentifier{
 				GroupResource: groupResource,
 				Namespace:     namespace,
@@ -355,7 +350,7 @@ func (ib *itemBackupper) executeActions(
 					Created: &now,
 				},
 			}
-			newOperation.Spec.ItemsToUpdate = itemsToUpdate
+			newOperation.Spec.PostOperationItems = postOperationItems
 			itemOperList := ib.backupRequest.GetItemOperationsList()
 			*itemOperList = append(*itemOperList, &newOperation)
 		}
@@ -363,12 +358,12 @@ func (ib *itemBackupper) executeActions(
 		for _, additionalItem := range additionalItemIdentifiers {
 			gvr, resource, err := ib.discoveryHelper.ResourceFor(additionalItem.GroupResource.WithVersion(""))
 			if err != nil {
-				return nil, err
+				return nil, itemFiles, err
 			}

 			client, err := ib.dynamicFactory.ClientForGroupVersionResource(gvr.GroupVersion(), resource, additionalItem.Namespace)
 			if err != nil {
-				return nil, err
+				return nil, itemFiles, err
 			}

 			item, err := client.Get(additionalItem.Name, metav1.GetOptions{})
@@ -382,15 +377,17 @@ func (ib *itemBackupper) executeActions(
 				continue
 			}
 			if err != nil {
-				return nil, errors.WithStack(err)
+				return nil, itemFiles, errors.WithStack(err)
 			}

-			if _, err = ib.backupItem(log, item, gvr.GroupResource(), gvr, mustInclude); err != nil {
-				return nil, err
+			_, additionalItemFiles, err := ib.backupItem(log, item, gvr.GroupResource(), gvr, mustInclude, finalize)
+			if err != nil {
+				return nil, itemFiles, err
 			}
+			itemFiles = append(itemFiles, additionalItemFiles...)
 		}
 	}
-	return obj, nil
+	return obj, itemFiles, nil
 }

 // volumeSnapshotter instantiates and initializes a VolumeSnapshotter given a VolumeSnapshotLocation,
--- a/pkg/backup/item_collector.go
+++ b/pkg/backup/item_collector.go
@@ -75,7 +75,14 @@ func (r *itemCollector) getAllItems() []*kubernetesResource {
 	return r.getItems(nil)
 }

-// getAllItems gets all relevant items from all API groups.
+// getItems gets all relevant items from all API groups.
+// If resourceIDsMap is nil, then all items from the cluster are
+// pulled for each API group, subject to include/exclude rules.
+// If resourceIDsMap is supplied, then only those resources are
+// returned, with the appropriate APIGroup information filled in. In
+// this case, include/exclude rules are not invoked, since we already
+// have the list of items, we just need the item collector/discovery
+// helper to fill in the missing GVR, etc. context.
 func (r *itemCollector) getItems(resourceIDsMap map[schema.GroupResource][]velero.ResourceIdentifier) []*kubernetesResource {
 	var resources []*kubernetesResource
 	for _, group := range r.discoveryHelper.Resources() {
@@ -92,6 +99,8 @@ func (r *itemCollector) getItems(resourceIDsMap map[schema.GroupResource][]veler
 }

 // getGroupItems collects all relevant items from a single API group.
+// If resourceIDsMap is supplied, then only those items are returned,
+// with GVR/APIResource metadata supplied.
 func (r *itemCollector) getGroupItems(log logrus.FieldLogger, group *metav1.APIResourceList, resourceIDsMap map[schema.GroupResource][]velero.ResourceIdentifier) ([]*kubernetesResource, error) {
 	log = log.WithField("group", group.GroupVersion)

@@ -181,6 +190,8 @@ func getOrderedResourcesForType(orderedResources map[string]string, resourceType
 }

 // getResourceItems collects all relevant items for a given group-version-resource.
+// If resourceIDsMap is supplied, the items will be pulled from here
+// rather than from the cluster and applying include/exclude rules.
 func (r *itemCollector) getResourceItems(log logrus.FieldLogger, gv schema.GroupVersion, resource metav1.APIResource, resourceIDsMap map[schema.GroupResource][]velero.ResourceIdentifier) ([]*kubernetesResource, error) {
 	log = log.WithField("resource", resource.Name)

--- a/pkg/cmd/server/server.go
+++ b/pkg/cmd/server/server.go
@@ -65,6 +65,7 @@ import (
 	velerodiscovery "github.com/vmware-tanzu/velero/pkg/discovery"
 	"github.com/vmware-tanzu/velero/pkg/features"
 	clientset "github.com/vmware-tanzu/velero/pkg/generated/clientset/versioned"
+	"github.com/vmware-tanzu/velero/pkg/itemoperationmap"
 	"github.com/vmware-tanzu/velero/pkg/metrics"
 	"github.com/vmware-tanzu/velero/pkg/nodeagent"
 	"github.com/vmware-tanzu/velero/pkg/persistence"
@@ -226,7 +227,7 @@ func NewCommand(f client.Factory) *cobra.Command {
 	command.Flags().DurationVar(&config.defaultBackupTTL, "default-backup-ttl", config.defaultBackupTTL, "How long to wait by default before backups can be garbage collected.")
 	command.Flags().DurationVar(&config.repoMaintenanceFrequency, "default-repo-maintain-frequency", config.repoMaintenanceFrequency, "How often 'maintain' is run for backup repositories by default.")
 	command.Flags().DurationVar(&config.garbageCollectionFrequency, "garbage-collection-frequency", config.garbageCollectionFrequency, "How often garbage collection is run for expired backups.")
-	command.Flags().DurationVar(&config.itemOperationSyncFrequency, "item-operation-sync-frequency", config.itemOperationSyncFrequency, "How often to check status on async backup/restore operations after backup processing.")
+	command.Flags().DurationVar(&config.itemOperationSyncFrequency, "item-operation-sync-frequency", config.itemOperationSyncFrequency, "How often to check status on backup/restore operations after backup/restore processing.")
 	command.Flags().BoolVar(&config.defaultVolumesToFsBackup, "default-volumes-to-fs-backup", config.defaultVolumesToFsBackup, "Backup all volumes with pod volume file system backup by default.")
 	command.Flags().StringVar(&config.uploaderType, "uploader-type", config.uploaderType, "Type of uploader to handle the transfer of data of pod volumes")
 	command.Flags().DurationVar(&config.defaultItemOperationTimeout, "default-item-operation-timeout", config.defaultItemOperationTimeout, "How long to wait on asynchronous BackupItemActions and RestoreItemActions to complete before timing out.")
@@ -642,26 +643,26 @@ func (s *server) runControllers(defaultVolumeSnapshotLocations map[string]string
 	// and BSL controller is mandatory for Velero to work.
 	// Note: all runtime type controllers that can be disabled are grouped separately, below:
 	enabledRuntimeControllers := map[string]struct{}{
-		controller.AsyncBackupOperations: {},
-		controller.Backup:                {},
-		controller.BackupDeletion:        {},
-		controller.BackupFinalizer:       {},
-		controller.BackupRepo:            {},
-		controller.BackupSync:            {},
-		controller.DownloadRequest:       {},
-		controller.GarbageCollection:     {},
-		controller.Restore:               {},
-		controller.Schedule:              {},
-		controller.ServerStatusRequest:   {},
+		controller.Backup:              {},
+		controller.BackupDeletion:      {},
+		controller.BackupFinalizer:     {},
+		controller.BackupOperations:    {},
+		controller.BackupRepo:          {},
+		controller.BackupSync:          {},
+		controller.DownloadRequest:     {},
+		controller.GarbageCollection:   {},
+		controller.Restore:             {},
+		controller.Schedule:            {},
+		controller.ServerStatusRequest: {},
 	}

 	if s.config.restoreOnly {
 		s.logger.Info("Restore only mode - not starting the backup, schedule, delete-backup, or GC controllers")
 		s.config.disabledControllers = append(s.config.disabledControllers,
-			controller.AsyncBackupOperations,
 			controller.Backup,
 			controller.BackupDeletion,
 			controller.BackupFinalizer,
+			controller.BackupOperations,
 			controller.GarbageCollection,
 			controller.Schedule,
 		)
@@ -690,22 +691,6 @@ func (s *server) runControllers(defaultVolumeSnapshotLocations map[string]string
 		s.logger.Fatal(err, "unable to create controller", "controller", controller.BackupStorageLocation)
 	}

-	var backupOpsMap *controller.BackupItemOperationsMap
-	if _, ok := enabledRuntimeControllers[controller.AsyncBackupOperations]; ok {
-		r, m := controller.NewAsyncBackupOperationsReconciler(
-			s.logger,
-			s.mgr.GetClient(),
-			s.config.itemOperationSyncFrequency,
-			newPluginManager,
-			backupStoreGetter,
-			s.metrics,
-		)
-		if err := r.SetupWithManager(s.mgr); err != nil {
-			s.logger.Fatal(err, "unable to create controller", "controller", controller.AsyncBackupOperations)
-		}
-		backupOpsMap = m
-	}
-
 	if _, ok := enabledRuntimeControllers[controller.Backup]; ok {
 		backupper, err := backup.NewKubernetesBackupper(
 			s.mgr.GetClient(),
@@ -770,6 +755,22 @@ func (s *server) runControllers(defaultVolumeSnapshotLocations map[string]string
 		}
 	}

+	backupOpsMap := itemoperationmap.NewBackupItemOperationsMap()
+	if _, ok := enabledRuntimeControllers[controller.BackupOperations]; ok {
+		r := controller.NewBackupOperationsReconciler(
+			s.logger,
+			s.mgr.GetClient(),
+			s.config.itemOperationSyncFrequency,
+			newPluginManager,
+			backupStoreGetter,
+			s.metrics,
+			backupOpsMap,
+		)
+		if err := r.SetupWithManager(s.mgr); err != nil {
+			s.logger.Fatal(err, "unable to create controller", "controller", controller.BackupOperations)
+		}
+	}
+
 	if _, ok := enabledRuntimeControllers[controller.BackupFinalizer]; ok {
 		backupper, err := backup.NewKubernetesBackupper(
 			s.mgr.GetClient(),
--- a/pkg/cmd/server/server_test.go
+++ b/pkg/cmd/server/server_test.go
@@ -89,7 +89,7 @@ func TestRemoveControllers(t *testing.T) {
 		{
 			name: "Remove all disable controllers",
 			disabledControllers: []string{
-				controller.AsyncBackupOperations,
+				controller.BackupOperations,
 				controller.Backup,
 				controller.BackupDeletion,
 				controller.BackupSync,
@@ -121,16 +121,16 @@ func TestRemoveControllers(t *testing.T) {
 	for _, tt := range tests {
 		t.Run(tt.name, func(t *testing.T) {
 			enabledRuntimeControllers := map[string]struct{}{
-				controller.BackupSync:            {},
-				controller.Backup:                {},
-				controller.GarbageCollection:     {},
-				controller.Restore:               {},
-				controller.ServerStatusRequest:   {},
-				controller.Schedule:              {},
-				controller.BackupDeletion:        {},
-				controller.BackupRepo:            {},
-				controller.DownloadRequest:       {},
-				controller.AsyncBackupOperations: {},
+				controller.BackupSync:          {},
+				controller.Backup:              {},
+				controller.GarbageCollection:   {},
+				controller.Restore:             {},
+				controller.ServerStatusRequest: {},
+				controller.Schedule:            {},
+				controller.BackupDeletion:      {},
+				controller.BackupRepo:          {},
+				controller.DownloadRequest:     {},
+				controller.BackupOperations:    {},
 			}

 			totalNumOriginalControllers := len(enabledRuntimeControllers)
--- a/pkg/cmd/util/output/backup_describer.go
+++ b/pkg/cmd/util/output/backup_describer.go
@@ -68,7 +68,7 @@ func DescribeBackup(
 			phaseString = color.GreenString(phaseString)
 		case velerov1api.BackupPhaseDeleting:
 		case velerov1api.BackupPhaseWaitingForPluginOperations, velerov1api.BackupPhaseWaitingForPluginOperationsPartiallyFailed:
-		case velerov1api.BackupPhaseFinalizingAfterPluginOperations, velerov1api.BackupPhaseFinalizingAfterPluginOperationsPartiallyFailed:
+		case velerov1api.BackupPhaseFinalizing, velerov1api.BackupPhaseFinalizingPartiallyFailed:
 		case velerov1api.BackupPhaseInProgress:
 		case velerov1api.BackupPhaseNew:
 		}
@@ -288,7 +288,7 @@ func DescribeBackupStatus(ctx context.Context, kbClient kbclient.Client, d *Desc
 		d.Println()
 	}

-	describeAsyncBackupItemOperations(ctx, kbClient, d, backup, details, insecureSkipTLSVerify, caCertPath)
+	describeBackupItemOperations(ctx, kbClient, d, backup, details, insecureSkipTLSVerify, caCertPath)

 	if details {
 		describeBackupResourceList(ctx, kbClient, d, backup, insecureSkipTLSVerify, caCertPath)
@@ -323,29 +323,29 @@ func DescribeBackupStatus(ctx context.Context, kbClient kbclient.Client, d *Desc
 	d.Printf("Velero-Native Snapshots: <none included>\n")
 }

-func describeAsyncBackupItemOperations(ctx context.Context, kbClient kbclient.Client, d *Describer, backup *velerov1api.Backup, details bool, insecureSkipTLSVerify bool, caCertPath string) {
+func describeBackupItemOperations(ctx context.Context, kbClient kbclient.Client, d *Describer, backup *velerov1api.Backup, details bool, insecureSkipTLSVerify bool, caCertPath string) {
 	status := backup.Status
-	if status.AsyncBackupItemOperationsAttempted > 0 {
+	if status.BackupItemOperationsAttempted > 0 {
 		if !details {
-			d.Printf("Async Backup Item Operations:\t%d of %d completed successfully, %d failed (specify --details for more information)\n", status.AsyncBackupItemOperationsCompleted, status.AsyncBackupItemOperationsAttempted, status.AsyncBackupItemOperationsFailed)
+			d.Printf("Backup Item Operations:\t%d of %d completed successfully, %d failed (specify --details for more information)\n", status.BackupItemOperationsCompleted, status.BackupItemOperationsAttempted, status.BackupItemOperationsFailed)
 			return
 		}

 		buf := new(bytes.Buffer)
 		if err := downloadrequest.Stream(ctx, kbClient, backup.Namespace, backup.Name, velerov1api.DownloadTargetKindBackupItemOperations, buf, downloadRequestTimeout, insecureSkipTLSVerify, caCertPath); err != nil {
-			d.Printf("Async Backup Item Operations:\t<error getting operation info: %v>\n", err)
+			d.Printf("Backup Item Operations:\t<error getting operation info: %v>\n", err)
 			return
 		}

 		var operations []*itemoperation.BackupOperation
 		if err := json.NewDecoder(buf).Decode(&operations); err != nil {
-			d.Printf("Async Backup Item Operations:\t<error reading operation info: %v>\n", err)
+			d.Printf("Backup Item Operations:\t<error reading operation info: %v>\n", err)
 			return
 		}

-		d.Printf("Async Backup Item Operations:\n")
+		d.Printf("Backup Item Operations:\n")
 		for _, operation := range operations {
-			describeAsyncBackupItemOperation(d, operation)
+			describeBackupItemOperation(d, operation)
 		}
 	}
 }
@@ -398,14 +398,14 @@ func describeSnapshot(d *Describer, pvName, snapshotID, volumeType, volumeAZ str
 	d.Printf("\t\tIOPS:\t%s\n", iopsString)
 }

-func describeAsyncBackupItemOperation(d *Describer, operation *itemoperation.BackupOperation) {
+func describeBackupItemOperation(d *Describer, operation *itemoperation.BackupOperation) {
 	d.Printf("\tOperation for %s %s/%s:\n", operation.Spec.ResourceIdentifier, operation.Spec.ResourceIdentifier.Namespace, operation.Spec.ResourceIdentifier.Name)
 	d.Printf("\t\tBackup Item Action Plugin:\t%s\n", operation.Spec.BackupItemAction)
 	d.Printf("\t\tOperation ID:\t%s\n", operation.Spec.OperationID)
-	if len(operation.Spec.ItemsToUpdate) > 0 {
+	if len(operation.Spec.PostOperationItems) > 0 {
 		d.Printf("\t\tItems to Update:\n")
 	}
-	for _, item := range operation.Spec.ItemsToUpdate {
+	for _, item := range operation.Spec.PostOperationItems {
 		d.Printf("\t\t\t%s %s/%s\n", item, item.Namespace, item.Name)
 	}
 	d.Printf("\t\tPhase:\t%s\n", operation.Status.Phase)
--- a/pkg/controller/backup_controller.go
+++ b/pkg/controller/backup_controller.go
@@ -18,9 +18,7 @@ package controller

 import (
 	"bytes"
-	"compress/gzip"
 	"context"
-	"encoding/json"
 	"fmt"
 	"io/ioutil"
 	"os"
@@ -687,9 +685,9 @@ func (b *backupReconciler) runBackup(backup *pkgbackup.Request) error {
 		}
 	}

-	backup.Status.AsyncBackupItemOperationsAttempted = len(*backup.GetItemOperationsList())
-	backup.Status.AsyncBackupItemOperationsCompleted = opsCompleted
-	backup.Status.AsyncBackupItemOperationsFailed = opsFailed
+	backup.Status.BackupItemOperationsAttempted = len(*backup.GetItemOperationsList())
+	backup.Status.BackupItemOperationsCompleted = opsCompleted
+	backup.Status.BackupItemOperationsFailed = opsFailed

 	backup.Status.Warnings = logCounter.GetCount(logrus.WarnLevel)
 	backup.Status.Errors = logCounter.GetCount(logrus.ErrorLevel)
@@ -714,13 +712,13 @@ func (b *backupReconciler) runBackup(backup *pkgbackup.Request) error {
 		if inProgressOperations {
 			backup.Status.Phase = velerov1api.BackupPhaseWaitingForPluginOperationsPartiallyFailed
 		} else {
-			backup.Status.Phase = velerov1api.BackupPhaseFinalizingAfterPluginOperationsPartiallyFailed
+			backup.Status.Phase = velerov1api.BackupPhaseFinalizingPartiallyFailed
 		}
 	default:
 		if inProgressOperations {
 			backup.Status.Phase = velerov1api.BackupPhaseWaitingForPluginOperations
 		} else {
-			backup.Status.Phase = velerov1api.BackupPhaseFinalizingAfterPluginOperations
+			backup.Status.Phase = velerov1api.BackupPhaseFinalizing
 		}
 	}
 	// Mark completion timestamp before serializing and uploading.
@@ -811,42 +809,42 @@ func persistBackup(backup *pkgbackup.Request,
 	}

 	// Velero-native volume snapshots (as opposed to CSI ones)
-	nativeVolumeSnapshots, errs := encodeToJSONGzip(backup.VolumeSnapshots, "native volumesnapshots list")
+	nativeVolumeSnapshots, errs := encode.EncodeToJSONGzip(backup.VolumeSnapshots, "native volumesnapshots list")
 	if errs != nil {
 		persistErrs = append(persistErrs, errs...)
 	}

 	var backupItemOperations *bytes.Buffer
-	backupItemOperations, errs = encodeToJSONGzip(backup.GetItemOperationsList(), "backup item operations list")
+	backupItemOperations, errs = encode.EncodeToJSONGzip(backup.GetItemOperationsList(), "backup item operations list")
 	if errs != nil {
 		persistErrs = append(persistErrs, errs...)
 	}

-	podVolumeBackups, errs := encodeToJSONGzip(backup.PodVolumeBackups, "pod volume backups list")
+	podVolumeBackups, errs := encode.EncodeToJSONGzip(backup.PodVolumeBackups, "pod volume backups list")
 	if errs != nil {
 		persistErrs = append(persistErrs, errs...)
 	}

-	csiSnapshotJSON, errs := encodeToJSONGzip(csiVolumeSnapshots, "csi volume snapshots list")
+	csiSnapshotJSON, errs := encode.EncodeToJSONGzip(csiVolumeSnapshots, "csi volume snapshots list")
 	if errs != nil {
 		persistErrs = append(persistErrs, errs...)
 	}

-	csiSnapshotContentsJSON, errs := encodeToJSONGzip(csiVolumeSnapshotContents, "csi volume snapshot contents list")
+	csiSnapshotContentsJSON, errs := encode.EncodeToJSONGzip(csiVolumeSnapshotContents, "csi volume snapshot contents list")
 	if errs != nil {
 		persistErrs = append(persistErrs, errs...)
 	}
-	csiSnapshotClassesJSON, errs := encodeToJSONGzip(csiVolumesnapshotClasses, "csi volume snapshot classes list")
+	csiSnapshotClassesJSON, errs := encode.EncodeToJSONGzip(csiVolumesnapshotClasses, "csi volume snapshot classes list")
 	if errs != nil {
 		persistErrs = append(persistErrs, errs...)
 	}

-	backupResourceList, errs := encodeToJSONGzip(backup.BackupResourceList(), "backup resources list")
+	backupResourceList, errs := encode.EncodeToJSONGzip(backup.BackupResourceList(), "backup resources list")
 	if errs != nil {
 		persistErrs = append(persistErrs, errs...)
 	}

-	backupResult, errs := encodeToJSONGzip(results, "backup results")
+	backupResult, errs := encode.EncodeToJSONGzip(results, "backup results")
 	if errs != nil {
 		persistErrs = append(persistErrs, errs...)
 	}
@@ -898,29 +896,6 @@ func closeAndRemoveFile(file *os.File, log logrus.FieldLogger) {
 	}
 }

-// encodeToJSONGzip takes arbitrary Go data and encodes it to GZip compressed JSON in a buffer, as well as a description of the data to put into an error should encoding fail.
-func encodeToJSONGzip(data interface{}, desc string) (*bytes.Buffer, []error) {
-	buf := new(bytes.Buffer)
-	gzw := gzip.NewWriter(buf)
-
-	// Since both encoding and closing the gzip writer could fail separately and both errors are useful,
-	// collect both errors to report back.
-	errs := []error{}
-
-	if err := json.NewEncoder(gzw).Encode(data); err != nil {
-		errs = append(errs, errors.Wrapf(err, "error encoding %s", desc))
-	}
-	if err := gzw.Close(); err != nil {
-		errs = append(errs, errors.Wrapf(err, "error closing gzip writer for %s", desc))
-	}
-
-	if len(errs) > 0 {
-		return nil, errs
-	}
-
-	return buf, nil
-}
-
 // waitVolumeSnapshotReadyToUse is used to wait VolumeSnapshot turned to ReadyToUse.
 // Waiting for VolumeSnapshot ReadyToUse to true is time consuming. Try to make the process parallel by
 // using goroutine here instead of waiting in CSI plugin, because it's not easy to make BackupItemAction
--- a/pkg/controller/backup_controller_test.go
+++ b/pkg/controller/backup_controller_test.go
@@ -580,7 +580,7 @@ func TestProcessBackupCompletions(t *testing.T) {
 		backupExists             bool
 		existenceCheckError      error
 	}{
-		// FinalizingAfterPluginOperations
+		// Finalizing
 		{
 			name:                     "backup with no backup location gets the default",
 			backup:                   defaultBackup().Result(),
@@ -608,7 +608,7 @@ func TestProcessBackupCompletions(t *testing.T) {
 					DefaultVolumesToFsBackup: boolptr.True(),
 				},
 				Status: velerov1api.BackupStatus{
-					Phase:          velerov1api.BackupPhaseFinalizingAfterPluginOperations,
+					Phase:          velerov1api.BackupPhaseFinalizing,
 					Version:        1,
 					FormatVersion:  "1.1.0",
 					StartTimestamp: &timestamp,
@@ -643,7 +643,7 @@ func TestProcessBackupCompletions(t *testing.T) {
 					DefaultVolumesToFsBackup: boolptr.False(),
 				},
 				Status: velerov1api.BackupStatus{
-					Phase:          velerov1api.BackupPhaseFinalizingAfterPluginOperations,
+					Phase:          velerov1api.BackupPhaseFinalizing,
 					Version:        1,
 					FormatVersion:  "1.1.0",
 					StartTimestamp: &timestamp,
@@ -681,7 +681,7 @@ func TestProcessBackupCompletions(t *testing.T) {
 					DefaultVolumesToFsBackup: boolptr.True(),
 				},
 				Status: velerov1api.BackupStatus{
-					Phase:          velerov1api.BackupPhaseFinalizingAfterPluginOperations,
+					Phase:          velerov1api.BackupPhaseFinalizing,
 					Version:        1,
 					FormatVersion:  "1.1.0",
 					StartTimestamp: &timestamp,
@@ -717,7 +717,7 @@ func TestProcessBackupCompletions(t *testing.T) {
 					DefaultVolumesToFsBackup: boolptr.False(),
 				},
 				Status: velerov1api.BackupStatus{
-					Phase:          velerov1api.BackupPhaseFinalizingAfterPluginOperations,
+					Phase:          velerov1api.BackupPhaseFinalizing,
 					Version:        1,
 					FormatVersion:  "1.1.0",
 					Expiration:     &metav1.Time{now.Add(10 * time.Minute)},
@@ -753,7 +753,7 @@ func TestProcessBackupCompletions(t *testing.T) {
 					DefaultVolumesToFsBackup: boolptr.True(),
 				},
 				Status: velerov1api.BackupStatus{
-					Phase:          velerov1api.BackupPhaseFinalizingAfterPluginOperations,
+					Phase:          velerov1api.BackupPhaseFinalizing,
 					Version:        1,
 					FormatVersion:  "1.1.0",
 					StartTimestamp: &timestamp,
@@ -790,7 +790,7 @@ func TestProcessBackupCompletions(t *testing.T) {
 					DefaultVolumesToFsBackup: boolptr.False(),
 				},
 				Status: velerov1api.BackupStatus{
-					Phase:          velerov1api.BackupPhaseFinalizingAfterPluginOperations,
+					Phase:          velerov1api.BackupPhaseFinalizing,
 					Version:        1,
 					FormatVersion:  "1.1.0",
 					StartTimestamp: &timestamp,
@@ -827,7 +827,7 @@ func TestProcessBackupCompletions(t *testing.T) {
 					DefaultVolumesToFsBackup: boolptr.True(),
 				},
 				Status: velerov1api.BackupStatus{
-					Phase:          velerov1api.BackupPhaseFinalizingAfterPluginOperations,
+					Phase:          velerov1api.BackupPhaseFinalizing,
 					Version:        1,
 					FormatVersion:  "1.1.0",
 					StartTimestamp: &timestamp,
@@ -864,7 +864,7 @@ func TestProcessBackupCompletions(t *testing.T) {
 					DefaultVolumesToFsBackup: boolptr.True(),
 				},
 				Status: velerov1api.BackupStatus{
-					Phase:          velerov1api.BackupPhaseFinalizingAfterPluginOperations,
+					Phase:          velerov1api.BackupPhaseFinalizing,
 					Version:        1,
 					FormatVersion:  "1.1.0",
 					StartTimestamp: &timestamp,
@@ -901,7 +901,7 @@ func TestProcessBackupCompletions(t *testing.T) {
 					DefaultVolumesToFsBackup: boolptr.False(),
 				},
 				Status: velerov1api.BackupStatus{
-					Phase:          velerov1api.BackupPhaseFinalizingAfterPluginOperations,
+					Phase:          velerov1api.BackupPhaseFinalizing,
 					Version:        1,
 					FormatVersion:  "1.1.0",
 					StartTimestamp: &timestamp,
--- a/pkg/controller/backup_finalizer_controller.go
+++ b/pkg/controller/backup_finalizer_controller.go
@@ -94,7 +94,7 @@ func (r *backupFinalizerReconciler) Reconcile(ctx context.Context, req ctrl.Requ
 	}

 	switch backup.Status.Phase {
-	case velerov1api.BackupPhaseFinalizingAfterPluginOperations, velerov1api.BackupPhaseFinalizingAfterPluginOperationsPartiallyFailed:
+	case velerov1api.BackupPhaseFinalizing, velerov1api.BackupPhaseFinalizingPartiallyFailed:
 		// only process backups finalizing after  plugin operations are complete
 	default:
 		log.Debug("Backup is not awaiting finalizing, skipping")
@@ -168,11 +168,11 @@ func (r *backupFinalizerReconciler) Reconcile(ctx context.Context, req ctrl.Requ
 	}
 	backupScheduleName := backupRequest.GetLabels()[velerov1api.ScheduleNameLabel]
 	switch backup.Status.Phase {
-	case velerov1api.BackupPhaseFinalizingAfterPluginOperations:
+	case velerov1api.BackupPhaseFinalizing:
 		backup.Status.Phase = velerov1api.BackupPhaseCompleted
 		r.metrics.RegisterBackupSuccess(backupScheduleName)
 		r.metrics.RegisterBackupLastStatus(backupScheduleName, metrics.BackupLastStatusSucc)
-	case velerov1api.BackupPhaseFinalizingAfterPluginOperationsPartiallyFailed:
+	case velerov1api.BackupPhaseFinalizingPartiallyFailed:
 		backup.Status.Phase = velerov1api.BackupPhasePartiallyFailed
 		r.metrics.RegisterBackupPartialFailure(backupScheduleName)
 		r.metrics.RegisterBackupLastStatus(backupScheduleName, metrics.BackupLastStatusFailure)
--- a/pkg/controller/backup_finalizer_controller_test.go
+++ b/pkg/controller/backup_finalizer_controller_test.go
@@ -72,12 +72,12 @@ func TestBackupFinalizerReconcile(t *testing.T) {
 		expectPhase      velerov1api.BackupPhase
 	}{
 		{
-			name: "FinalizingAfterPluginOperations backup is completed",
+			name: "Finalizing backup is completed",
 			backup: builder.ForBackup(velerov1api.DefaultNamespace, "backup-1").
 				StorageLocation("default").
 				ObjectMeta(builder.WithUID("foo")).
 				StartTimestamp(fakeClock.Now()).
-				Phase(velerov1api.BackupPhaseFinalizingAfterPluginOperations).Result(),
+				Phase(velerov1api.BackupPhaseFinalizing).Result(),
 			backupLocation: defaultBackupLocation,
 			expectPhase:    velerov1api.BackupPhaseCompleted,
 			backupOperations: []*itemoperation.BackupOperation{
@@ -91,7 +91,7 @@ func TestBackupFinalizerReconcile(t *testing.T) {
 							Namespace:     "ns-1",
 							Name:          "pod-1",
 						},
-						ItemsToUpdate: []velero.ResourceIdentifier{
+						PostOperationItems: []velero.ResourceIdentifier{
 							{
 								GroupResource: kuberesource.Secrets,
 								Namespace:     "ns-1",
@@ -108,12 +108,12 @@ func TestBackupFinalizerReconcile(t *testing.T) {
 			},
 		},
 		{
-			name: "FinalizingAfterPluginOperationsPartiallyFailed backup is partially failed",
+			name: "FinalizingPartiallyFailed backup is partially failed",
 			backup: builder.ForBackup(velerov1api.DefaultNamespace, "backup-2").
 				StorageLocation("default").
 				ObjectMeta(builder.WithUID("foo")).
 				StartTimestamp(fakeClock.Now()).
-				Phase(velerov1api.BackupPhaseFinalizingAfterPluginOperationsPartiallyFailed).Result(),
+				Phase(velerov1api.BackupPhaseFinalizingPartiallyFailed).Result(),
 			backupLocation: defaultBackupLocation,
 			expectPhase:    velerov1api.BackupPhasePartiallyFailed,
 			backupOperations: []*itemoperation.BackupOperation{
@@ -127,7 +127,7 @@ func TestBackupFinalizerReconcile(t *testing.T) {
 							Namespace:     "ns-2",
 							Name:          "pod-2",
 						},
-						ItemsToUpdate: []velero.ResourceIdentifier{
+						PostOperationItems: []velero.ResourceIdentifier{
 							{
 								GroupResource: kuberesource.Secrets,
 								Namespace:     "ns-2",
--- a/pkg/controller/async_backup_operations_controller.go
+++ b/pkg/controller/async_backup_operations_controller.go
@@ -19,7 +19,6 @@ package controller
 import (
 	"bytes"
 	"context"
-	"sync"
 	"time"

 	"github.com/pkg/errors"
@@ -33,6 +32,7 @@ import (

 	velerov1api "github.com/vmware-tanzu/velero/pkg/apis/velero/v1"
 	"github.com/vmware-tanzu/velero/pkg/itemoperation"
+	"github.com/vmware-tanzu/velero/pkg/itemoperationmap"
 	"github.com/vmware-tanzu/velero/pkg/metrics"
 	"github.com/vmware-tanzu/velero/pkg/persistence"
 	"github.com/vmware-tanzu/velero/pkg/plugin/clientmgmt"
@@ -41,129 +41,46 @@ import (
 )

 const (
-	defaultAsyncBackupOperationsFrequency = 2 * time.Minute
+	defaultBackupOperationsFrequency = 2 * time.Minute
 )

-type operationsForBackup struct {
-	operations         []*itemoperation.BackupOperation
-	changesSinceUpdate bool
-	errsSinceUpdate    []string
-}
-
-// FIXME: remove if handled by backup finalizer controller
-func (o *operationsForBackup) anyItemsToUpdate() bool {
-	for _, op := range o.operations {
-		if len(op.Spec.ItemsToUpdate) > 0 {
-			return true
-		}
-	}
-	return false
-}
-func (in *operationsForBackup) DeepCopy() *operationsForBackup {
-	if in == nil {
-		return nil
-	}
-	out := new(operationsForBackup)
-	in.DeepCopyInto(out)
-	return out
-}
-
-func (in *operationsForBackup) DeepCopyInto(out *operationsForBackup) {
-	*out = *in
-	if in.operations != nil {
-		in, out := &in.operations, &out.operations
-		*out = make([]*itemoperation.BackupOperation, len(*in))
-		for i := range *in {
-			if (*in)[i] != nil {
-				in, out := &(*in)[i], &(*out)[i]
-				*out = new(itemoperation.BackupOperation)
-				(*in).DeepCopyInto(*out)
-			}
-		}
-	}
-	if in.errsSinceUpdate != nil {
-		in, out := &in.errsSinceUpdate, &out.errsSinceUpdate
-		*out = make([]string, len(*in))
-		copy(*out, *in)
-	}
-}
-
-func (o *operationsForBackup) uploadProgress(backupStore persistence.BackupStore, backupName string) error {
-	if len(o.operations) > 0 {
-		var backupItemOperations *bytes.Buffer
-		backupItemOperations, errs := encodeToJSONGzip(o.operations, "backup item operations list")
-		if errs != nil {
-			return errors.Wrap(errs[0], "error encoding item operations json")
-		}
-		err := backupStore.PutBackupItemOperations(backupName, backupItemOperations)
-		if err != nil {
-			return errors.Wrap(err, "error uploading item operations json")
-		}
-	}
-	o.changesSinceUpdate = false
-	o.errsSinceUpdate = nil
-	return nil
-}
-
-type BackupItemOperationsMap struct {
-	operations map[string]*operationsForBackup
-	opsLock    sync.Mutex
-}
-
-// If backup has changes not yet uploaded, upload them now
-func (m *BackupItemOperationsMap) UpdateForBackup(backupStore persistence.BackupStore, backupName string) error {
-	// lock operations map
-	m.opsLock.Lock()
-	defer m.opsLock.Unlock()
-
-	operations, ok := m.operations[backupName]
-	// if operations for this backup aren't found, or if there are no changes
-	// or errors since last update, do nothing
-	if !ok || (!operations.changesSinceUpdate && len(operations.errsSinceUpdate) == 0) {
-		return nil
-	}
-	if err := operations.uploadProgress(backupStore, backupName); err != nil {
-		return err
-	}
-	return nil
-}
-
-type asyncBackupOperationsReconciler struct {
+type backupOperationsReconciler struct {
 	client.Client
 	logger            logrus.FieldLogger
 	clock             clocks.WithTickerAndDelayedExecution
 	frequency         time.Duration
-	itemOperationsMap *BackupItemOperationsMap
+	itemOperationsMap *itemoperationmap.BackupItemOperationsMap
 	newPluginManager  func(logger logrus.FieldLogger) clientmgmt.Manager
 	backupStoreGetter persistence.ObjectBackupStoreGetter
 	metrics           *metrics.ServerMetrics
 }

-func NewAsyncBackupOperationsReconciler(
+func NewBackupOperationsReconciler(
 	logger logrus.FieldLogger,
 	client client.Client,
 	frequency time.Duration,
 	newPluginManager func(logrus.FieldLogger) clientmgmt.Manager,
 	backupStoreGetter persistence.ObjectBackupStoreGetter,
 	metrics *metrics.ServerMetrics,
-) (*asyncBackupOperationsReconciler, *BackupItemOperationsMap) {
-	abor := &asyncBackupOperationsReconciler{
+	itemOperationsMap *itemoperationmap.BackupItemOperationsMap,
+) *backupOperationsReconciler {
+	abor := &backupOperationsReconciler{
 		Client:            client,
 		logger:            logger,
 		clock:             clocks.RealClock{},
 		frequency:         frequency,
-		itemOperationsMap: &BackupItemOperationsMap{operations: make(map[string]*operationsForBackup)},
+		itemOperationsMap: itemOperationsMap,
 		newPluginManager:  newPluginManager,
 		backupStoreGetter: backupStoreGetter,
 		metrics:           metrics,
 	}
 	if abor.frequency <= 0 {
-		abor.frequency = defaultAsyncBackupOperationsFrequency
+		abor.frequency = defaultBackupOperationsFrequency
 	}
-	return abor, abor.itemOperationsMap
+	return abor
 }

-func (c *asyncBackupOperationsReconciler) SetupWithManager(mgr ctrl.Manager) error {
+func (c *backupOperationsReconciler) SetupWithManager(mgr ctrl.Manager) error {
 	s := kube.NewPeriodicalEnqueueSource(c.logger, mgr.GetClient(), &velerov1api.BackupList{}, c.frequency, kube.PeriodicalEnqueueSourceOption{})
 	return ctrl.NewControllerManagedBy(mgr).
 		For(&velerov1api.Backup{}, builder.WithPredicates(kube.FalsePredicate{})).
@@ -175,10 +92,10 @@ func (c *asyncBackupOperationsReconciler) SetupWithManager(mgr ctrl.Manager) err
 // +kubebuilder:rbac:groups=velero.io,resources=backups/status,verbs=get
 // +kubebuilder:rbac:groups=velero.io,resources=backupstoragelocations,verbs=get

-func (c *asyncBackupOperationsReconciler) Reconcile(ctx context.Context, req ctrl.Request) (ctrl.Result, error) {
-	log := c.logger.WithField("async backup operations for backup", req.String())
+func (c *backupOperationsReconciler) Reconcile(ctx context.Context, req ctrl.Request) (ctrl.Result, error) {
+	log := c.logger.WithField("backup operations for backup", req.String())
 	// FIXME: make this log.Debug
-	log.Info("asyncBackupOperationsReconciler getting backup")
+	log.Info("backupOperationsReconciler getting backup")

 	original := &velerov1api.Backup{}
 	if err := c.Get(ctx, req.NamespacedName, original); err != nil {
@@ -201,7 +118,7 @@ func (c *asyncBackupOperationsReconciler) Reconcile(ctx context.Context, req ctr
 	case velerov1api.BackupPhaseWaitingForPluginOperations, velerov1api.BackupPhaseWaitingForPluginOperationsPartiallyFailed:
 		// only process backups waiting for plugin operations to complete
 	default:
-		log.Debug("Backup has no ongoing async plugin operations, skipping")
+		log.Debug("Backup has no ongoing plugin operations, skipping")
 		return ctrl.Result{}, nil
 	}

@@ -211,13 +128,13 @@ func (c *asyncBackupOperationsReconciler) Reconcile(ctx context.Context, req ctr
 		Name:      backup.Spec.StorageLocation,
 	}, loc); err != nil {
 		if apierrors.IsNotFound(err) {
-			log.Warnf("Cannot check progress on async Backup operations because backup storage location %s does not exist; marking backup PartiallyFailed", backup.Spec.StorageLocation)
+			log.Warnf("Cannot check progress on Backup operations because backup storage location %s does not exist; marking backup PartiallyFailed", backup.Spec.StorageLocation)
 			backup.Status.Phase = velerov1api.BackupPhasePartiallyFailed
 		} else {
-			log.Warnf("Cannot check progress on async Backup operations because backup storage location %s could not be retrieved: %s; marking backup PartiallyFailed", backup.Spec.StorageLocation, err.Error())
+			log.Warnf("Cannot check progress on Backup operations because backup storage location %s could not be retrieved: %s; marking backup PartiallyFailed", backup.Spec.StorageLocation, err.Error())
 			backup.Status.Phase = velerov1api.BackupPhasePartiallyFailed
 		}
-		err2 := c.updateBackupAndOperationsJSON(ctx, original, backup, nil, &operationsForBackup{errsSinceUpdate: []string{err.Error()}}, false, false)
+		err2 := c.updateBackupAndOperationsJSON(ctx, original, backup, nil, &itemoperationmap.OperationsForBackup{ErrsSinceUpdate: []string{err.Error()}}, false, false)
 		if err2 != nil {
 			log.WithError(err2).Error("error updating Backup")
 		}
@@ -225,10 +142,10 @@ func (c *asyncBackupOperationsReconciler) Reconcile(ctx context.Context, req ctr
 	}

 	if loc.Spec.AccessMode == velerov1api.BackupStorageLocationAccessModeReadOnly {
-		log.Infof("Cannot check progress on async Backup operations because backup storage location %s is currently in read-only mode; marking backup PartiallyFailed", loc.Name)
+		log.Infof("Cannot check progress on Backup operations because backup storage location %s is currently in read-only mode; marking backup PartiallyFailed", loc.Name)
 		backup.Status.Phase = velerov1api.BackupPhasePartiallyFailed

-		err := c.updateBackupAndOperationsJSON(ctx, original, backup, nil, &operationsForBackup{errsSinceUpdate: []string{"BSL is read-only"}}, false, false)
+		err := c.updateBackupAndOperationsJSON(ctx, original, backup, nil, &itemoperationmap.OperationsForBackup{ErrsSinceUpdate: []string{"BSL is read-only"}}, false, false)
 		if err != nil {
 			log.WithError(err).Error("error updating Backup")
 		}
@@ -242,87 +159,85 @@ func (c *asyncBackupOperationsReconciler) Reconcile(ctx context.Context, req ctr
 		return ctrl.Result{}, errors.Wrap(err, "error getting backup store")
 	}

-	operations, err := c.getOperationsForBackup(backupStore, backup.Name)
+	operations, err := c.itemOperationsMap.GetOperationsForBackup(backupStore, backup.Name)
 	if err != nil {
-		err2 := c.updateBackupAndOperationsJSON(ctx, original, backup, backupStore, &operationsForBackup{errsSinceUpdate: []string{err.Error()}}, false, false)
+		err2 := c.updateBackupAndOperationsJSON(ctx, original, backup, backupStore, &itemoperationmap.OperationsForBackup{ErrsSinceUpdate: []string{err.Error()}}, false, false)
 		if err2 != nil {
 			return ctrl.Result{}, errors.Wrap(err2, "error updating Backup")
 		}
 		return ctrl.Result{}, errors.Wrap(err, "error getting backup operations")
 	}
-	stillInProgress, changes, opsCompleted, opsFailed, errs := getBackupItemOperationProgress(backup, pluginManager, operations.operations)
+	stillInProgress, changes, opsCompleted, opsFailed, errs := getBackupItemOperationProgress(backup, pluginManager, operations.Operations)
 	// if len(errs)>0, need to update backup errors and error log
-	operations.errsSinceUpdate = append(operations.errsSinceUpdate, errs...)
-	backup.Status.Errors += len(operations.errsSinceUpdate)
-	asyncCompletionChanges := false
-	if backup.Status.AsyncBackupItemOperationsCompleted != opsCompleted || backup.Status.AsyncBackupItemOperationsFailed != opsFailed {
-		asyncCompletionChanges = true
-		backup.Status.AsyncBackupItemOperationsCompleted = opsCompleted
-		backup.Status.AsyncBackupItemOperationsFailed = opsFailed
+	operations.ErrsSinceUpdate = append(operations.ErrsSinceUpdate, errs...)
+	backup.Status.Errors += len(operations.ErrsSinceUpdate)
+	completionChanges := false
+	if backup.Status.BackupItemOperationsCompleted != opsCompleted || backup.Status.BackupItemOperationsFailed != opsFailed {
+		completionChanges = true
+		backup.Status.BackupItemOperationsCompleted = opsCompleted
+		backup.Status.BackupItemOperationsFailed = opsFailed
 	}
 	if changes {
-		operations.changesSinceUpdate = true
+		operations.ChangesSinceUpdate = true
 	}

 	// if stillInProgress is false, backup moves to finalize phase and needs update
-	// if operations.errsSinceUpdate is not empty, then backup phase needs to change to
+	// if operations.ErrsSinceUpdate is not empty, then backup phase needs to change to
 	// BackupPhaseWaitingForPluginOperationsPartiallyFailed and needs update
 	// If the only changes are incremental progress, then no write is necessary, progress can remain in memory
 	if !stillInProgress {
-		if len(operations.errsSinceUpdate) > 0 {
+		if len(operations.ErrsSinceUpdate) > 0 {
 			backup.Status.Phase = velerov1api.BackupPhaseWaitingForPluginOperationsPartiallyFailed
 		}
 		if backup.Status.Phase == velerov1api.BackupPhaseWaitingForPluginOperations {
-			log.Infof("Marking backup %s FinalizingAfterPluginOperations", backup.Name)
-			backup.Status.Phase = velerov1api.BackupPhaseFinalizingAfterPluginOperations
+			log.Infof("Marking backup %s Finalizing", backup.Name)
+			backup.Status.Phase = velerov1api.BackupPhaseFinalizing
 		} else {
-			log.Infof("Marking backup %s FinalizingAfterPluginOperationsPartiallyFailed", backup.Name)
-			backup.Status.Phase = velerov1api.BackupPhaseFinalizingAfterPluginOperationsPartiallyFailed
+			log.Infof("Marking backup %s FinalizingPartiallyFailed", backup.Name)
+			backup.Status.Phase = velerov1api.BackupPhaseFinalizingPartiallyFailed
 		}
 	}
-	err = c.updateBackupAndOperationsJSON(ctx, original, backup, backupStore, operations, asyncCompletionChanges, changes)
+	err = c.updateBackupAndOperationsJSON(ctx, original, backup, backupStore, operations, changes, completionChanges)
 	if err != nil {
 		return ctrl.Result{}, errors.Wrap(err, "error updating Backup")
 	}
 	return ctrl.Result{}, nil
 }

-func (c *asyncBackupOperationsReconciler) updateBackupAndOperationsJSON(
+func (c *backupOperationsReconciler) updateBackupAndOperationsJSON(
 	ctx context.Context,
 	original, backup *velerov1api.Backup,
 	backupStore persistence.BackupStore,
-	operations *operationsForBackup,
+	operations *itemoperationmap.OperationsForBackup,
 	changes bool,
-	asyncCompletionChanges bool) error {
+	completionChanges bool) error {

 	backupScheduleName := backup.GetLabels()[velerov1api.ScheduleNameLabel]

-	if len(operations.errsSinceUpdate) > 0 {
+	if len(operations.ErrsSinceUpdate) > 0 {
 		c.metrics.RegisterBackupItemsErrorsGauge(backupScheduleName, backup.Status.Errors)
 		// FIXME: download/upload results once https://github.com/vmware-tanzu/velero/pull/5576 is merged
 	}
 	removeIfComplete := true
 	defer func() {
 		// remove local operations list if complete
-		c.itemOperationsMap.opsLock.Lock()
 		if removeIfComplete && (backup.Status.Phase == velerov1api.BackupPhaseCompleted ||
 			backup.Status.Phase == velerov1api.BackupPhasePartiallyFailed ||
-			backup.Status.Phase == velerov1api.BackupPhaseFinalizingAfterPluginOperations ||
-			backup.Status.Phase == velerov1api.BackupPhaseFinalizingAfterPluginOperationsPartiallyFailed) {
+			backup.Status.Phase == velerov1api.BackupPhaseFinalizing ||
+			backup.Status.Phase == velerov1api.BackupPhaseFinalizingPartiallyFailed) {

-			c.deleteOperationsForBackup(backup.Name)
+			c.itemOperationsMap.DeleteOperationsForBackup(backup.Name)
 		} else if changes {
-			c.putOperationsForBackup(operations, backup.Name)
+			c.itemOperationsMap.PutOperationsForBackup(operations, backup.Name)
 		}
-		c.itemOperationsMap.opsLock.Unlock()
 	}()

 	// update backup and upload progress if errs or complete
-	if len(operations.errsSinceUpdate) > 0 ||
+	if len(operations.ErrsSinceUpdate) > 0 ||
 		backup.Status.Phase == velerov1api.BackupPhaseCompleted ||
 		backup.Status.Phase == velerov1api.BackupPhasePartiallyFailed ||
-		backup.Status.Phase == velerov1api.BackupPhaseFinalizingAfterPluginOperations ||
-		backup.Status.Phase == velerov1api.BackupPhaseFinalizingAfterPluginOperationsPartiallyFailed {
+		backup.Status.Phase == velerov1api.BackupPhaseFinalizing ||
+		backup.Status.Phase == velerov1api.BackupPhaseFinalizingPartiallyFailed {
 		// update file store
 		if backupStore != nil {
 			backupJSON := new(bytes.Buffer)
@@ -335,7 +250,7 @@ func (c *asyncBackupOperationsReconciler) updateBackupAndOperationsJSON(
 				removeIfComplete = false
 				return errors.Wrap(err, "error uploading backup json")
 			}
-			if err := operations.uploadProgress(backupStore, backup.Name); err != nil {
+			if err := c.itemOperationsMap.UploadProgressAndPutOperationsForBackup(backupStore, operations, backup.Name); err != nil {
 				removeIfComplete = false
 				return err
 			}
@@ -346,7 +261,7 @@ func (c *asyncBackupOperationsReconciler) updateBackupAndOperationsJSON(
 			removeIfComplete = false
 			return errors.Wrapf(err, "error updating Backup %s", backup.Name)
 		}
-	} else if asyncCompletionChanges {
+	} else if completionChanges {
 		// If backup is still incomplete and no new errors are found but there are some new operations
 		// completed, patch backup to reflect new completion numbers, but don't upload detailed json file
 		err := c.Client.Patch(ctx, backup, client.MergeFrom(original))
@@ -357,41 +272,6 @@ func (c *asyncBackupOperationsReconciler) updateBackupAndOperationsJSON(
 	return nil
 }

-// returns a deep copy so we can minimize the time the map is locked
-func (c *asyncBackupOperationsReconciler) getOperationsForBackup(
-	backupStore persistence.BackupStore,
-	backupName string) (*operationsForBackup, error) {
-	var err error
-	// lock operations map
-	c.itemOperationsMap.opsLock.Lock()
-	defer c.itemOperationsMap.opsLock.Unlock()
-
-	operations, ok := c.itemOperationsMap.operations[backupName]
-	if !ok || len(operations.operations) == 0 {
-		operations = &operationsForBackup{}
-		operations.operations, err = backupStore.GetBackupItemOperations(backupName)
-		if err == nil {
-			c.itemOperationsMap.operations[backupName] = operations
-		}
-	}
-	return operations.DeepCopy(), err
-}
-
-func (c *asyncBackupOperationsReconciler) putOperationsForBackup(
-	operations *operationsForBackup,
-	backupName string) {
-	if operations != nil {
-		c.itemOperationsMap.operations[backupName] = operations
-	}
-}
-
-func (c *asyncBackupOperationsReconciler) deleteOperationsForBackup(backupName string) {
-	if _, ok := c.itemOperationsMap.operations[backupName]; ok {
-		delete(c.itemOperationsMap.operations, backupName)
-	}
-	return
-}
-
 func getBackupItemOperationProgress(
 	backup *velerov1api.Backup,
 	pluginManager clientmgmt.Manager,
--- a/pkg/controller/async_backup_operations_controller_test.go
+++ b/pkg/controller/async_backup_operations_controller_test.go
@@ -35,6 +35,7 @@ import (
 	velerov1api "github.com/vmware-tanzu/velero/pkg/apis/velero/v1"
 	"github.com/vmware-tanzu/velero/pkg/builder"
 	"github.com/vmware-tanzu/velero/pkg/itemoperation"
+	"github.com/vmware-tanzu/velero/pkg/itemoperationmap"
 	"github.com/vmware-tanzu/velero/pkg/kuberesource"
 	"github.com/vmware-tanzu/velero/pkg/metrics"
 	persistencemocks "github.com/vmware-tanzu/velero/pkg/persistence/mocks"
@@ -51,20 +52,21 @@ var (
 	bia           = &biav2mocks.BackupItemAction{}
 )

-func mockAsyncBackupOperationsReconciler(fakeClient kbclient.Client, fakeClock *testclocks.FakeClock, freq time.Duration) (*asyncBackupOperationsReconciler, *BackupItemOperationsMap) {
-	abor, biaMap := NewAsyncBackupOperationsReconciler(
+func mockBackupOperationsReconciler(fakeClient kbclient.Client, fakeClock *testclocks.FakeClock, freq time.Duration) *backupOperationsReconciler {
+	abor := NewBackupOperationsReconciler(
 		logrus.StandardLogger(),
 		fakeClient,
 		freq,
 		func(logrus.FieldLogger) clientmgmt.Manager { return pluginManager },
 		NewFakeSingleObjectBackupStoreGetter(backupStore),
 		metrics.NewServerMetrics(),
+		itemoperationmap.NewBackupItemOperationsMap(),
 	)
 	abor.clock = fakeClock
-	return abor, biaMap
+	return abor
 }

-func TestAsyncBackupOperationsReconcile(t *testing.T) {
+func TestBackupOperationsReconcile(t *testing.T) {
 	fakeClock := testclocks.NewFakeClock(time.Now())
 	metav1Now := metav1.NewTime(fakeClock.Now())

@@ -81,27 +83,27 @@ func TestAsyncBackupOperationsReconcile(t *testing.T) {
 		expectPhase       velerov1api.BackupPhase
 	}{
 		{
-			name: "WaitingForPluginOperations backup with completed operations is FinalizingAfterPluginOperations",
-			backup: builder.ForBackup(velerov1api.DefaultNamespace, "backup-1").
+			name: "WaitingForPluginOperations backup with completed operations is Finalizing",
+			backup: builder.ForBackup(velerov1api.DefaultNamespace, "backup-11").
 				StorageLocation("default").
 				ItemOperationTimeout(60 * time.Minute).
-				ObjectMeta(builder.WithUID("foo")).
+				ObjectMeta(builder.WithUID("foo-11")).
 				Phase(velerov1api.BackupPhaseWaitingForPluginOperations).Result(),
 			backupLocation:    defaultBackupLocation,
 			operationComplete: true,
-			expectPhase:       velerov1api.BackupPhaseFinalizingAfterPluginOperations,
+			expectPhase:       velerov1api.BackupPhaseFinalizing,
 			backupOperations: []*itemoperation.BackupOperation{
 				{
 					Spec: itemoperation.BackupOperationSpec{
-						BackupName:       "backup-1",
-						BackupUID:        "foo",
-						BackupItemAction: "foo",
+						BackupName:       "backup-11",
+						BackupUID:        "foo-11",
+						BackupItemAction: "foo-11",
 						ResourceIdentifier: velero.ResourceIdentifier{
 							GroupResource: kuberesource.Pods,
 							Namespace:     "ns-1",
 							Name:          "pod-1",
 						},
-						OperationID: "operation-1",
+						OperationID: "operation-11",
 					},
 					Status: itemoperation.OperationStatus{
 						Phase:   itemoperation.OperationPhaseInProgress,
@@ -112,10 +114,10 @@ func TestAsyncBackupOperationsReconcile(t *testing.T) {
 		},
 		{
 			name: "WaitingForPluginOperations backup with incomplete operations is still incomplete",
-			backup: builder.ForBackup(velerov1api.DefaultNamespace, "backup-2").
+			backup: builder.ForBackup(velerov1api.DefaultNamespace, "backup-12").
 				StorageLocation("default").
 				ItemOperationTimeout(60 * time.Minute).
-				ObjectMeta(builder.WithUID("foo")).
+				ObjectMeta(builder.WithUID("foo-12")).
 				Phase(velerov1api.BackupPhaseWaitingForPluginOperations).Result(),
 			backupLocation:    defaultBackupLocation,
 			operationComplete: false,
@@ -123,15 +125,15 @@ func TestAsyncBackupOperationsReconcile(t *testing.T) {
 			backupOperations: []*itemoperation.BackupOperation{
 				{
 					Spec: itemoperation.BackupOperationSpec{
-						BackupName:       "backup-2",
-						BackupUID:        "foo-2",
-						BackupItemAction: "foo-2",
+						BackupName:       "backup-12",
+						BackupUID:        "foo-12",
+						BackupItemAction: "foo-12",
 						ResourceIdentifier: velero.ResourceIdentifier{
 							GroupResource: kuberesource.Pods,
 							Namespace:     "ns-1",
 							Name:          "pod-1",
 						},
-						OperationID: "operation-2",
+						OperationID: "operation-12",
 					},
 					Status: itemoperation.OperationStatus{
 						Phase:   itemoperation.OperationPhaseInProgress,
@@ -141,28 +143,28 @@ func TestAsyncBackupOperationsReconcile(t *testing.T) {
 			},
 		},
 		{
-			name: "WaitingForPluginOperations backup with completed failed operations is FinalizingAfterPluginOperationsPartiallyFailed",
-			backup: builder.ForBackup(velerov1api.DefaultNamespace, "backup-3").
+			name: "WaitingForPluginOperations backup with completed failed operations is FinalizingPartiallyFailed",
+			backup: builder.ForBackup(velerov1api.DefaultNamespace, "backup-13").
 				StorageLocation("default").
 				ItemOperationTimeout(60 * time.Minute).
-				ObjectMeta(builder.WithUID("foo")).
+				ObjectMeta(builder.WithUID("foo-13")).
 				Phase(velerov1api.BackupPhaseWaitingForPluginOperations).Result(),
 			backupLocation:    defaultBackupLocation,
 			operationComplete: true,
 			operationErr:      "failed",
-			expectPhase:       velerov1api.BackupPhaseFinalizingAfterPluginOperationsPartiallyFailed,
+			expectPhase:       velerov1api.BackupPhaseFinalizingPartiallyFailed,
 			backupOperations: []*itemoperation.BackupOperation{
 				{
 					Spec: itemoperation.BackupOperationSpec{
-						BackupName:       "backup-3",
-						BackupUID:        "foo-3",
-						BackupItemAction: "foo-3",
+						BackupName:       "backup-13",
+						BackupUID:        "foo-13",
+						BackupItemAction: "foo-13",
 						ResourceIdentifier: velero.ResourceIdentifier{
 							GroupResource: kuberesource.Pods,
 							Namespace:     "ns-1",
 							Name:          "pod-1",
 						},
-						OperationID: "operation-3",
+						OperationID: "operation-13",
 					},
 					Status: itemoperation.OperationStatus{
 						Phase:   itemoperation.OperationPhaseInProgress,
@@ -172,27 +174,27 @@ func TestAsyncBackupOperationsReconcile(t *testing.T) {
 			},
 		},
 		{
-			name: "WaitingForPluginOperationsPartiallyFailed backup with completed operations is FinalizingAfterPluginOperationsPartiallyFailed",
-			backup: builder.ForBackup(velerov1api.DefaultNamespace, "backup-1").
+			name: "WaitingForPluginOperationsPartiallyFailed backup with completed operations is FinalizingPartiallyFailed",
+			backup: builder.ForBackup(velerov1api.DefaultNamespace, "backup-14").
 				StorageLocation("default").
 				ItemOperationTimeout(60 * time.Minute).
-				ObjectMeta(builder.WithUID("foo")).
+				ObjectMeta(builder.WithUID("foo-14")).
 				Phase(velerov1api.BackupPhaseWaitingForPluginOperationsPartiallyFailed).Result(),
 			backupLocation:    defaultBackupLocation,
 			operationComplete: true,
-			expectPhase:       velerov1api.BackupPhaseFinalizingAfterPluginOperationsPartiallyFailed,
+			expectPhase:       velerov1api.BackupPhaseFinalizingPartiallyFailed,
 			backupOperations: []*itemoperation.BackupOperation{
 				{
 					Spec: itemoperation.BackupOperationSpec{
-						BackupName:       "backup-4",
-						BackupUID:        "foo-4",
-						BackupItemAction: "foo-4",
+						BackupName:       "backup-14",
+						BackupUID:        "foo-14",
+						BackupItemAction: "foo-14",
 						ResourceIdentifier: velero.ResourceIdentifier{
 							GroupResource: kuberesource.Pods,
 							Namespace:     "ns-1",
 							Name:          "pod-1",
 						},
-						OperationID: "operation-4",
+						OperationID: "operation-14",
 					},
 					Status: itemoperation.OperationStatus{
 						Phase:   itemoperation.OperationPhaseInProgress,
@@ -203,10 +205,10 @@ func TestAsyncBackupOperationsReconcile(t *testing.T) {
 		},
 		{
 			name: "WaitingForPluginOperationsPartiallyFailed backup with incomplete operations is still incomplete",
-			backup: builder.ForBackup(velerov1api.DefaultNamespace, "backup-2").
+			backup: builder.ForBackup(velerov1api.DefaultNamespace, "backup-15").
 				StorageLocation("default").
 				ItemOperationTimeout(60 * time.Minute).
-				ObjectMeta(builder.WithUID("foo")).
+				ObjectMeta(builder.WithUID("foo-15")).
 				Phase(velerov1api.BackupPhaseWaitingForPluginOperationsPartiallyFailed).Result(),
 			backupLocation:    defaultBackupLocation,
 			operationComplete: false,
@@ -214,15 +216,15 @@ func TestAsyncBackupOperationsReconcile(t *testing.T) {
 			backupOperations: []*itemoperation.BackupOperation{
 				{
 					Spec: itemoperation.BackupOperationSpec{
-						BackupName:       "backup-5",
-						BackupUID:        "foo-5",
-						BackupItemAction: "foo-5",
+						BackupName:       "backup-15",
+						BackupUID:        "foo-15",
+						BackupItemAction: "foo-15",
 						ResourceIdentifier: velero.ResourceIdentifier{
 							GroupResource: kuberesource.Pods,
 							Namespace:     "ns-1",
 							Name:          "pod-1",
 						},
-						OperationID: "operation-5",
+						OperationID: "operation-15",
 					},
 					Status: itemoperation.OperationStatus{
 						Phase:   itemoperation.OperationPhaseInProgress,
@@ -232,28 +234,28 @@ func TestAsyncBackupOperationsReconcile(t *testing.T) {
 			},
 		},
 		{
-			name: "WaitingForPluginOperationsPartiallyFailed backup with completed failed operations is FinalizingAfterPluginOperationsPartiallyFailed",
-			backup: builder.ForBackup(velerov1api.DefaultNamespace, "backup-3").
+			name: "WaitingForPluginOperationsPartiallyFailed backup with completed failed operations is FinalizingPartiallyFailed",
+			backup: builder.ForBackup(velerov1api.DefaultNamespace, "backup-16").
 				StorageLocation("default").
 				ItemOperationTimeout(60 * time.Minute).
-				ObjectMeta(builder.WithUID("foo")).
+				ObjectMeta(builder.WithUID("foo-16")).
 				Phase(velerov1api.BackupPhaseWaitingForPluginOperationsPartiallyFailed).Result(),
 			backupLocation:    defaultBackupLocation,
 			operationComplete: true,
 			operationErr:      "failed",
-			expectPhase:       velerov1api.BackupPhaseFinalizingAfterPluginOperationsPartiallyFailed,
+			expectPhase:       velerov1api.BackupPhaseFinalizingPartiallyFailed,
 			backupOperations: []*itemoperation.BackupOperation{
 				{
 					Spec: itemoperation.BackupOperationSpec{
-						BackupName:       "backup-6",
-						BackupUID:        "foo-6",
-						BackupItemAction: "foo-6",
+						BackupName:       "backup-16",
+						BackupUID:        "foo-16",
+						BackupItemAction: "foo-16",
 						ResourceIdentifier: velero.ResourceIdentifier{
 							GroupResource: kuberesource.Pods,
 							Namespace:     "ns-1",
 							Name:          "pod-1",
 						},
-						OperationID: "operation-6",
+						OperationID: "operation-16",
 					},
 					Status: itemoperation.OperationStatus{
 						Phase:   itemoperation.OperationPhaseInProgress,
@@ -278,7 +280,7 @@ func TestAsyncBackupOperationsReconcile(t *testing.T) {
 			}

 			fakeClient := velerotest.NewFakeControllerRuntimeClient(t, initObjs...)
-			reconciler, _ := mockAsyncBackupOperationsReconciler(fakeClient, fakeClock, defaultAsyncBackupOperationsFrequency)
+			reconciler := mockBackupOperationsReconciler(fakeClient, fakeClock, defaultBackupOperationsFrequency)
 			pluginManager.On("CleanupClients").Return(nil)
 			backupStore.On("GetBackupItemOperations", test.backup.Name).Return(test.backupOperations, nil)
 			backupStore.On("PutBackupItemOperations", mock.Anything, mock.Anything).Return(nil)
--- a/pkg/controller/backup_sync_controller.go
+++ b/pkg/controller/backup_sync_controller.go
@@ -149,23 +149,15 @@ func (b *backupSyncReconciler) Reconcile(ctx context.Context, req ctrl.Request)
 		}

 		if backup.Status.Phase == velerov1api.BackupPhaseWaitingForPluginOperations ||
-			backup.Status.Phase == velerov1api.BackupPhaseWaitingForPluginOperationsPartiallyFailed {
+			backup.Status.Phase == velerov1api.BackupPhaseWaitingForPluginOperationsPartiallyFailed ||
+			backup.Status.Phase == velerov1api.BackupPhaseFinalizing ||
+			backup.Status.Phase == velerov1api.BackupPhaseFinalizingPartiallyFailed {

 			if backup.Status.Expiration == nil || backup.Status.Expiration.After(time.Now()) {
-				log.Debugf("Skipping non-expired WaitingForPluginOperations backup %v", backup.Name)
+				log.Debugf("Skipping non-expired incomplete backup %v", backup.Name)
 				continue
 			}
-			log.Debug("WaitingForPluginOperations Backup is past expiration, syncing for garbage collection")
-			backup.Status.Phase = velerov1api.BackupPhasePartiallyFailed
-		}
-		if backup.Status.Phase == velerov1api.BackupPhaseFinalizingAfterPluginOperations ||
-			backup.Status.Phase == velerov1api.BackupPhaseFinalizingAfterPluginOperationsPartiallyFailed {
-
-			if backup.Status.Expiration == nil || backup.Status.Expiration.After(time.Now()) {
-				log.Debugf("Skipping non-expired FinalizingAfterPluginOperations backup %v", backup.Name)
-				continue
-			}
-			log.Debug("FinalizingAfterPluginOperations Backup is past expiration, syncing for garbage collection")
+			log.Debugf("%v Backup is past expiration, syncing for garbage collection", backup.Status.Phase)
 			backup.Status.Phase = velerov1api.BackupPhasePartiallyFailed
 		}
 		backup.Namespace = b.namespace
--- a/pkg/controller/backup_sync_controller_test.go
+++ b/pkg/controller/backup_sync_controller_test.go
@@ -216,17 +216,17 @@ var _ = Describe("Backup Sync Reconciler", func() {
 					},
 					{
 						backup: builder.ForBackup("ns-1", "backup-4").
-							Phase(velerov1api.BackupPhaseFinalizingAfterPluginOperations).Result(),
+							Phase(velerov1api.BackupPhaseFinalizing).Result(),
 						backupShouldSkipSync: true,
 					},
 					{
 						backup: builder.ForBackup("ns-1", "backup-5").
-							Phase(velerov1api.BackupPhaseFinalizingAfterPluginOperationsPartiallyFailed).Result(),
+							Phase(velerov1api.BackupPhaseFinalizingPartiallyFailed).Result(),
 						backupShouldSkipSync: true,
 					},
 					{
 						backup: builder.ForBackup("ns-1", "backup-6").
-							Phase(velerov1api.BackupPhaseFinalizingAfterPluginOperations).Result(),
+							Phase(velerov1api.BackupPhaseFinalizing).Result(),
 						podVolumeBackups: []*velerov1api.PodVolumeBackup{
 							builder.ForPodVolumeBackup("ns-1", "pvb-2").Result(),
 						},
@@ -262,19 +262,19 @@ var _ = Describe("Backup Sync Reconciler", func() {
 					},
 					{
 						backup: builder.ForBackup("ns-1", "backup-4").
-							Phase(velerov1api.BackupPhaseFinalizingAfterPluginOperations).
+							Phase(velerov1api.BackupPhaseFinalizing).
 							Expiration(fakeClock.Now().Add(-time.Hour)).Result(),
 						backupShouldSkipSync: true,
 					},
 					{
 						backup: builder.ForBackup("ns-1", "backup-5").
-							Phase(velerov1api.BackupPhaseFinalizingAfterPluginOperationsPartiallyFailed).
+							Phase(velerov1api.BackupPhaseFinalizingPartiallyFailed).
 							Expiration(fakeClock.Now().Add(-time.Hour)).Result(),
 						backupShouldSkipSync: true,
 					},
 					{
 						backup: builder.ForBackup("ns-1", "backup-6").
-							Phase(velerov1api.BackupPhaseFinalizingAfterPluginOperations).
+							Phase(velerov1api.BackupPhaseFinalizing).
 							Expiration(fakeClock.Now().Add(-time.Hour)).Result(),
 						podVolumeBackups: []*velerov1api.PodVolumeBackup{
 							builder.ForPodVolumeBackup("ns-1", "pvb-2").Result(),
--- a/pkg/controller/constants.go
+++ b/pkg/controller/constants.go
@@ -17,7 +17,7 @@ limitations under the License.
 package controller

 const (
-	AsyncBackupOperations = "async-backup-operations"
+	BackupOperations      = "backup-operations"
 	Backup                = "backup"
 	BackupDeletion        = "backup-deletion"
 	BackupFinalizer       = "backup-finalizer"
@@ -35,7 +35,7 @@ const (

 // DisableableControllers is a list of controllers that can be disabled
 var DisableableControllers = []string{
-	AsyncBackupOperations,
+	BackupOperations,
 	Backup,
 	BackupDeletion,
 	BackupFinalizer,
--- a/pkg/controller/download_request_controller.go
+++ b/pkg/controller/download_request_controller.go
@@ -28,6 +28,7 @@ import (
 	kbclient "sigs.k8s.io/controller-runtime/pkg/client"

 	velerov1api "github.com/vmware-tanzu/velero/pkg/apis/velero/v1"
+	"github.com/vmware-tanzu/velero/pkg/itemoperationmap"
 	"github.com/vmware-tanzu/velero/pkg/persistence"
 	"github.com/vmware-tanzu/velero/pkg/plugin/clientmgmt"
 )
@@ -42,7 +43,7 @@ type downloadRequestReconciler struct {
 	backupStoreGetter persistence.ObjectBackupStoreGetter

 	// used to force update of async backup item operations before processing download request
-	backupItemOperationsMap *BackupItemOperationsMap
+	backupItemOperationsMap *itemoperationmap.BackupItemOperationsMap

 	log logrus.FieldLogger
 }
@@ -54,7 +55,7 @@ func NewDownloadRequestReconciler(
 	newPluginManager func(logrus.FieldLogger) clientmgmt.Manager,
 	backupStoreGetter persistence.ObjectBackupStoreGetter,
 	log logrus.FieldLogger,
-	backupItemOperationsMap *BackupItemOperationsMap,
+	backupItemOperationsMap *itemoperationmap.BackupItemOperationsMap,
 ) *downloadRequestReconciler {
 	return &downloadRequestReconciler{
 		client:                  client,
@@ -164,8 +165,8 @@ func (r *downloadRequestReconciler) Reconcile(ctx context.Context, req ctrl.Requ
 			return ctrl.Result{}, errors.WithStack(err)
 		}

-		// If this is a request for backup item operations, force update of in-memory operations that
-		// are not yet uploaded
+		// If this is a request for backup item operations, force upload of in-memory operations that
+		// are not yet uploaded (if there are any)
 		if downloadRequest.Spec.Target.Kind == velerov1api.DownloadTargetKindBackupItemOperations &&
 			r.backupItemOperationsMap != nil {
 			// ignore errors here. If we can't upload anything here, process the download as usual
--- a/pkg/itemoperation/backup_operation.go
+++ b/pkg/itemoperation/backup_operation.go
@@ -61,8 +61,8 @@ type BackupOperationSpec struct {
 	// OperationID returned by the BIA plugin
 	OperationID string "json:operationID"

-	// Items needing update after all async operations have completed
-	ItemsToUpdate []velero.ResourceIdentifier "json:itemsToUpdate"
+	// Items needing to be added to the backup after all async operations have completed
+	PostOperationItems []velero.ResourceIdentifier "json:postOperationItems"
 }

 func (in *BackupOperationSpec) DeepCopy() *BackupOperationSpec {
@@ -77,8 +77,8 @@ func (in *BackupOperationSpec) DeepCopy() *BackupOperationSpec {
 func (in *BackupOperationSpec) DeepCopyInto(out *BackupOperationSpec) {
 	*out = *in
 	in.ResourceIdentifier.DeepCopyInto(&out.ResourceIdentifier)
-	if in.ItemsToUpdate != nil {
-		in, out := &in.ItemsToUpdate, &out.ItemsToUpdate
+	if in.PostOperationItems != nil {
+		in, out := &in.PostOperationItems, &out.PostOperationItems
 		*out = make([]velero.ResourceIdentifier, len(*in))
 		for i := range *in {
 			(*in)[i].DeepCopyInto(&(*out)[i])
--- a/pkg/itemoperationmap/backup_operation_map.go
+++ b/pkg/itemoperationmap/backup_operation_map.go
@@ -0,0 +1,170 @@
+/*
+Copyright the Velero contributors.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package itemoperationmap
+
+import (
+	"bytes"
+	"sync"
+
+	"github.com/pkg/errors"
+
+	"github.com/vmware-tanzu/velero/pkg/itemoperation"
+	"github.com/vmware-tanzu/velero/pkg/persistence"
+	"github.com/vmware-tanzu/velero/pkg/util/encode"
+)
+
+type BackupItemOperationsMap struct {
+	opsMap  map[string]*OperationsForBackup
+	opsLock sync.Mutex
+}
+
+// Returns a pointer to a new BackupItemOperationsMap
+func NewBackupItemOperationsMap() *BackupItemOperationsMap {
+	return &BackupItemOperationsMap{opsMap: make(map[string]*OperationsForBackup)}
+}
+
+// returns a deep copy so we can minimize the time the map is locked
+func (m *BackupItemOperationsMap) GetOperationsForBackup(
+	backupStore persistence.BackupStore,
+	backupName string) (*OperationsForBackup, error) {
+	var err error
+	// lock operations map
+	m.opsLock.Lock()
+	defer m.opsLock.Unlock()
+
+	operations, ok := m.opsMap[backupName]
+	if !ok || len(operations.Operations) == 0 {
+		operations = &OperationsForBackup{}
+		operations.Operations, err = backupStore.GetBackupItemOperations(backupName)
+		if err == nil {
+			m.opsMap[backupName] = operations
+		}
+	}
+	return operations.DeepCopy(), err
+}
+
+func (m *BackupItemOperationsMap) PutOperationsForBackup(
+	operations *OperationsForBackup,
+	backupName string) {
+	// lock operations map
+	m.opsLock.Lock()
+	defer m.opsLock.Unlock()
+	if operations != nil {
+		m.opsMap[backupName] = operations
+	}
+}
+
+func (m *BackupItemOperationsMap) DeleteOperationsForBackup(backupName string) {
+	// lock operations map
+	m.opsLock.Lock()
+	defer m.opsLock.Unlock()
+	if _, ok := m.opsMap[backupName]; ok {
+		delete(m.opsMap, backupName)
+	}
+	return
+}
+
+// UploadProgressAndPutOperationsForBackup will upload the item operations for this backup to
+// the object store and update the map for this backup with the modified operations
+func (m *BackupItemOperationsMap) UploadProgressAndPutOperationsForBackup(
+	backupStore persistence.BackupStore,
+	operations *OperationsForBackup,
+	backupName string) error {
+
+	m.opsLock.Lock()
+	defer m.opsLock.Unlock()
+
+	if operations == nil {
+		return errors.New("nil operations passed in")
+	}
+	if err := operations.uploadProgress(backupStore, backupName); err != nil {
+		return err
+	}
+	m.opsMap[backupName] = operations
+	return nil
+}
+
+// UpdateForBackup will upload the item operations for this backup to
+// the object store, if it has changes not yet uploaded
+func (m *BackupItemOperationsMap) UpdateForBackup(backupStore persistence.BackupStore, backupName string) error {
+	// lock operations map
+	m.opsLock.Lock()
+	defer m.opsLock.Unlock()
+
+	operations, ok := m.opsMap[backupName]
+	// if operations for this backup aren't found, or if there are no changes
+	// or errors since last update, do nothing
+	if !ok || (!operations.ChangesSinceUpdate && len(operations.ErrsSinceUpdate) == 0) {
+		return nil
+	}
+	if err := operations.uploadProgress(backupStore, backupName); err != nil {
+		return err
+	}
+	return nil
+}
+
+type OperationsForBackup struct {
+	Operations         []*itemoperation.BackupOperation
+	ChangesSinceUpdate bool
+	ErrsSinceUpdate    []string
+}
+
+func (in *OperationsForBackup) DeepCopy() *OperationsForBackup {
+	if in == nil {
+		return nil
+	}
+	out := new(OperationsForBackup)
+	in.DeepCopyInto(out)
+	return out
+}
+
+func (in *OperationsForBackup) DeepCopyInto(out *OperationsForBackup) {
+	*out = *in
+	if in.Operations != nil {
+		in, out := &in.Operations, &out.Operations
+		*out = make([]*itemoperation.BackupOperation, len(*in))
+		for i := range *in {
+			if (*in)[i] != nil {
+				in, out := &(*in)[i], &(*out)[i]
+				*out = new(itemoperation.BackupOperation)
+				(*in).DeepCopyInto(*out)
+			}
+		}
+	}
+	if in.ErrsSinceUpdate != nil {
+		in, out := &in.ErrsSinceUpdate, &out.ErrsSinceUpdate
+		*out = make([]string, len(*in))
+		copy(*out, *in)
+	}
+}
+
+func (o *OperationsForBackup) uploadProgress(backupStore persistence.BackupStore, backupName string) error {
+	if len(o.Operations) > 0 {
+		var backupItemOperations *bytes.Buffer
+		backupItemOperations, errs := encode.EncodeToJSONGzip(o.Operations, "backup item operations list")
+		if errs != nil {
+			return errors.Wrap(errs[0], "error encoding item operations json")
+		}
+		err := backupStore.PutBackupItemOperations(backupName, backupItemOperations)
+		if err != nil {
+			return errors.Wrap(err, "error uploading item operations json")
+		}
+	}
+	o.ChangesSinceUpdate = false
+	o.ErrsSinceUpdate = nil
+	return nil
+}
--- a/pkg/plugin/framework/backupitemaction/v2/backup_item_action_client.go
+++ b/pkg/plugin/framework/backupitemaction/v2/backup_item_action_client.go
@@ -118,9 +118,9 @@ func (c *BackupItemActionGRPCClient) Execute(item runtime.Unstructured, backup *
 		additionalItems = append(additionalItems, newItem)
 	}

-	var itemsToUpdate []velero.ResourceIdentifier
+	var postOperationItems []velero.ResourceIdentifier

-	for _, itm := range res.ItemsToUpdate {
+	for _, itm := range res.PostOperationItems {
 		newItem := velero.ResourceIdentifier{
 			GroupResource: schema.GroupResource{
 				Group:    itm.Group,
@@ -130,10 +130,10 @@ func (c *BackupItemActionGRPCClient) Execute(item runtime.Unstructured, backup *
 			Name:      itm.Name,
 		}

-		itemsToUpdate = append(itemsToUpdate, newItem)
+		postOperationItems = append(postOperationItems, newItem)
 	}

-	return &updatedItem, additionalItems, res.OperationID, itemsToUpdate, nil
+	return &updatedItem, additionalItems, res.OperationID, postOperationItems, nil
 }

 func (c *BackupItemActionGRPCClient) Progress(operationID string, backup *api.Backup) (velero.OperationProgress, error) {
--- a/pkg/plugin/framework/backupitemaction/v2/backup_item_action_server.go
+++ b/pkg/plugin/framework/backupitemaction/v2/backup_item_action_server.go
@@ -107,7 +107,7 @@ func (s *BackupItemActionGRPCServer) Execute(
 		return nil, common.NewGRPCError(errors.WithStack(err))
 	}

-	updatedItem, additionalItems, operationID, itemsToUpdate, err := impl.Execute(&item, &backup)
+	updatedItem, additionalItems, operationID, postOperationItems, err := impl.Execute(&item, &backup)
 	if err != nil {
 		return nil, common.NewGRPCError(err)
 	}
@@ -132,8 +132,8 @@ func (s *BackupItemActionGRPCServer) Execute(
 	for _, item := range additionalItems {
 		res.AdditionalItems = append(res.AdditionalItems, backupResourceIdentifierToProto(item))
 	}
-	for _, item := range itemsToUpdate {
-		res.ItemsToUpdate = append(res.ItemsToUpdate, backupResourceIdentifierToProto(item))
+	for _, item := range postOperationItems {
+		res.PostOperationItems = append(res.PostOperationItems, backupResourceIdentifierToProto(item))
 	}

 	return res, nil
--- a/pkg/plugin/framework/backupitemaction/v2/backup_item_action_test.go
+++ b/pkg/plugin/framework/backupitemaction/v2/backup_item_action_test.go
@@ -91,16 +91,16 @@ func TestBackupItemActionGRPCServerExecute(t *testing.T) {
 	require.NoError(t, err)

 	tests := []struct {
-		name                string
-		backup              []byte
-		item                []byte
-		implUpdatedItem     runtime.Unstructured
-		implAdditionalItems []velero.ResourceIdentifier
-		implOperationID     string
-		implItemsToUpdate   []velero.ResourceIdentifier
-		implError           error
-		expectError         bool
-		skipMock            bool
+		name                   string
+		backup                 []byte
+		item                   []byte
+		implUpdatedItem        runtime.Unstructured
+		implAdditionalItems    []velero.ResourceIdentifier
+		implOperationID        string
+		implPostOperationItems []velero.ResourceIdentifier
+		implError              error
+		expectError            bool
+		skipMock               bool
 	}{
 		{
 			name:        "error unmarshaling item",
@@ -154,7 +154,7 @@ func TestBackupItemActionGRPCServerExecute(t *testing.T) {
 			defer itemAction.AssertExpectations(t)

 			if !test.skipMock {
-				itemAction.On("Execute", &validItemObject, &validBackupObject).Return(test.implUpdatedItem, test.implAdditionalItems, test.implOperationID, test.implItemsToUpdate, test.implError)
+				itemAction.On("Execute", &validItemObject, &validBackupObject).Return(test.implUpdatedItem, test.implAdditionalItems, test.implOperationID, test.implPostOperationItems, test.implError)
 			}

 			s := &BackupItemActionGRPCServer{mux: &common.ServerMux{
--- a/pkg/plugin/generated/backupitemaction/v2/BackupItemAction.pb.go
+++ b/pkg/plugin/generated/backupitemaction/v2/BackupItemAction.pb.go
@@ -99,10 +99,10 @@ type ExecuteResponse struct {
 	sizeCache     protoimpl.SizeCache
 	unknownFields protoimpl.UnknownFields

-	Item            []byte                          `protobuf:"bytes,1,opt,name=item,proto3" json:"item,omitempty"`
-	AdditionalItems []*generated.ResourceIdentifier `protobuf:"bytes,2,rep,name=additionalItems,proto3" json:"additionalItems,omitempty"`
-	OperationID     string                          `protobuf:"bytes,3,opt,name=operationID,proto3" json:"operationID,omitempty"`
-	ItemsToUpdate   []*generated.ResourceIdentifier `protobuf:"bytes,4,rep,name=itemsToUpdate,proto3" json:"itemsToUpdate,omitempty"`
+	Item               []byte                          `protobuf:"bytes,1,opt,name=item,proto3" json:"item,omitempty"`
+	AdditionalItems    []*generated.ResourceIdentifier `protobuf:"bytes,2,rep,name=additionalItems,proto3" json:"additionalItems,omitempty"`
+	OperationID        string                          `protobuf:"bytes,3,opt,name=operationID,proto3" json:"operationID,omitempty"`
+	PostOperationItems []*generated.ResourceIdentifier `protobuf:"bytes,4,rep,name=postOperationItems,proto3" json:"postOperationItems,omitempty"`
 }

 func (x *ExecuteResponse) Reset() {
@@ -158,9 +158,9 @@ func (x *ExecuteResponse) GetOperationID() string {
 	return ""
 }

-func (x *ExecuteResponse) GetItemsToUpdate() []*generated.ResourceIdentifier {
+func (x *ExecuteResponse) GetPostOperationItems() []*generated.ResourceIdentifier {
 	if x != nil {
-		return x.ItemsToUpdate
+		return x.PostOperationItems
 	}
 	return nil
 }
@@ -446,7 +446,7 @@ var file_backupitemaction_v2_BackupItemAction_proto_rawDesc = []byte{
 	0x6c, 0x75, 0x67, 0x69, 0x6e, 0x12, 0x12, 0x0a, 0x04, 0x69, 0x74, 0x65, 0x6d, 0x18, 0x02, 0x20,
 	0x01, 0x28, 0x0c, 0x52, 0x04, 0x69, 0x74, 0x65, 0x6d, 0x12, 0x16, 0x0a, 0x06, 0x62, 0x61, 0x63,
 	0x6b, 0x75, 0x70, 0x18, 0x03, 0x20, 0x01, 0x28, 0x0c, 0x52, 0x06, 0x62, 0x61, 0x63, 0x6b, 0x75,
-	0x70, 0x22, 0xd5, 0x01, 0x0a, 0x0f, 0x45, 0x78, 0x65, 0x63, 0x75, 0x74, 0x65, 0x52, 0x65, 0x73,
+	0x70, 0x22, 0xdf, 0x01, 0x0a, 0x0f, 0x45, 0x78, 0x65, 0x63, 0x75, 0x74, 0x65, 0x52, 0x65, 0x73,
 	0x70, 0x6f, 0x6e, 0x73, 0x65, 0x12, 0x12, 0x0a, 0x04, 0x69, 0x74, 0x65, 0x6d, 0x18, 0x01, 0x20,
 	0x01, 0x28, 0x0c, 0x52, 0x04, 0x69, 0x74, 0x65, 0x6d, 0x12, 0x47, 0x0a, 0x0f, 0x61, 0x64, 0x64,
 	0x69, 0x74, 0x69, 0x6f, 0x6e, 0x61, 0x6c, 0x49, 0x74, 0x65, 0x6d, 0x73, 0x18, 0x02, 0x20, 0x03,
@@ -455,67 +455,68 @@ var file_backupitemaction_v2_BackupItemAction_proto_rawDesc = []byte{
 	0x72, 0x52, 0x0f, 0x61, 0x64, 0x64, 0x69, 0x74, 0x69, 0x6f, 0x6e, 0x61, 0x6c, 0x49, 0x74, 0x65,
 	0x6d, 0x73, 0x12, 0x20, 0x0a, 0x0b, 0x6f, 0x70, 0x65, 0x72, 0x61, 0x74, 0x69, 0x6f, 0x6e, 0x49,
 	0x44, 0x18, 0x03, 0x20, 0x01, 0x28, 0x09, 0x52, 0x0b, 0x6f, 0x70, 0x65, 0x72, 0x61, 0x74, 0x69,
-	0x6f, 0x6e, 0x49, 0x44, 0x12, 0x43, 0x0a, 0x0d, 0x69, 0x74, 0x65, 0x6d, 0x73, 0x54, 0x6f, 0x55,
-	0x70, 0x64, 0x61, 0x74, 0x65, 0x18, 0x04, 0x20, 0x03, 0x28, 0x0b, 0x32, 0x1d, 0x2e, 0x67, 0x65,
-	0x6e, 0x65, 0x72, 0x61, 0x74, 0x65, 0x64, 0x2e, 0x52, 0x65, 0x73, 0x6f, 0x75, 0x72, 0x63, 0x65,
-	0x49, 0x64, 0x65, 0x6e, 0x74, 0x69, 0x66, 0x69, 0x65, 0x72, 0x52, 0x0d, 0x69, 0x74, 0x65, 0x6d,
-	0x73, 0x54, 0x6f, 0x55, 0x70, 0x64, 0x61, 0x74, 0x65, 0x22, 0x3a, 0x0a, 0x20, 0x42, 0x61, 0x63,
-	0x6b, 0x75, 0x70, 0x49, 0x74, 0x65, 0x6d, 0x41, 0x63, 0x74, 0x69, 0x6f, 0x6e, 0x41, 0x70, 0x70,
-	0x6c, 0x69, 0x65, 0x73, 0x54, 0x6f, 0x52, 0x65, 0x71, 0x75, 0x65, 0x73, 0x74, 0x12, 0x16, 0x0a,
-	0x06, 0x70, 0x6c, 0x75, 0x67, 0x69, 0x6e, 0x18, 0x01, 0x20, 0x01, 0x28, 0x09, 0x52, 0x06, 0x70,
-	0x6c, 0x75, 0x67, 0x69, 0x6e, 0x22, 0x6c, 0x0a, 0x21, 0x42, 0x61, 0x63, 0x6b, 0x75, 0x70, 0x49,
-	0x74, 0x65, 0x6d, 0x41, 0x63, 0x74, 0x69, 0x6f, 0x6e, 0x41, 0x70, 0x70, 0x6c, 0x69, 0x65, 0x73,
-	0x54, 0x6f, 0x52, 0x65, 0x73, 0x70, 0x6f, 0x6e, 0x73, 0x65, 0x12, 0x47, 0x0a, 0x10, 0x52, 0x65,
-	0x73, 0x6f, 0x75, 0x72, 0x63, 0x65, 0x53, 0x65, 0x6c, 0x65, 0x63, 0x74, 0x6f, 0x72, 0x18, 0x01,
-	0x20, 0x01, 0x28, 0x0b, 0x32, 0x1b, 0x2e, 0x67, 0x65, 0x6e, 0x65, 0x72, 0x61, 0x74, 0x65, 0x64,
-	0x2e, 0x52, 0x65, 0x73, 0x6f, 0x75, 0x72, 0x63, 0x65, 0x53, 0x65, 0x6c, 0x65, 0x63, 0x74, 0x6f,
-	0x72, 0x52, 0x10, 0x52, 0x65, 0x73, 0x6f, 0x75, 0x72, 0x63, 0x65, 0x53, 0x65, 0x6c, 0x65, 0x63,
-	0x74, 0x6f, 0x72, 0x22, 0x73, 0x0a, 0x1f, 0x42, 0x61, 0x63, 0x6b, 0x75, 0x70, 0x49, 0x74, 0x65,
-	0x6d, 0x41, 0x63, 0x74, 0x69, 0x6f, 0x6e, 0x50, 0x72, 0x6f, 0x67, 0x72, 0x65, 0x73, 0x73, 0x52,
-	0x65, 0x71, 0x75, 0x65, 0x73, 0x74, 0x12, 0x16, 0x0a, 0x06, 0x70, 0x6c, 0x75, 0x67, 0x69, 0x6e,
-	0x18, 0x01, 0x20, 0x01, 0x28, 0x09, 0x52, 0x06, 0x70, 0x6c, 0x75, 0x67, 0x69, 0x6e, 0x12, 0x20,
-	0x0a, 0x0b, 0x6f, 0x70, 0x65, 0x72, 0x61, 0x74, 0x69, 0x6f, 0x6e, 0x49, 0x44, 0x18, 0x02, 0x20,
-	0x01, 0x28, 0x09, 0x52, 0x0b, 0x6f, 0x70, 0x65, 0x72, 0x61, 0x74, 0x69, 0x6f, 0x6e, 0x49, 0x44,
-	0x12, 0x16, 0x0a, 0x06, 0x62, 0x61, 0x63, 0x6b, 0x75, 0x70, 0x18, 0x03, 0x20, 0x01, 0x28, 0x0c,
-	0x52, 0x06, 0x62, 0x61, 0x63, 0x6b, 0x75, 0x70, 0x22, 0x5c, 0x0a, 0x20, 0x42, 0x61, 0x63, 0x6b,
-	0x75, 0x70, 0x49, 0x74, 0x65, 0x6d, 0x41, 0x63, 0x74, 0x69, 0x6f, 0x6e, 0x50, 0x72, 0x6f, 0x67,
-	0x72, 0x65, 0x73, 0x73, 0x52, 0x65, 0x73, 0x70, 0x6f, 0x6e, 0x73, 0x65, 0x12, 0x38, 0x0a, 0x08,
-	0x70, 0x72, 0x6f, 0x67, 0x72, 0x65, 0x73, 0x73, 0x18, 0x01, 0x20, 0x01, 0x28, 0x0b, 0x32, 0x1c,
-	0x2e, 0x67, 0x65, 0x6e, 0x65, 0x72, 0x61, 0x74, 0x65, 0x64, 0x2e, 0x4f, 0x70, 0x65, 0x72, 0x61,
-	0x74, 0x69, 0x6f, 0x6e, 0x50, 0x72, 0x6f, 0x67, 0x72, 0x65, 0x73, 0x73, 0x52, 0x08, 0x70, 0x72,
-	0x6f, 0x67, 0x72, 0x65, 0x73, 0x73, 0x22, 0x71, 0x0a, 0x1d, 0x42, 0x61, 0x63, 0x6b, 0x75, 0x70,
-	0x49, 0x74, 0x65, 0x6d, 0x41, 0x63, 0x74, 0x69, 0x6f, 0x6e, 0x43, 0x61, 0x6e, 0x63, 0x65, 0x6c,
-	0x52, 0x65, 0x71, 0x75, 0x65, 0x73, 0x74, 0x12, 0x16, 0x0a, 0x06, 0x70, 0x6c, 0x75, 0x67, 0x69,
-	0x6e, 0x18, 0x01, 0x20, 0x01, 0x28, 0x09, 0x52, 0x06, 0x70, 0x6c, 0x75, 0x67, 0x69, 0x6e, 0x12,
-	0x20, 0x0a, 0x0b, 0x6f, 0x70, 0x65, 0x72, 0x61, 0x74, 0x69, 0x6f, 0x6e, 0x49, 0x44, 0x18, 0x02,
-	0x20, 0x01, 0x28, 0x09, 0x52, 0x0b, 0x6f, 0x70, 0x65, 0x72, 0x61, 0x74, 0x69, 0x6f, 0x6e, 0x49,
-	0x44, 0x12, 0x16, 0x0a, 0x06, 0x62, 0x61, 0x63, 0x6b, 0x75, 0x70, 0x18, 0x03, 0x20, 0x01, 0x28,
-	0x0c, 0x52, 0x06, 0x62, 0x61, 0x63, 0x6b, 0x75, 0x70, 0x32, 0xbc, 0x02, 0x0a, 0x10, 0x42, 0x61,
-	0x63, 0x6b, 0x75, 0x70, 0x49, 0x74, 0x65, 0x6d, 0x41, 0x63, 0x74, 0x69, 0x6f, 0x6e, 0x12, 0x58,
-	0x0a, 0x09, 0x41, 0x70, 0x70, 0x6c, 0x69, 0x65, 0x73, 0x54, 0x6f, 0x12, 0x24, 0x2e, 0x76, 0x32,
-	0x2e, 0x42, 0x61, 0x63, 0x6b, 0x75, 0x70, 0x49, 0x74, 0x65, 0x6d, 0x41, 0x63, 0x74, 0x69, 0x6f,
-	0x6e, 0x41, 0x70, 0x70, 0x6c, 0x69, 0x65, 0x73, 0x54, 0x6f, 0x52, 0x65, 0x71, 0x75, 0x65, 0x73,
-	0x74, 0x1a, 0x25, 0x2e, 0x76, 0x32, 0x2e, 0x42, 0x61, 0x63, 0x6b, 0x75, 0x70, 0x49, 0x74, 0x65,
+	0x6f, 0x6e, 0x49, 0x44, 0x12, 0x4d, 0x0a, 0x12, 0x70, 0x6f, 0x73, 0x74, 0x4f, 0x70, 0x65, 0x72,
+	0x61, 0x74, 0x69, 0x6f, 0x6e, 0x49, 0x74, 0x65, 0x6d, 0x73, 0x18, 0x04, 0x20, 0x03, 0x28, 0x0b,
+	0x32, 0x1d, 0x2e, 0x67, 0x65, 0x6e, 0x65, 0x72, 0x61, 0x74, 0x65, 0x64, 0x2e, 0x52, 0x65, 0x73,
+	0x6f, 0x75, 0x72, 0x63, 0x65, 0x49, 0x64, 0x65, 0x6e, 0x74, 0x69, 0x66, 0x69, 0x65, 0x72, 0x52,
+	0x12, 0x70, 0x6f, 0x73, 0x74, 0x4f, 0x70, 0x65, 0x72, 0x61, 0x74, 0x69, 0x6f, 0x6e, 0x49, 0x74,
+	0x65, 0x6d, 0x73, 0x22, 0x3a, 0x0a, 0x20, 0x42, 0x61, 0x63, 0x6b, 0x75, 0x70, 0x49, 0x74, 0x65,
 	0x6d, 0x41, 0x63, 0x74, 0x69, 0x6f, 0x6e, 0x41, 0x70, 0x70, 0x6c, 0x69, 0x65, 0x73, 0x54, 0x6f,
-	0x52, 0x65, 0x73, 0x70, 0x6f, 0x6e, 0x73, 0x65, 0x12, 0x32, 0x0a, 0x07, 0x45, 0x78, 0x65, 0x63,
-	0x75, 0x74, 0x65, 0x12, 0x12, 0x2e, 0x76, 0x32, 0x2e, 0x45, 0x78, 0x65, 0x63, 0x75, 0x74, 0x65,
-	0x52, 0x65, 0x71, 0x75, 0x65, 0x73, 0x74, 0x1a, 0x13, 0x2e, 0x76, 0x32, 0x2e, 0x45, 0x78, 0x65,
-	0x63, 0x75, 0x74, 0x65, 0x52, 0x65, 0x73, 0x70, 0x6f, 0x6e, 0x73, 0x65, 0x12, 0x55, 0x0a, 0x08,
-	0x50, 0x72, 0x6f, 0x67, 0x72, 0x65, 0x73, 0x73, 0x12, 0x23, 0x2e, 0x76, 0x32, 0x2e, 0x42, 0x61,
-	0x63, 0x6b, 0x75, 0x70, 0x49, 0x74, 0x65, 0x6d, 0x41, 0x63, 0x74, 0x69, 0x6f, 0x6e, 0x50, 0x72,
-	0x6f, 0x67, 0x72, 0x65, 0x73, 0x73, 0x52, 0x65, 0x71, 0x75, 0x65, 0x73, 0x74, 0x1a, 0x24, 0x2e,
-	0x76, 0x32, 0x2e, 0x42, 0x61, 0x63, 0x6b, 0x75, 0x70, 0x49, 0x74, 0x65, 0x6d, 0x41, 0x63, 0x74,
-	0x69, 0x6f, 0x6e, 0x50, 0x72, 0x6f, 0x67, 0x72, 0x65, 0x73, 0x73, 0x52, 0x65, 0x73, 0x70, 0x6f,
-	0x6e, 0x73, 0x65, 0x12, 0x43, 0x0a, 0x06, 0x43, 0x61, 0x6e, 0x63, 0x65, 0x6c, 0x12, 0x21, 0x2e,
-	0x76, 0x32, 0x2e, 0x42, 0x61, 0x63, 0x6b, 0x75, 0x70, 0x49, 0x74, 0x65, 0x6d, 0x41, 0x63, 0x74,
-	0x69, 0x6f, 0x6e, 0x43, 0x61, 0x6e, 0x63, 0x65, 0x6c, 0x52, 0x65, 0x71, 0x75, 0x65, 0x73, 0x74,
-	0x1a, 0x16, 0x2e, 0x67, 0x6f, 0x6f, 0x67, 0x6c, 0x65, 0x2e, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x62,
-	0x75, 0x66, 0x2e, 0x45, 0x6d, 0x70, 0x74, 0x79, 0x42, 0x49, 0x5a, 0x47, 0x67, 0x69, 0x74, 0x68,
-	0x75, 0x62, 0x2e, 0x63, 0x6f, 0x6d, 0x2f, 0x76, 0x6d, 0x77, 0x61, 0x72, 0x65, 0x2d, 0x74, 0x61,
-	0x6e, 0x7a, 0x75, 0x2f, 0x76, 0x65, 0x6c, 0x65, 0x72, 0x6f, 0x2f, 0x70, 0x6b, 0x67, 0x2f, 0x70,
-	0x6c, 0x75, 0x67, 0x69, 0x6e, 0x2f, 0x67, 0x65, 0x6e, 0x65, 0x72, 0x61, 0x74, 0x65, 0x64, 0x2f,
-	0x62, 0x61, 0x63, 0x6b, 0x75, 0x70, 0x69, 0x74, 0x65, 0x6d, 0x61, 0x63, 0x74, 0x69, 0x6f, 0x6e,
-	0x2f, 0x76, 0x32, 0x62, 0x06, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x33,
+	0x52, 0x65, 0x71, 0x75, 0x65, 0x73, 0x74, 0x12, 0x16, 0x0a, 0x06, 0x70, 0x6c, 0x75, 0x67, 0x69,
+	0x6e, 0x18, 0x01, 0x20, 0x01, 0x28, 0x09, 0x52, 0x06, 0x70, 0x6c, 0x75, 0x67, 0x69, 0x6e, 0x22,
+	0x6c, 0x0a, 0x21, 0x42, 0x61, 0x63, 0x6b, 0x75, 0x70, 0x49, 0x74, 0x65, 0x6d, 0x41, 0x63, 0x74,
+	0x69, 0x6f, 0x6e, 0x41, 0x70, 0x70, 0x6c, 0x69, 0x65, 0x73, 0x54, 0x6f, 0x52, 0x65, 0x73, 0x70,
+	0x6f, 0x6e, 0x73, 0x65, 0x12, 0x47, 0x0a, 0x10, 0x52, 0x65, 0x73, 0x6f, 0x75, 0x72, 0x63, 0x65,
+	0x53, 0x65, 0x6c, 0x65, 0x63, 0x74, 0x6f, 0x72, 0x18, 0x01, 0x20, 0x01, 0x28, 0x0b, 0x32, 0x1b,
+	0x2e, 0x67, 0x65, 0x6e, 0x65, 0x72, 0x61, 0x74, 0x65, 0x64, 0x2e, 0x52, 0x65, 0x73, 0x6f, 0x75,
+	0x72, 0x63, 0x65, 0x53, 0x65, 0x6c, 0x65, 0x63, 0x74, 0x6f, 0x72, 0x52, 0x10, 0x52, 0x65, 0x73,
+	0x6f, 0x75, 0x72, 0x63, 0x65, 0x53, 0x65, 0x6c, 0x65, 0x63, 0x74, 0x6f, 0x72, 0x22, 0x73, 0x0a,
+	0x1f, 0x42, 0x61, 0x63, 0x6b, 0x75, 0x70, 0x49, 0x74, 0x65, 0x6d, 0x41, 0x63, 0x74, 0x69, 0x6f,
+	0x6e, 0x50, 0x72, 0x6f, 0x67, 0x72, 0x65, 0x73, 0x73, 0x52, 0x65, 0x71, 0x75, 0x65, 0x73, 0x74,
+	0x12, 0x16, 0x0a, 0x06, 0x70, 0x6c, 0x75, 0x67, 0x69, 0x6e, 0x18, 0x01, 0x20, 0x01, 0x28, 0x09,
+	0x52, 0x06, 0x70, 0x6c, 0x75, 0x67, 0x69, 0x6e, 0x12, 0x20, 0x0a, 0x0b, 0x6f, 0x70, 0x65, 0x72,
+	0x61, 0x74, 0x69, 0x6f, 0x6e, 0x49, 0x44, 0x18, 0x02, 0x20, 0x01, 0x28, 0x09, 0x52, 0x0b, 0x6f,
+	0x70, 0x65, 0x72, 0x61, 0x74, 0x69, 0x6f, 0x6e, 0x49, 0x44, 0x12, 0x16, 0x0a, 0x06, 0x62, 0x61,
+	0x63, 0x6b, 0x75, 0x70, 0x18, 0x03, 0x20, 0x01, 0x28, 0x0c, 0x52, 0x06, 0x62, 0x61, 0x63, 0x6b,
+	0x75, 0x70, 0x22, 0x5c, 0x0a, 0x20, 0x42, 0x61, 0x63, 0x6b, 0x75, 0x70, 0x49, 0x74, 0x65, 0x6d,
+	0x41, 0x63, 0x74, 0x69, 0x6f, 0x6e, 0x50, 0x72, 0x6f, 0x67, 0x72, 0x65, 0x73, 0x73, 0x52, 0x65,
+	0x73, 0x70, 0x6f, 0x6e, 0x73, 0x65, 0x12, 0x38, 0x0a, 0x08, 0x70, 0x72, 0x6f, 0x67, 0x72, 0x65,
+	0x73, 0x73, 0x18, 0x01, 0x20, 0x01, 0x28, 0x0b, 0x32, 0x1c, 0x2e, 0x67, 0x65, 0x6e, 0x65, 0x72,
+	0x61, 0x74, 0x65, 0x64, 0x2e, 0x4f, 0x70, 0x65, 0x72, 0x61, 0x74, 0x69, 0x6f, 0x6e, 0x50, 0x72,
+	0x6f, 0x67, 0x72, 0x65, 0x73, 0x73, 0x52, 0x08, 0x70, 0x72, 0x6f, 0x67, 0x72, 0x65, 0x73, 0x73,
+	0x22, 0x71, 0x0a, 0x1d, 0x42, 0x61, 0x63, 0x6b, 0x75, 0x70, 0x49, 0x74, 0x65, 0x6d, 0x41, 0x63,
+	0x74, 0x69, 0x6f, 0x6e, 0x43, 0x61, 0x6e, 0x63, 0x65, 0x6c, 0x52, 0x65, 0x71, 0x75, 0x65, 0x73,
+	0x74, 0x12, 0x16, 0x0a, 0x06, 0x70, 0x6c, 0x75, 0x67, 0x69, 0x6e, 0x18, 0x01, 0x20, 0x01, 0x28,
+	0x09, 0x52, 0x06, 0x70, 0x6c, 0x75, 0x67, 0x69, 0x6e, 0x12, 0x20, 0x0a, 0x0b, 0x6f, 0x70, 0x65,
+	0x72, 0x61, 0x74, 0x69, 0x6f, 0x6e, 0x49, 0x44, 0x18, 0x02, 0x20, 0x01, 0x28, 0x09, 0x52, 0x0b,
+	0x6f, 0x70, 0x65, 0x72, 0x61, 0x74, 0x69, 0x6f, 0x6e, 0x49, 0x44, 0x12, 0x16, 0x0a, 0x06, 0x62,
+	0x61, 0x63, 0x6b, 0x75, 0x70, 0x18, 0x03, 0x20, 0x01, 0x28, 0x0c, 0x52, 0x06, 0x62, 0x61, 0x63,
+	0x6b, 0x75, 0x70, 0x32, 0xbc, 0x02, 0x0a, 0x10, 0x42, 0x61, 0x63, 0x6b, 0x75, 0x70, 0x49, 0x74,
+	0x65, 0x6d, 0x41, 0x63, 0x74, 0x69, 0x6f, 0x6e, 0x12, 0x58, 0x0a, 0x09, 0x41, 0x70, 0x70, 0x6c,
+	0x69, 0x65, 0x73, 0x54, 0x6f, 0x12, 0x24, 0x2e, 0x76, 0x32, 0x2e, 0x42, 0x61, 0x63, 0x6b, 0x75,
+	0x70, 0x49, 0x74, 0x65, 0x6d, 0x41, 0x63, 0x74, 0x69, 0x6f, 0x6e, 0x41, 0x70, 0x70, 0x6c, 0x69,
+	0x65, 0x73, 0x54, 0x6f, 0x52, 0x65, 0x71, 0x75, 0x65, 0x73, 0x74, 0x1a, 0x25, 0x2e, 0x76, 0x32,
+	0x2e, 0x42, 0x61, 0x63, 0x6b, 0x75, 0x70, 0x49, 0x74, 0x65, 0x6d, 0x41, 0x63, 0x74, 0x69, 0x6f,
+	0x6e, 0x41, 0x70, 0x70, 0x6c, 0x69, 0x65, 0x73, 0x54, 0x6f, 0x52, 0x65, 0x73, 0x70, 0x6f, 0x6e,
+	0x73, 0x65, 0x12, 0x32, 0x0a, 0x07, 0x45, 0x78, 0x65, 0x63, 0x75, 0x74, 0x65, 0x12, 0x12, 0x2e,
+	0x76, 0x32, 0x2e, 0x45, 0x78, 0x65, 0x63, 0x75, 0x74, 0x65, 0x52, 0x65, 0x71, 0x75, 0x65, 0x73,
+	0x74, 0x1a, 0x13, 0x2e, 0x76, 0x32, 0x2e, 0x45, 0x78, 0x65, 0x63, 0x75, 0x74, 0x65, 0x52, 0x65,
+	0x73, 0x70, 0x6f, 0x6e, 0x73, 0x65, 0x12, 0x55, 0x0a, 0x08, 0x50, 0x72, 0x6f, 0x67, 0x72, 0x65,
+	0x73, 0x73, 0x12, 0x23, 0x2e, 0x76, 0x32, 0x2e, 0x42, 0x61, 0x63, 0x6b, 0x75, 0x70, 0x49, 0x74,
+	0x65, 0x6d, 0x41, 0x63, 0x74, 0x69, 0x6f, 0x6e, 0x50, 0x72, 0x6f, 0x67, 0x72, 0x65, 0x73, 0x73,
+	0x52, 0x65, 0x71, 0x75, 0x65, 0x73, 0x74, 0x1a, 0x24, 0x2e, 0x76, 0x32, 0x2e, 0x42, 0x61, 0x63,
+	0x6b, 0x75, 0x70, 0x49, 0x74, 0x65, 0x6d, 0x41, 0x63, 0x74, 0x69, 0x6f, 0x6e, 0x50, 0x72, 0x6f,
+	0x67, 0x72, 0x65, 0x73, 0x73, 0x52, 0x65, 0x73, 0x70, 0x6f, 0x6e, 0x73, 0x65, 0x12, 0x43, 0x0a,
+	0x06, 0x43, 0x61, 0x6e, 0x63, 0x65, 0x6c, 0x12, 0x21, 0x2e, 0x76, 0x32, 0x2e, 0x42, 0x61, 0x63,
+	0x6b, 0x75, 0x70, 0x49, 0x74, 0x65, 0x6d, 0x41, 0x63, 0x74, 0x69, 0x6f, 0x6e, 0x43, 0x61, 0x6e,
+	0x63, 0x65, 0x6c, 0x52, 0x65, 0x71, 0x75, 0x65, 0x73, 0x74, 0x1a, 0x16, 0x2e, 0x67, 0x6f, 0x6f,
+	0x67, 0x6c, 0x65, 0x2e, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x62, 0x75, 0x66, 0x2e, 0x45, 0x6d, 0x70,
+	0x74, 0x79, 0x42, 0x49, 0x5a, 0x47, 0x67, 0x69, 0x74, 0x68, 0x75, 0x62, 0x2e, 0x63, 0x6f, 0x6d,
+	0x2f, 0x76, 0x6d, 0x77, 0x61, 0x72, 0x65, 0x2d, 0x74, 0x61, 0x6e, 0x7a, 0x75, 0x2f, 0x76, 0x65,
+	0x6c, 0x65, 0x72, 0x6f, 0x2f, 0x70, 0x6b, 0x67, 0x2f, 0x70, 0x6c, 0x75, 0x67, 0x69, 0x6e, 0x2f,
+	0x67, 0x65, 0x6e, 0x65, 0x72, 0x61, 0x74, 0x65, 0x64, 0x2f, 0x62, 0x61, 0x63, 0x6b, 0x75, 0x70,
+	0x69, 0x74, 0x65, 0x6d, 0x61, 0x63, 0x74, 0x69, 0x6f, 0x6e, 0x2f, 0x76, 0x32, 0x62, 0x06, 0x70,
+	0x72, 0x6f, 0x74, 0x6f, 0x33,
 }

 var (
@@ -546,7 +547,7 @@ var file_backupitemaction_v2_BackupItemAction_proto_goTypes = []interface{}{
 }
 var file_backupitemaction_v2_BackupItemAction_proto_depIdxs = []int32{
 	7,  // 0: v2.ExecuteResponse.additionalItems:type_name -> generated.ResourceIdentifier
-	7,  // 1: v2.ExecuteResponse.itemsToUpdate:type_name -> generated.ResourceIdentifier
+	7,  // 1: v2.ExecuteResponse.postOperationItems:type_name -> generated.ResourceIdentifier
 	8,  // 2: v2.BackupItemActionAppliesToResponse.ResourceSelector:type_name -> generated.ResourceSelector
 	9,  // 3: v2.BackupItemActionProgressResponse.progress:type_name -> generated.OperationProgress
 	2,  // 4: v2.BackupItemAction.AppliesTo:input_type -> v2.BackupItemActionAppliesToRequest
--- a/pkg/plugin/proto/backupitemaction/v2/BackupItemAction.proto
+++ b/pkg/plugin/proto/backupitemaction/v2/BackupItemAction.proto
@@ -16,7 +16,7 @@ message ExecuteResponse {
    bytes item = 1;
    repeated generated.ResourceIdentifier additionalItems = 2;
    string operationID = 3;
-    repeated generated.ResourceIdentifier itemsToUpdate = 4;
+    repeated generated.ResourceIdentifier postOperationItems = 4;
 }

 service BackupItemAction {
--- a/pkg/plugin/velero/backupitemaction/v2/backup_item_action.go
+++ b/pkg/plugin/velero/backupitemaction/v2/backup_item_action.go
@@ -44,11 +44,17 @@ type BackupItemAction interface {
 	// including mutating the item itself prior to backup. The item (unmodified or modified)
 	// should be returned, along with an optional slice of ResourceIdentifiers specifying
 	// additional related items that should be backed up now, an optional operationID for actions which
-	// initiate asynchronous actions, and a second slice of ResourceIdentifiers specifying related items
-	// which should be backed up after all asynchronous operations have completed. This last field will be
+	// initiate (asynchronous) operations, and a second slice of ResourceIdentifiers specifying related items
+	// which should be backed up after all operations have completed. This last field will be
 	// ignored if operationID is empty, and should not be filled in unless the resource must be updated in the
-	// backup after async operations complete (i.e. some of the item's kubernetes metadata will be updated
-	// during the asynch operation which will be required during restore)
+	// backup after operations complete (i.e. some of the item's kubernetes metadata will be updated
+	// during the operation which will be required during restore)
+	// Note that (async) operations are not supported for items being backed up during Finalize phases,
+	// so a plugin should not return an OperationID if the backup phase is "Finalizing"
+	// or "FinalizingPartiallyFailed". The plugin should check the incoming
+	// backup.Status.Phase before initiating operations, since the backup has already passed the waiting
+	// for plugin operations phase. Plugins being called during Finalize will only be called for resources
+	// that were returned as postOperationItems.
 	Execute(item runtime.Unstructured, backup *api.Backup) (runtime.Unstructured, []velero.ResourceIdentifier, string, []velero.ResourceIdentifier, error)

 	// Progress allows the BackupItemAction to report on progress of an asynchronous action.
--- a/pkg/util/encode/encode.go
+++ b/pkg/util/encode/encode.go
@@ -18,6 +18,8 @@ package encode

 import (
 	"bytes"
+	"compress/gzip"
+	"encoding/json"
 	"fmt"
 	"io"

@@ -70,3 +72,26 @@ func EncoderFor(format string, obj runtime.Object) (runtime.Encoder, error) {
 	encoder = scheme.Codecs.EncoderForVersion(encoder, v1.SchemeGroupVersion)
 	return encoder, nil
 }
+
+// EncodeToJSONGzip takes arbitrary Go data and encodes it to GZip compressed JSON in a buffer, as well as a description of the data to put into an error should encoding fail.
+func EncodeToJSONGzip(data interface{}, desc string) (*bytes.Buffer, []error) {
+	buf := new(bytes.Buffer)
+	gzw := gzip.NewWriter(buf)
+
+	// Since both encoding and closing the gzip writer could fail separately and both errors are useful,
+	// collect both errors to report back.
+	errs := []error{}
+
+	if err := json.NewEncoder(gzw).Encode(data); err != nil {
+		errs = append(errs, errors.Wrapf(err, "error encoding %s", desc))
+	}
+	if err := gzw.Close(); err != nil {
+		errs = append(errs, errors.Wrapf(err, "error closing gzip writer for %s", desc))
+	}
+
+	if len(errs) > 0 {
+		return nil, errs
+	}
+
+	return buf, nil
+}
--- a/site/content/docs/main/api-types/backup.md
+++ b/site/content/docs/main/api-types/backup.md
@@ -36,7 +36,7 @@ spec:
  # ItemOperationTimeout specifies the time used to wait for
  # asynchronous BackupItemAction operations
  # The default value is 1 hour.
-  csiSnapshotTimeout: 1h
+  itemOperationTimeout: 1h
  # Array of namespaces to include in the backup. If unspecified, all namespaces are included.
  # Optional.
  includedNamespaces:
@@ -151,7 +151,7 @@ status:
  # The current phase.
  # Valid values are New, FailedValidation, InProgress, WaitingForPluginOperations,
  # WaitingForPluginOperationsPartiallyFailed, FinalizingafterPluginOperations,
-  # FinalizingAfterPluginOperationsPartiallyFailed, Completed, PartiallyFailed, Failed.
+  # FinalizingPartiallyFailed, Completed, PartiallyFailed, Failed.
  phase: ""
  # An array of any validation errors encountered.
  validationErrors: null
@@ -163,12 +163,12 @@ status:
  volumeSnapshotsAttempted: 2
  # Number of volume snapshots that Velero successfully created for this backup.
  volumeSnapshotsCompleted: 1
-  # Number of attempted async BackupItemAction operations for this backup.
-  asyncBackupItemOperationsAttempted: 2
-  # Number of async BackupItemAction operations that Velero successfully completed for this backup.
-  asyncBackupItemOperationsCompleted: 1
-  # Number of async BackupItemAction operations that ended in failure for this backup.
-  asyncBackupItemOperationsFailed: 0
+  # Number of attempted BackupItemAction operations for this backup.
+  backupItemOperationsAttempted: 2
+  # Number of BackupItemAction operations that Velero successfully completed for this backup.
+  backupItemOperationsCompleted: 1
+  # Number of BackupItemAction operations that ended in failure for this backup.
+  backupItemOperationsFailed: 0
  # Number of warnings that were logged by the backup.
  warnings: 2
  # Number of errors that were logged by the backup.