Core: Send snapshot events and commit reports after a transaction commits

core/src/main/java/org/apache/iceberg/BaseTransaction.java

@@ -123,12 +123,13 @@
   protected final <T extends PendingUpdate> T appendUpdate(T update) {
     checkLastOperationCommitted(update.getClass());
 
     if (update instanceof SnapshotUpdate) {
       ((SnapshotUpdate) update).deleteWith(enqueueDelete);
     }
 
     if (update instanceof SnapshotProducer) {
       ((SnapshotProducer) update).reportWith(reporter);
+      ((SnapshotProducer) update).deferNotifications();
     }
 
     this.updates.add(update);
@@ -251,7 +252,7 @@
     Preconditions.checkState(
         hasLastOpCommitted, "Cannot commit transaction: last operation has not committed");
 
     switch (type) {
       case CREATE_TABLE:
         commitCreateTransaction();
         break;
@@ -268,6 +269,8 @@
         commitSimpleTransaction();
         break;
     }
+
+    sendDeferredNotifications();
   }
 
   private void commitCreateTransaction() {
@@ -426,12 +429,20 @@
         .suppressFailureWhenFinished()
         .run(
             update -> {
               if (update instanceof SnapshotProducer) {
                 ((SnapshotProducer) update).cleanAll();
               }
             });
   }
 
+  private void sendDeferredNotifications() {
+    for (PendingUpdate update : updates) {
+      if (update instanceof SnapshotProducer<?> producer) {
+        producer.sendDeferredNotifications();
+      }
+    }
+  }
+
   private void deleteUncommittedFiles(Iterable<String> paths) {
     CatalogUtil.deleteFiles(ops.io(), paths, "uncommitted");
   }

core/src/main/java/org/apache/iceberg/SnapshotProducer.java

