diff --git a/core/src/main/java/org/apache/iceberg/SnapshotProducer.java b/core/src/main/java/org/apache/iceberg/SnapshotProducer.java
index 7825500bc7cb..f6b3a2d03713 100644
--- a/core/src/main/java/org/apache/iceberg/SnapshotProducer.java
+++ b/core/src/main/java/org/apache/iceberg/SnapshotProducer.java
@@ -282,14 +282,18 @@ public void commit() {
           .run(taskOps -> {
             Snapshot newSnapshot = apply();
             newSnapshotId.set(newSnapshot.snapshotId());
-            TableMetadata updated;
-            if (stageOnly) {
-              updated = base.addStagedSnapshot(newSnapshot);
+            TableMetadata.Builder update = TableMetadata.buildFrom(base);
+            if (base.snapshot(newSnapshot.snapshotId()) != null) {
+              // this is a rollback or cherrypick operation
+              update.setCurrentSnapshot(newSnapshot.snapshotId());
+            } else if (stageOnly) {
+              update.addSnapshot(newSnapshot);
             } else {
-              updated = base.replaceCurrentSnapshot(newSnapshot);
+              update.setCurrentSnapshot(newSnapshot);
             }
 
-            if (updated == base) {
+            TableMetadata updated = update.build();
+            if (updated.changes().isEmpty()) {
               // do not commit if the metadata has not changed. for example, this may happen when setting the current
               // snapshot to an ID that is already current. note that this check uses identity.
               return;
diff --git a/core/src/main/java/org/apache/iceberg/TableMetadata.java b/core/src/main/java/org/apache/iceberg/TableMetadata.java
index 164e29505c84..89ddee69ee14 100644
--- a/core/src/main/java/org/apache/iceberg/TableMetadata.java
+++ b/core/src/main/java/org/apache/iceberg/TableMetadata.java
@@ -501,6 +501,10 @@ public TableMetadata replaceCurrentSnapshot(Snapshot snapshot) {
     return new Builder(this).setCurrentSnapshot(snapshot).build();
   }
 
+  public TableMetadata replaceCurrentSnapshot(long snapshotId) {
+    return new Builder(this).setCurrentSnapshot(snapshotId).build();
+  }
+
   public TableMetadata removeSnapshotsIf(Predicate<Snapshot> removeIf) {
     List<Snapshot> toRemove = snapshots.stream().filter(removeIf).collect(Collectors.toList());
     return new Builder(this).removeSnapshots(toRemove).build();