From 744212e67f6d52704c9a712657552a0ce6b38114 Mon Sep 17 00:00:00 2001
From: ashmehta <ashmehta@adobe.com>
Date: Tue, 25 Feb 2020 11:12:10 -0800
Subject: [PATCH] @trivial: Expire snapshots file leaking problem.

---
 .../apache/iceberg/TestRemoveSnapshots.java   | 43 +++++++++++++++++++
 1 file changed, 43 insertions(+)

diff --git a/core/src/test/java/org/apache/iceberg/TestRemoveSnapshots.java b/core/src/test/java/org/apache/iceberg/TestRemoveSnapshots.java
index 7f4d2f773c8f..f345299e76e9 100644
--- a/core/src/test/java/org/apache/iceberg/TestRemoveSnapshots.java
+++ b/core/src/test/java/org/apache/iceberg/TestRemoveSnapshots.java
@@ -387,4 +387,47 @@ public void dataFilesCleanup() throws IOException {
     Assert.assertTrue("FILE_A should be deleted", deletedFiles.contains(FILE_A.path().toString()));
     Assert.assertTrue("FILE_B should be deleted", deletedFiles.contains(FILE_B.path().toString()));
   }
+
+  /**
+   * Test on table below, and expiring the staged commit `B` using `expireOlderThan` API.
+   * Table: A - C
+   *          ` B (staged)
+   */
+  @Test
+  public void testWithExpiringDanglingStageCommitFail() {
+    // `A` commit
+    table.newAppend()
+        .appendFile(FILE_A)
+        .commit();
+
+    // `B` staged commit
+    table.newAppend()
+        .appendFile(FILE_B)
+        .stageOnly()
+        .commit();
+
+    TableMetadata base = readMetadata();
+    Snapshot snapshotB = base.snapshots().get(1);
+
+    // `C` commit
+    table.newAppend()
+        .appendFile(FILE_C)
+        .commit();
+
+    List<Snapshot> existingSnapshots = readMetadata().snapshots();
+    try {
+      // Expire all commits including dangling staged snapshot.
+      table.expireSnapshots()
+          .deleteWith(item -> {
+            throw new RuntimeException("Failure from underlying storage.");
+          })
+          .expireOlderThan(snapshotB.timestampMillis() + 1)
+          .commit();
+    } catch (Exception e) {
+      e.printStackTrace();
+    }
+
+    //Considering that snapshot is gone from history, how are we support to try clean up?
+    assert (readMetadata().snapshots().size() != existingSnapshots.size());
+  }
 }