From 4514d4e452381c2003df6b33d9b29d15daefc156 Mon Sep 17 00:00:00 2001 From: binarybrain-009 Date: Tue, 20 Oct 2026 01:35:40 -0400 Subject: [PATCH] Core: Send snapshot events and commit reports after a transaction commits Snapshot producers in a transaction notified listeners and reported commit metrics when their change was only staged, so events were sent for snapshots that might never be committed, and sent again when a retried transaction re-applied its updates. The transaction now sends each producer's latest event once the transaction has committed. Closes #7278 --- .../org/apache/iceberg/BaseTransaction.java | 11 +++++++ .../org/apache/iceberg/SnapshotProducer.java | 28 ++++++++++++++-- .../apache/iceberg/TestCommitReporting.java | 30 +++++++++++++++++ .../iceberg/TestCreateSnapshotEvent.java | 26 +++++++++++++++ .../apache/iceberg/catalog/CatalogTests.java | 32 ++++++++----------- 5 files changed, 106 insertions(+), 21 deletions(-) diff --git a/core/src/main/java/org/apache/iceberg/BaseTransaction.java b/core/src/main/java/org/apache/iceberg/BaseTransaction.java index af353f9429bf..333ce91fa74e 100644 --- a/core/src/main/java/org/apache/iceberg/BaseTransaction.java +++ b/core/src/main/java/org/apache/iceberg/BaseTransaction.java @@ -129,6 +129,7 @@ protected final T appendUpdate(T update) { if (update instanceof SnapshotProducer) { ((SnapshotProducer) update).reportWith(reporter); + ((SnapshotProducer) update).deferNotifications(); } this.updates.add(update); @@ -268,6 +269,8 @@ public void commitTransaction() { commitSimpleTransaction(); break; } + + sendDeferredNotifications(); } private void commitCreateTransaction() { @@ -432,6 +435,14 @@ private void cleanAllUpdates() { }); } + private void sendDeferredNotifications() { + for (PendingUpdate update : updates) { + if (update instanceof SnapshotProducer producer) { + producer.sendDeferredNotifications(); + } + } + } + private void deleteUncommittedFiles(Iterable paths) { CatalogUtil.deleteFiles(ops.io(), paths, "uncommitted"); } diff --git a/core/src/main/java/org/apache/iceberg/SnapshotProducer.java b/core/src/main/java/org/apache/iceberg/SnapshotProducer.java index 06281188c32a..00d9016d22da 100644 --- a/core/src/main/java/org/apache/iceberg/SnapshotProducer.java +++ b/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 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); diff --git a/core/src/test/java/org/apache/iceberg/TestCommitReporting.java b/core/src/test/java/org/apache/iceberg/TestCommitReporting.java index 3d39dc6d3896..c8d2ce0aadf0 100644 --- a/core/src/test/java/org/apache/iceberg/TestCommitReporting.java +++ b/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 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"; diff --git a/core/src/test/java/org/apache/iceberg/TestCreateSnapshotEvent.java b/core/src/test/java/org/apache/iceberg/TestCreateSnapshotEvent.java index e2fa80707a2f..6675726d90fa 100644 --- a/core/src/test/java/org/apache/iceberg/TestCreateSnapshotEvent.java +++ b/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 { @Override public void notify(CreateSnapshotEvent event) { diff --git a/core/src/test/java/org/apache/iceberg/catalog/CatalogTests.java b/core/src/test/java/org/apache/iceberg/catalog/CatalogTests.java index 9e6ad888c601..eb0ab2befbda 100644 --- a/core/src/test/java/org/apache/iceberg/catalog/CatalogTests.java +++ b/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);