@@ -118,6 +118,8 @@ public void accept(String file) {
   // Encryption keys required by the manifest list written in the current commit attempt.
   private FileEncryptionKeys manifestListEncryptionKeys;
   private boolean stageOnly = false;
+  private boolean deferNotifications = false;
+  private Object deferredEvent = null;
   private Consumer<String> deleteFunc = defaultDelete;
   private SnapshotAncestryValidator snapshotAncestryValidator =
       SnapshotAncestryValidator.NON_VALIDATING;
@@ -212,6 +214,14 @@ protected ThisT reportWith(MetricsReporter newReporter) {
     return self();
   }
 
+  /**
+   * Stops {@link #commit()} from notifying listeners and reporting commit metrics; the caller must
+   * call {@link #sendDeferredNotifications()} once the last commit is visible in the table.
+   */
+  void deferNotifications() {
+    this.deferNotifications = true;
+  }
+
   /**
    * A setter for the target branch on which snapshot producer operation should be performed
    *
@@ -577,15 +587,27 @@ public void commit() {
     }
 
     try {
-      notifyListeners();
+      Object event = updateEvent();
+      if (deferNotifications) {
+        this.deferredEvent = event;
+      } else {
+        notifyListeners(event);
+      }
+    } catch (Throwable e) {
+      LOG.warn("Failed to notify event listeners", e);
+    }
+  }
+
+  void sendDeferredNotifications() {
+    try {
+      notifyListeners(deferredEvent);
     } catch (Throwable e) {
       LOG.warn("Failed to notify event listeners", e);
     }
   }
 
-  private void notifyListeners() {
+  private void notifyListeners(Object event) {
     try {
-      Object event = updateEvent();
       if (event != null) {
         Listeners.notifyAll(event);
 

core/src/test/java/org/apache/iceberg/TestCommitReporting.java

@@ -24,7 +24,9 @@
 import org.apache.iceberg.ScanPlanningAndReportingTestBase.TestMetricsReporter;
 import org.apache.iceberg.metrics.CommitMetricsResult;
 import org.apache.iceberg.metrics.CommitReport;
+import org.apache.iceberg.metrics.MetricsReport;
 import org.apache.iceberg.relocated.com.google.common.collect.ImmutableSet;
+import org.apache.iceberg.relocated.com.google.common.collect.Lists;
 import org.junit.jupiter.api.TestTemplate;
 import org.junit.jupiter.api.extension.ExtendWith;
 
@@ -185,6 +187,34 @@ public void addAndDeleteDeleteFiles() {
     assertThat(metrics.manifestsReplaced().value()).isEqualTo(1L);
   }
 
+  @TestTemplate
+  void oneCommitReportForRetriedTransaction() {
+    List<MetricsReport> reports = Lists.newArrayList();
+    Table table =
+        TestTables.create(
+            tableDir,
+            "retried-txn",
+            SCHEMA,
+            SPEC,
+            SortOrder.unsorted(),
+            formatVersion,
+            reports::add);
+
+    Transaction transaction = table.newTransaction();
+    transaction.newAppend().appendFile(FILE_A).commit();
+    // a concurrent commit makes the transaction re-apply its append
+    table.newAppend().appendFile(FILE_B).commit();
+    transaction.commitTransaction();
+
+    Snapshot committed = table.currentSnapshot();
+    assertThat(reports)
+        .map(CommitReport.class::cast)
+        .filteredOn(report -> report.snapshotId() == committed.snapshotId())
+        .singleElement()
+        .extracting(CommitReport::sequenceNumber)
+        .isEqualTo(committed.sequenceNumber());
+  }
+
   @TestTemplate
   public void addAndDeleteManifests() {
     String tableName = "add-and-delete-manifests";

core/src/test/java/org/apache/iceberg/TestCreateSnapshotEvent.java

@@ -19,10 +19,12 @@
 package org.apache.iceberg;
 
 import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
 
 import org.apache.iceberg.events.CreateSnapshotEvent;
 import org.apache.iceberg.events.Listener;
 import org.apache.iceberg.events.Listeners;
+import org.apache.iceberg.exceptions.CommitFailedException;
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.TestTemplate;
 
@@ -77,6 +79,30 @@ public void testAppendAndDeleteCommitEvent() {
         .containsEntry("total-data-files", "1");
   }
 
+  @TestTemplate
+  void snapshotEventAfterTransactionCommit() {
+    Transaction transaction = table.newTransaction();
+    transaction.newAppend().appendFile(FILE_A).commit();
+    assertThat(currentEvent).isNull();
+
+    transaction.commitTransaction();
+    assertThat(currentEvent).isNotNull();
+    assertThat(currentEvent.snapshotId()).isEqualTo(table.currentSnapshot().snapshotId());
+  }
+
+  @TestTemplate
+  void noSnapshotEventWhenTransactionCommitFails() {
+    table.updateProperties().set(TableProperties.COMMIT_NUM_RETRIES, "0").commit();
+    Transaction transaction = table.newTransaction();
+    transaction.newAppend().appendFile(FILE_A).commit();
+    table.ops().failCommits(1);
+
+    assertThatThrownBy(transaction::commitTransaction)
+        .isInstanceOf(CommitFailedException.class)
+        .hasMessage("Injected failure");
+    assertThat(currentEvent).isNull();
+  }
+
   class MyListener implements Listener<CreateSnapshotEvent> {
     @Override
     public void notify(CreateSnapshotEvent event) {

core/src/test/java/org/apache/iceberg/catalog/CatalogTests.java

@@ -3528,32 +3528,28 @@ public void testCatalogWithCustomMetricsReporter() throws IOException {
 
     TableIdentifier identifier = TableIdentifier.of(NS, "custom_metrics_reporter_table");
     // append file through createTransaction() and check and reset counter
-    catalogWithCustomReporter
-        .buildTable(identifier, SCHEMA)
-        .createTransaction()
-        .newFastAppend()
-        .appendFile(dataFile)
-        .commit();
+    Transaction create =
+        catalogWithCustomReporter.buildTable(identifier, SCHEMA).createTransaction();
+    create.newFastAppend().appendFile(dataFile).commit();
+    assertThat(CustomMetricsReporter.COMMIT_COUNTER.get()).isZero();
+    create.commitTransaction();
     assertThat(CustomMetricsReporter.COMMIT_COUNTER.get()).isEqualTo(1);
     CustomMetricsReporter.COMMIT_COUNTER.set(0);
 
     // append file through createOrReplaceTransaction() and check and reset counter
-    catalogWithCustomReporter
-        .buildTable(identifier, SCHEMA)
-        .createOrReplaceTransaction()
-        .newFastAppend()
-        .appendFile(dataFile)
-        .commit();
+    Transaction createOrReplace =
+        catalogWithCustomReporter.buildTable(identifier, SCHEMA).createOrReplaceTransaction();
+    createOrReplace.newFastAppend().appendFile(dataFile).commit();
+    assertThat(CustomMetricsReporter.COMMIT_COUNTER.get()).isZero();
+    createOrReplace.commitTransaction();
     assertThat(CustomMetricsReporter.COMMIT_COUNTER.get()).isEqualTo(1);
     CustomMetricsReporter.COMMIT_COUNTER.set(0);
 
     // append file through replaceTransaction() and check and reset counter
-    catalogWithCustomReporter
-        .buildTable(TABLE, SCHEMA)
-        .replaceTransaction()
-        .newFastAppend()
-        .appendFile(dataFile)
-        .commit();
+    Transaction replace = catalogWithCustomReporter.buildTable(TABLE, SCHEMA).replaceTransaction();
+    replace.newFastAppend().appendFile(dataFile).commit();
+    assertThat(CustomMetricsReporter.COMMIT_COUNTER.get()).isZero();
+    replace.commitTransaction();
     assertThat(CustomMetricsReporter.COMMIT_COUNTER.get()).isEqualTo(1);
     CustomMetricsReporter.COMMIT_COUNTER.set(0);