From 63a3d689172cbe06627d969d4a7073a396bdea65 Mon Sep 17 00:00:00 2001
From: Will-Lo <wlo@apache.org>
Date: Thu, 12 Dec 2024 16:11:39 -0500
Subject: [PATCH 1/8] WIP Implement history configurations in expiration jobs

---
 .../tasks/TableSnapshotsExpirationTask.java   | 18 +++++++--
 .../openhouse/jobs/spark/Operations.java      |  7 ++--
 .../spark/SnapshotsExpirationSparkApp.java    | 40 +++++++++++++------
 .../openhouse/jobs/util/HistoryConfig.java    | 18 +++++++++
 .../openhouse/jobs/util/TableMetadata.java    |  1 +
 .../openhouse/jobs/spark/OperationsTest.java  | 34 +++++++++++++++-
 6 files changed, 97 insertions(+), 21 deletions(-)
 create mode 100644 apps/spark/src/main/java/com/linkedin/openhouse/jobs/util/HistoryConfig.java

diff --git a/apps/spark/src/main/java/com/linkedin/openhouse/jobs/scheduler/tasks/TableSnapshotsExpirationTask.java b/apps/spark/src/main/java/com/linkedin/openhouse/jobs/scheduler/tasks/TableSnapshotsExpirationTask.java
index ed7d69096..8a32eb87b 100644
--- a/apps/spark/src/main/java/com/linkedin/openhouse/jobs/scheduler/tasks/TableSnapshotsExpirationTask.java
+++ b/apps/spark/src/main/java/com/linkedin/openhouse/jobs/scheduler/tasks/TableSnapshotsExpirationTask.java
@@ -3,7 +3,9 @@
 import com.linkedin.openhouse.jobs.client.JobsClient;
 import com.linkedin.openhouse.jobs.client.TablesClient;
 import com.linkedin.openhouse.jobs.client.model.JobConf;
+import com.linkedin.openhouse.jobs.util.HistoryConfig;
 import com.linkedin.openhouse.jobs.util.TableMetadata;
+import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.List;
 
@@ -28,10 +30,18 @@ public JobConf.JobTypeEnum getType() {
 
   @Override
   protected List<String> getArgs() {
-    return Arrays.asList(
-        "--tableName", metadata.fqtn(),
-        "--granularity", "days",
-        "--count", "3");
+    HistoryConfig config = metadata.getHistoryConfig();
+    List<String> jobArgs = new ArrayList<>();
+    if (config.getMaxAge() > 0) {
+      jobArgs.addAll(
+          Arrays.asList(
+              "--maxAge", Integer.toString(config.getMaxAge()),
+              "--granularity", config.getGranularity().getValue()));
+    }
+    if (config.getMinVersions() > 0) {
+      jobArgs.addAll(Arrays.asList("--minVersions", Integer.toString(config.getMinVersions())));
+    }
+    return jobArgs;
   }
 
   @Override
diff --git a/apps/spark/src/main/java/com/linkedin/openhouse/jobs/spark/Operations.java b/apps/spark/src/main/java/com/linkedin/openhouse/jobs/spark/Operations.java
index 02d5ffbc9..6842690a8 100644
--- a/apps/spark/src/main/java/com/linkedin/openhouse/jobs/spark/Operations.java
+++ b/apps/spark/src/main/java/com/linkedin/openhouse/jobs/spark/Operations.java
@@ -201,16 +201,17 @@ public void deleteStagedOrphanDirectory(
   }
 
   /** Expire snapshots on a given fully-qualified table name. */
-  public void expireSnapshots(String fqtn, long expireBeforeTimestampMs) {
-    expireSnapshots(getTable(fqtn), expireBeforeTimestampMs);
+  public void expireSnapshots(String fqtn, long expireBeforeTimestampMs, int minVersions) {
+    expireSnapshots(getTable(fqtn), expireBeforeTimestampMs, minVersions);
   }
 
   /** Expire snapshots on a given {@link Table}. */
-  public void expireSnapshots(Table table, long expireBeforeTimestampMs) {
+  public void expireSnapshots(Table table, long expireBeforeTimestampMs, int minVersions) {
     table
         .expireSnapshots()
         .cleanExpiredFiles(false)
         .expireOlderThan(expireBeforeTimestampMs)
+        .retainLast(minVersions)
         .commit();
   }
 
diff --git a/apps/spark/src/main/java/com/linkedin/openhouse/jobs/spark/SnapshotsExpirationSparkApp.java b/apps/spark/src/main/java/com/linkedin/openhouse/jobs/spark/SnapshotsExpirationSparkApp.java
index 72c52324e..1a071a155 100644
--- a/apps/spark/src/main/java/com/linkedin/openhouse/jobs/spark/SnapshotsExpirationSparkApp.java
+++ b/apps/spark/src/main/java/com/linkedin/openhouse/jobs/spark/SnapshotsExpirationSparkApp.java
@@ -18,41 +18,57 @@
 @Slf4j
 public class SnapshotsExpirationSparkApp extends BaseTableSparkApp {
   private final String granularity;
-  private final int count;
+  private final int maxAge;
+  private final int minVersions;
+
+  private static final String DEFAULT_MAX_AGE = "3";
+
+  private static final String DEFAULT_GRANULARITY = "days";
+
+  private static final String DEFAULT_MIN_VERSIONS = "100";
 
   public SnapshotsExpirationSparkApp(
-      String jobId, StateManager stateManager, String fqtn, String granularity, int count) {
+      String jobId,
+      StateManager stateManager,
+      String fqtn,
+      int maxAge,
+      String granularity,
+      int minVersions) {
     super(jobId, stateManager, fqtn);
     this.granularity = granularity;
-    this.count = count;
+    this.maxAge = maxAge;
+    this.minVersions = minVersions;
   }
 
   @Override
   protected void runInner(Operations ops) {
     log.info(
-        "Snapshot expiration app start for table {}, expiring older than {} {}s",
+        "Snapshot expiration app start for table {}, expiring older than {} {}s or with more than {} versions",
         fqtn,
-        count,
-        granularity);
-    long expireBeforeTimestampMs = System.currentTimeMillis() - TimeUnit.DAYS.toMillis(count);
+        maxAge,
+        granularity,
+        minVersions);
+    long expireBeforeTimestampMs = System.currentTimeMillis() - TimeUnit.DAYS.toMillis(maxAge);
     log.info("Expire snapshots before timestamp ms {}", expireBeforeTimestampMs);
-    ops.expireSnapshots(fqtn, expireBeforeTimestampMs);
+    ops.expireSnapshots(fqtn, expireBeforeTimestampMs, minVersions);
   }
 
   public static void main(String[] args) {
     List<Option> extraOptions = new ArrayList<>();
     extraOptions.add(new Option("t", "tableName", true, "Fully-qualified table name"));
-    extraOptions.add(new Option("g", "granularity", true, "Granularity: day"));
     extraOptions.add(
-        new Option("c", "count", true, "Delete snapshots older than <count> <granularity>s"));
+        new Option("a", "maxAge", true, "Delete snapshots older than <maxAge> <granularity>s"));
+    extraOptions.add(new Option("g", "granularity", true, "Granularity: day"));
+    extraOptions.add(new Option("v", "minVersions", true, "Minimum number of versions to keep"));
     CommandLine cmdLine = createCommandLine(args, extraOptions);
     SnapshotsExpirationSparkApp app =
         new SnapshotsExpirationSparkApp(
             getJobId(cmdLine),
             createStateManager(cmdLine),
             cmdLine.getOptionValue("tableName"),
-            cmdLine.getOptionValue("granularity"),
-            Integer.parseInt(cmdLine.getOptionValue("count")));
+            Integer.parseInt(cmdLine.getOptionValue("maxAge", DEFAULT_MAX_AGE)),
+            cmdLine.getOptionValue("granularity", DEFAULT_GRANULARITY),
+            Integer.parseInt(cmdLine.getOptionValue("minVersions", DEFAULT_MIN_VERSIONS)));
     app.run();
   }
 }
diff --git a/apps/spark/src/main/java/com/linkedin/openhouse/jobs/util/HistoryConfig.java b/apps/spark/src/main/java/com/linkedin/openhouse/jobs/util/HistoryConfig.java
new file mode 100644
index 000000000..fe649fc26
--- /dev/null
+++ b/apps/spark/src/main/java/com/linkedin/openhouse/jobs/util/HistoryConfig.java
@@ -0,0 +1,18 @@
+package com.linkedin.openhouse.jobs.util;
+
+import com.linkedin.openhouse.tables.client.model.Retention;
+import lombok.Builder;
+import lombok.EqualsAndHashCode;
+import lombok.Getter;
+import lombok.ToString;
+
+/** Table retention config class. This is app side representation of /tables policies->retention */
+@Builder
+@Getter
+@EqualsAndHashCode
+@ToString
+public class HistoryConfig {
+  private final int maxAge;
+  private final int minVersions;
+  private final Retention.GranularityEnum granularity;
+}
diff --git a/apps/spark/src/main/java/com/linkedin/openhouse/jobs/util/TableMetadata.java b/apps/spark/src/main/java/com/linkedin/openhouse/jobs/util/TableMetadata.java
index 24623366f..4ae00253a 100644
--- a/apps/spark/src/main/java/com/linkedin/openhouse/jobs/util/TableMetadata.java
+++ b/apps/spark/src/main/java/com/linkedin/openhouse/jobs/util/TableMetadata.java
@@ -24,6 +24,7 @@ public class TableMetadata extends Metadata {
   protected boolean isClustered;
   @Builder.Default protected @NonNull Map<String, String> jobExecutionProperties = new HashMap<>();
   protected @Nullable RetentionConfig retentionConfig;
+  protected @Nullable HistoryConfig historyConfig;
 
   public String fqtn() {
     return String.format("%s.%s", dbName, tableName);
diff --git a/apps/spark/src/test/java/com/linkedin/openhouse/jobs/spark/OperationsTest.java b/apps/spark/src/test/java/com/linkedin/openhouse/jobs/spark/OperationsTest.java
index 2e5f6e772..4de939391 100644
--- a/apps/spark/src/test/java/com/linkedin/openhouse/jobs/spark/OperationsTest.java
+++ b/apps/spark/src/test/java/com/linkedin/openhouse/jobs/spark/OperationsTest.java
@@ -29,6 +29,8 @@
 import org.apache.iceberg.types.Types;
 import org.apache.spark.sql.Row;
 import org.assertj.core.util.Lists;
+import org.joda.time.DurationFieldType;
+import org.joda.time.Hours;
 import org.junit.jupiter.api.Assertions;
 import org.junit.jupiter.api.Test;
 
@@ -272,7 +274,7 @@ public void testOrphanFilesDeletionNoStaging() throws Exception {
   }
 
   @Test
-  public void testSnapshotsExpirationJavaAPI() throws Exception {
+  public void testSnapshotsExpirationMaxAge() throws Exception {
     final String tableName = "db.test_es_java";
     final int numInserts = 3;
     List<Long> snapshotIds;
@@ -286,7 +288,35 @@ public void testSnapshotsExpirationJavaAPI() throws Exception {
           String.format("There must be %d snapshot(s) after inserts", numInserts));
       Table table = ops.getTable(tableName);
       log.info("Loaded table {}, location {}", table.name(), table.location());
-      ops.expireSnapshots(table, System.currentTimeMillis());
+      ops.expireSnapshots(table, System.currentTimeMillis(), 3);
+      // verify that table object snapshots are updated
+      checkSnapshots(table, snapshotIds.subList(2, snapshotIds.size()));
+    }
+    // restart the app to reload catalog cache
+    try (Operations ops = Operations.withCatalog(getSparkSession(), meter)) {
+      // verify that new apps see snapshots correctly
+      checkSnapshots(ops, tableName, snapshotIds.subList(2, snapshotIds.size()));
+    }
+  }
+
+  @Test
+  public void testSnapshotsExpirationMinVersions() throws Exception {
+    final String tableName = "db.test_es_java";
+    final int numInserts = 3;
+    List<Long> snapshotIds;
+    try (Operations ops = Operations.withCatalog(getSparkSession(), meter)) {
+      prepareTable(ops, tableName);
+      populateTable(ops, tableName, numInserts);
+      snapshotIds = getSnapshotIds(ops, tableName);
+      Assertions.assertEquals(
+          numInserts,
+          snapshotIds.size(),
+          String.format("There must be %d snapshot(s) after inserts", numInserts));
+      Table table = ops.getTable(tableName);
+      log.info("Loaded table {}, location {}", table.name(), table.location());
+
+      ops.expireSnapshots(
+          table, System.currentTimeMillis() + Hours.ONE.get(DurationFieldType.millis()), 1);
       // verify that table object snapshots are updated
       checkSnapshots(table, snapshotIds.subList(2, snapshotIds.size()));
     }

From 55ae9d3bbeedb1412e12b724b06c429923bb8c96 Mon Sep 17 00:00:00 2001
From: Will-Lo <wlo@apache.org>
Date: Thu, 12 Dec 2024 16:14:30 -0500
Subject: [PATCH 2/8] Use different table names in tests

---
 .../com/linkedin/openhouse/jobs/spark/OperationsTest.java     | 4 ++--
 1 file changed, 2 insertions(+), 2 deletions(-)

diff --git a/apps/spark/src/test/java/com/linkedin/openhouse/jobs/spark/OperationsTest.java b/apps/spark/src/test/java/com/linkedin/openhouse/jobs/spark/OperationsTest.java
index 4de939391..164d26398 100644
--- a/apps/spark/src/test/java/com/linkedin/openhouse/jobs/spark/OperationsTest.java
+++ b/apps/spark/src/test/java/com/linkedin/openhouse/jobs/spark/OperationsTest.java
@@ -275,7 +275,7 @@ public void testOrphanFilesDeletionNoStaging() throws Exception {
 
   @Test
   public void testSnapshotsExpirationMaxAge() throws Exception {
-    final String tableName = "db.test_es_java";
+    final String tableName = "db.test_es_maxage_java";
     final int numInserts = 3;
     List<Long> snapshotIds;
     try (Operations ops = Operations.withCatalog(getSparkSession(), meter)) {
@@ -301,7 +301,7 @@ public void testSnapshotsExpirationMaxAge() throws Exception {
 
   @Test
   public void testSnapshotsExpirationMinVersions() throws Exception {
-    final String tableName = "db.test_es_java";
+    final String tableName = "db.test_es_minversions_java";
     final int numInserts = 3;
     List<Long> snapshotIds;
     try (Operations ops = Operations.withCatalog(getSparkSession(), meter)) {

From 444cc075a9d56ba81946fd593ccf8f909d8338ce Mon Sep 17 00:00:00 2001
From: Will-Lo <wlo@apache.org>
Date: Thu, 12 Dec 2024 17:55:44 -0500
Subject: [PATCH 3/8] Prioritize maintaining min count if defined but do not
 define a default

---
 .../openhouse/jobs/spark/Operations.java      | 30 +++++++++++++++----
 .../spark/SnapshotsExpirationSparkApp.java    |  3 +-
 2 files changed, 26 insertions(+), 7 deletions(-)

diff --git a/apps/spark/src/main/java/com/linkedin/openhouse/jobs/spark/Operations.java b/apps/spark/src/main/java/com/linkedin/openhouse/jobs/spark/Operations.java
index 6842690a8..7a0ad4719 100644
--- a/apps/spark/src/main/java/com/linkedin/openhouse/jobs/spark/Operations.java
+++ b/apps/spark/src/main/java/com/linkedin/openhouse/jobs/spark/Operations.java
@@ -1,5 +1,6 @@
 package com.linkedin.openhouse.jobs.spark;
 
+import com.google.common.collect.Iterators;
 import com.google.common.collect.Lists;
 import com.linkedin.openhouse.common.stats.model.IcebergTableStats;
 import com.linkedin.openhouse.jobs.util.SparkJobUtil;
@@ -207,12 +208,29 @@ public void expireSnapshots(String fqtn, long expireBeforeTimestampMs, int minVe
 
   /** Expire snapshots on a given {@link Table}. */
   public void expireSnapshots(Table table, long expireBeforeTimestampMs, int minVersions) {
-    table
-        .expireSnapshots()
-        .cleanExpiredFiles(false)
-        .expireOlderThan(expireBeforeTimestampMs)
-        .retainLast(minVersions)
-        .commit();
+    // min versions to keep not defined, expire based on timestamp of snapshots
+    if (minVersions == 0) {
+      log.info(
+          "Expiring snapshots for table: {} before timestamp: {}", table, expireBeforeTimestampMs);
+      table
+          .expireSnapshots()
+          .cleanExpiredFiles(false)
+          .expireOlderThan(expireBeforeTimestampMs)
+          .commit();
+    } else if (Iterators.size(table.snapshots().iterator()) > minVersions) {
+      log.info("Expiring snapshots for table: {} beyond last {} versions", table, minVersions);
+      table
+          .expireSnapshots()
+          .cleanExpiredFiles(false)
+          .expireOlderThan(expireBeforeTimestampMs)
+          .retainLast(minVersions)
+          .commit();
+    } else {
+      log.warn(
+          "Table {} has less than {} minimum configured versions, skipping snapshot expiration",
+          table,
+          minVersions);
+    }
   }
 
   /**
diff --git a/apps/spark/src/main/java/com/linkedin/openhouse/jobs/spark/SnapshotsExpirationSparkApp.java b/apps/spark/src/main/java/com/linkedin/openhouse/jobs/spark/SnapshotsExpirationSparkApp.java
index 1a071a155..882756b7f 100644
--- a/apps/spark/src/main/java/com/linkedin/openhouse/jobs/spark/SnapshotsExpirationSparkApp.java
+++ b/apps/spark/src/main/java/com/linkedin/openhouse/jobs/spark/SnapshotsExpirationSparkApp.java
@@ -25,7 +25,8 @@ public class SnapshotsExpirationSparkApp extends BaseTableSparkApp {
 
   private static final String DEFAULT_GRANULARITY = "days";
 
-  private static final String DEFAULT_MIN_VERSIONS = "100";
+  // Default do not define min versions, only retain snapshots based on max age
+  private static final String DEFAULT_MIN_VERSIONS = "0";
 
   public SnapshotsExpirationSparkApp(
       String jobId,

From a8cdfd3be43aa6f0efbb4d58582a02a5dba3874d Mon Sep 17 00:00:00 2001
From: Will-Lo <wlo@apache.org>
Date: Mon, 23 Dec 2024 18:31:48 -0500
Subject: [PATCH 4/8] Adds support in snapshot expiration job to use history
 policy configurations

---
 .../tasks/TableSnapshotsExpirationTask.java   |  4 +-
 .../openhouse/jobs/spark/Operations.java      | 53 ++++++-----
 .../spark/SnapshotsExpirationSparkApp.java    | 45 +++++----
 .../openhouse/jobs/util/HistoryConfig.java    |  6 +-
 .../openhouse/jobs/spark/OperationsTest.java  | 95 ++++++++++++++++---
 5 files changed, 143 insertions(+), 60 deletions(-)

diff --git a/apps/spark/src/main/java/com/linkedin/openhouse/jobs/scheduler/tasks/TableSnapshotsExpirationTask.java b/apps/spark/src/main/java/com/linkedin/openhouse/jobs/scheduler/tasks/TableSnapshotsExpirationTask.java
index 8a32eb87b..a543a1130 100644
--- a/apps/spark/src/main/java/com/linkedin/openhouse/jobs/scheduler/tasks/TableSnapshotsExpirationTask.java
+++ b/apps/spark/src/main/java/com/linkedin/openhouse/jobs/scheduler/tasks/TableSnapshotsExpirationTask.java
@@ -38,8 +38,8 @@ protected List<String> getArgs() {
               "--maxAge", Integer.toString(config.getMaxAge()),
               "--granularity", config.getGranularity().getValue()));
     }
-    if (config.getMinVersions() > 0) {
-      jobArgs.addAll(Arrays.asList("--minVersions", Integer.toString(config.getMinVersions())));
+    if (config.getVersions() > 0) {
+      jobArgs.addAll(Arrays.asList("--versions", Integer.toString(config.getVersions())));
     }
     return jobArgs;
   }
diff --git a/apps/spark/src/main/java/com/linkedin/openhouse/jobs/spark/Operations.java b/apps/spark/src/main/java/com/linkedin/openhouse/jobs/spark/Operations.java
index 7a0ad4719..c13b3eedf 100644
--- a/apps/spark/src/main/java/com/linkedin/openhouse/jobs/spark/Operations.java
+++ b/apps/spark/src/main/java/com/linkedin/openhouse/jobs/spark/Operations.java
@@ -1,6 +1,5 @@
 package com.linkedin.openhouse.jobs.spark;
 
-import com.google.common.collect.Iterators;
 import com.google.common.collect.Lists;
 import com.linkedin.openhouse.common.stats.model.IcebergTableStats;
 import com.linkedin.openhouse.jobs.util.SparkJobUtil;
@@ -21,6 +20,7 @@
 import org.apache.hadoop.fs.Path;
 import org.apache.hadoop.fs.RemoteIterator;
 import org.apache.iceberg.CatalogUtil;
+import org.apache.iceberg.ExpireSnapshots;
 import org.apache.iceberg.Schema;
 import org.apache.iceberg.StructLike;
 import org.apache.iceberg.Table;
@@ -202,34 +202,35 @@ public void deleteStagedOrphanDirectory(
   }
 
   /** Expire snapshots on a given fully-qualified table name. */
-  public void expireSnapshots(String fqtn, long expireBeforeTimestampMs, int minVersions) {
-    expireSnapshots(getTable(fqtn), expireBeforeTimestampMs, minVersions);
+  public void expireSnapshots(String fqtn, int maxAge, String granularity, int versions) {
+    expireSnapshots(getTable(fqtn), maxAge, granularity, versions);
   }
 
-  /** Expire snapshots on a given {@link Table}. */
-  public void expireSnapshots(Table table, long expireBeforeTimestampMs, int minVersions) {
-    // min versions to keep not defined, expire based on timestamp of snapshots
-    if (minVersions == 0) {
-      log.info(
-          "Expiring snapshots for table: {} before timestamp: {}", table, expireBeforeTimestampMs);
-      table
-          .expireSnapshots()
-          .cleanExpiredFiles(false)
-          .expireOlderThan(expireBeforeTimestampMs)
-          .commit();
-    } else if (Iterators.size(table.snapshots().iterator()) > minVersions) {
-      log.info("Expiring snapshots for table: {} beyond last {} versions", table, minVersions);
-      table
-          .expireSnapshots()
-          .cleanExpiredFiles(false)
-          .expireOlderThan(expireBeforeTimestampMs)
-          .retainLast(minVersions)
+  /**
+   * Expire snapshots on a given {@link Table}. If maxAge is provided, it will expire snapshots
+   * older than maxAge milliseconds. If versions is provided, it will retain the last versions
+   * snapshots. If both are provided, it will prioritize maxAge - only retain up to versions number
+   * of snapshots younger than the maxAge
+   */
+  public void expireSnapshots(Table table, int maxAge, String granularity, int versions) {
+    ExpireSnapshots expireSnapshotsCommand = table.expireSnapshots().cleanExpiredFiles(false);
+
+    // maxAge is always defined with granularity
+    if (!granularity.isEmpty()) {
+      TimeUnit timeUnitGranularity = TimeUnit.valueOf(granularity.toUpperCase());
+      long expireBeforeTimestampMs =
+          System.currentTimeMillis() - timeUnitGranularity.toMillis(maxAge);
+      log.info("Expiring snapshots for table: {} older than {}ms", table, expireBeforeTimestampMs);
+      expireSnapshotsCommand.expireOlderThan(expireBeforeTimestampMs).commit();
+    }
+    if (versions > 0) {
+      log.info("Expiring snapshots for table: {} retaining last {} versions", table, versions);
+      // Note: retainLast keeps the last N snapshots that WOULD be expired, hence expireOlderThan
+      // currentTime
+      expireSnapshotsCommand
+          .expireOlderThan(System.currentTimeMillis())
+          .retainLast(versions)
           .commit();
-    } else {
-      log.warn(
-          "Table {} has less than {} minimum configured versions, skipping snapshot expiration",
-          table,
-          minVersions);
     }
   }
 
diff --git a/apps/spark/src/main/java/com/linkedin/openhouse/jobs/spark/SnapshotsExpirationSparkApp.java b/apps/spark/src/main/java/com/linkedin/openhouse/jobs/spark/SnapshotsExpirationSparkApp.java
index 882756b7f..b858db8d3 100644
--- a/apps/spark/src/main/java/com/linkedin/openhouse/jobs/spark/SnapshotsExpirationSparkApp.java
+++ b/apps/spark/src/main/java/com/linkedin/openhouse/jobs/spark/SnapshotsExpirationSparkApp.java
@@ -3,7 +3,6 @@
 import com.linkedin.openhouse.jobs.spark.state.StateManager;
 import java.util.ArrayList;
 import java.util.List;
-import java.util.concurrent.TimeUnit;
 import lombok.extern.slf4j.Slf4j;
 import org.apache.commons.cli.CommandLine;
 import org.apache.commons.cli.Option;
@@ -19,14 +18,18 @@
 public class SnapshotsExpirationSparkApp extends BaseTableSparkApp {
   private final String granularity;
   private final int maxAge;
-  private final int minVersions;
+  private final int versions;
 
-  private static final String DEFAULT_MAX_AGE = "3";
+  public static class DEFAULT_CONFIGURATION {
+    public static final String MAX_AGE = "3";
+    public static final String GRANULARITY = "DAYS";
+    public static final String VERSIONS = "0";
+  }
 
-  private static final String DEFAULT_GRANULARITY = "days";
+  private static final String DEFAULT_GRANULARITY = "";
 
-  // Default do not define min versions, only retain snapshots based on max age
-  private static final String DEFAULT_MIN_VERSIONS = "0";
+  // By default do not define versions, and only retain snapshots based on max age
+  private static final String DEFAULT_VERSIONS = "0";
 
   public SnapshotsExpirationSparkApp(
       String jobId,
@@ -34,11 +37,17 @@ public SnapshotsExpirationSparkApp(
       String fqtn,
       int maxAge,
       String granularity,
-      int minVersions) {
+      int versions) {
     super(jobId, stateManager, fqtn);
-    this.granularity = granularity;
-    this.maxAge = maxAge;
-    this.minVersions = minVersions;
+    if (maxAge == 0 && versions == 0) {
+      this.maxAge = Integer.parseInt(DEFAULT_CONFIGURATION.MAX_AGE);
+      this.granularity = DEFAULT_CONFIGURATION.GRANULARITY;
+      this.versions = Integer.parseInt(DEFAULT_CONFIGURATION.VERSIONS);
+    } else {
+      this.granularity = granularity;
+      this.maxAge = maxAge;
+      this.versions = versions;
+    }
   }
 
   @Override
@@ -48,10 +57,8 @@ protected void runInner(Operations ops) {
         fqtn,
         maxAge,
         granularity,
-        minVersions);
-    long expireBeforeTimestampMs = System.currentTimeMillis() - TimeUnit.DAYS.toMillis(maxAge);
-    log.info("Expire snapshots before timestamp ms {}", expireBeforeTimestampMs);
-    ops.expireSnapshots(fqtn, expireBeforeTimestampMs, minVersions);
+        versions);
+    ops.expireSnapshots(fqtn, maxAge, granularity, versions);
   }
 
   public static void main(String[] args) {
@@ -60,16 +67,18 @@ public static void main(String[] args) {
     extraOptions.add(
         new Option("a", "maxAge", true, "Delete snapshots older than <maxAge> <granularity>s"));
     extraOptions.add(new Option("g", "granularity", true, "Granularity: day"));
-    extraOptions.add(new Option("v", "minVersions", true, "Minimum number of versions to keep"));
+    extraOptions.add(
+        new Option("v", "versions", true, "Number of versions to keep after snapshot expiration"));
     CommandLine cmdLine = createCommandLine(args, extraOptions);
+
     SnapshotsExpirationSparkApp app =
         new SnapshotsExpirationSparkApp(
             getJobId(cmdLine),
             createStateManager(cmdLine),
             cmdLine.getOptionValue("tableName"),
-            Integer.parseInt(cmdLine.getOptionValue("maxAge", DEFAULT_MAX_AGE)),
-            cmdLine.getOptionValue("granularity", DEFAULT_GRANULARITY),
-            Integer.parseInt(cmdLine.getOptionValue("minVersions", DEFAULT_MIN_VERSIONS)));
+            Integer.parseInt(cmdLine.getOptionValue("maxAge", "0")),
+            cmdLine.getOptionValue("granularity", ""),
+            Integer.parseInt(cmdLine.getOptionValue("minVersions", "0")));
     app.run();
   }
 }
diff --git a/apps/spark/src/main/java/com/linkedin/openhouse/jobs/util/HistoryConfig.java b/apps/spark/src/main/java/com/linkedin/openhouse/jobs/util/HistoryConfig.java
index fe649fc26..aa27816f2 100644
--- a/apps/spark/src/main/java/com/linkedin/openhouse/jobs/util/HistoryConfig.java
+++ b/apps/spark/src/main/java/com/linkedin/openhouse/jobs/util/HistoryConfig.java
@@ -1,6 +1,6 @@
 package com.linkedin.openhouse.jobs.util;
 
-import com.linkedin.openhouse.tables.client.model.Retention;
+import com.linkedin.openhouse.tables.client.model.History;
 import lombok.Builder;
 import lombok.EqualsAndHashCode;
 import lombok.Getter;
@@ -13,6 +13,6 @@
 @ToString
 public class HistoryConfig {
   private final int maxAge;
-  private final int minVersions;
-  private final Retention.GranularityEnum granularity;
+  private final int versions;
+  private final History.GranularityEnum granularity;
 }
diff --git a/apps/spark/src/test/java/com/linkedin/openhouse/jobs/spark/OperationsTest.java b/apps/spark/src/test/java/com/linkedin/openhouse/jobs/spark/OperationsTest.java
index 164d26398..648579893 100644
--- a/apps/spark/src/test/java/com/linkedin/openhouse/jobs/spark/OperationsTest.java
+++ b/apps/spark/src/test/java/com/linkedin/openhouse/jobs/spark/OperationsTest.java
@@ -29,8 +29,6 @@
 import org.apache.iceberg.types.Types;
 import org.apache.spark.sql.Row;
 import org.assertj.core.util.Lists;
-import org.joda.time.DurationFieldType;
-import org.joda.time.Hours;
 import org.junit.jupiter.api.Assertions;
 import org.junit.jupiter.api.Test;
 
@@ -275,8 +273,44 @@ public void testOrphanFilesDeletionNoStaging() throws Exception {
 
   @Test
   public void testSnapshotsExpirationMaxAge() throws Exception {
+    final String tableName = "db.test_es_maxage_noop_java";
+    final int numInserts = 3;
+    final int maxAge = 0;
+    // Not a realistic time setting that is accepted by the SQL, but tests that other snapshots are
+    // deleted
+    final String timeGranularity = "DAYS";
+
+    List<Long> snapshotIds;
+    try (Operations ops = Operations.withCatalog(getSparkSession(), meter)) {
+      prepareTable(ops, tableName);
+      populateTable(ops, tableName, numInserts);
+      snapshotIds = getSnapshotIds(ops, tableName);
+      Assertions.assertEquals(
+          numInserts,
+          snapshotIds.size(),
+          String.format("There must be %d snapshot(s) after inserts", numInserts));
+      Table table = ops.getTable(tableName);
+      log.info("Loaded table {}, location {}", table.name(), table.location());
+
+      ops.expireSnapshots(table, maxAge, timeGranularity, 0);
+      // No snapshots should be cleaned up as they are all within the max age
+      checkSnapshots(table, snapshotIds.subList(snapshotIds.size() - 1, snapshotIds.size()));
+    }
+    // restart the app to reload catalog cache
+    try (Operations ops = Operations.withCatalog(getSparkSession(), meter)) {
+      // verify that new apps see snapshots correctly
+      checkSnapshots(
+          ops, tableName, snapshotIds.subList(snapshotIds.size() - 1, snapshotIds.size()));
+    }
+  }
+
+  @Test
+  public void testSnapshotsExpirationMaxAgeNoop() throws Exception {
     final String tableName = "db.test_es_maxage_java";
     final int numInserts = 3;
+    final int maxAge = 3;
+    final String timeGranularity = "DAYS";
+
     List<Long> snapshotIds;
     try (Operations ops = Operations.withCatalog(getSparkSession(), meter)) {
       prepareTable(ops, tableName);
@@ -288,21 +322,57 @@ public void testSnapshotsExpirationMaxAge() throws Exception {
           String.format("There must be %d snapshot(s) after inserts", numInserts));
       Table table = ops.getTable(tableName);
       log.info("Loaded table {}, location {}", table.name(), table.location());
-      ops.expireSnapshots(table, System.currentTimeMillis(), 3);
+
+      ops.expireSnapshots(table, maxAge, timeGranularity, 0);
+      // No snapshots should be cleaned up as they are all within the max age
+      checkSnapshots(table, snapshotIds);
+    }
+    // restart the app to reload catalog cache
+    try (Operations ops = Operations.withCatalog(getSparkSession(), meter)) {
+      // verify that new apps see snapshots correctly
+      checkSnapshots(ops, tableName, snapshotIds);
+    }
+  }
+
+  @Test
+  public void testSnapshotsExpirationVersions() throws Exception {
+    final String tableName = "db.test_es_versions_java";
+    final int numInserts = 3;
+    final int versionsToKeep = 2;
+    List<Long> snapshotIds;
+    try (Operations ops = Operations.withCatalog(getSparkSession(), meter)) {
+      prepareTable(ops, tableName);
+      populateTable(ops, tableName, numInserts);
+      snapshotIds = getSnapshotIds(ops, tableName);
+      Assertions.assertEquals(
+          numInserts,
+          snapshotIds.size(),
+          String.format("There must be %d snapshot(s) after inserts", numInserts));
+      Table table = ops.getTable(tableName);
+      log.info("Loaded table {}, location {}", table.name(), table.location());
+
+      ops.expireSnapshots(table, 0, "", versionsToKeep);
       // verify that table object snapshots are updated
-      checkSnapshots(table, snapshotIds.subList(2, snapshotIds.size()));
+      checkSnapshots(
+          table, snapshotIds.subList(snapshotIds.size() - versionsToKeep, snapshotIds.size()));
     }
     // restart the app to reload catalog cache
     try (Operations ops = Operations.withCatalog(getSparkSession(), meter)) {
       // verify that new apps see snapshots correctly
-      checkSnapshots(ops, tableName, snapshotIds.subList(2, snapshotIds.size()));
+      checkSnapshots(
+          ops,
+          tableName,
+          snapshotIds.subList(snapshotIds.size() - versionsToKeep, snapshotIds.size()));
     }
   }
 
   @Test
-  public void testSnapshotsExpirationMinVersions() throws Exception {
-    final String tableName = "db.test_es_minversions_java";
+  public void testSnapshotsExpirationBothAgeAndVersions() throws Exception {
+    final String tableName = "db.test_es_age_and_versions_java";
     final int numInserts = 3;
+    final int maxAge = 3;
+    final String timeGranularity = "DAYS";
+    final int versionsToKeep = 1;
     List<Long> snapshotIds;
     try (Operations ops = Operations.withCatalog(getSparkSession(), meter)) {
       prepareTable(ops, tableName);
@@ -315,15 +385,18 @@ public void testSnapshotsExpirationMinVersions() throws Exception {
       Table table = ops.getTable(tableName);
       log.info("Loaded table {}, location {}", table.name(), table.location());
 
-      ops.expireSnapshots(
-          table, System.currentTimeMillis() + Hours.ONE.get(DurationFieldType.millis()), 1);
+      ops.expireSnapshots(table, maxAge, timeGranularity, versionsToKeep);
       // verify that table object snapshots are updated
-      checkSnapshots(table, snapshotIds.subList(2, snapshotIds.size()));
+      checkSnapshots(
+          table, snapshotIds.subList(snapshotIds.size() - versionsToKeep, snapshotIds.size()));
     }
     // restart the app to reload catalog cache
     try (Operations ops = Operations.withCatalog(getSparkSession(), meter)) {
       // verify that new apps see snapshots correctly
-      checkSnapshots(ops, tableName, snapshotIds.subList(2, snapshotIds.size()));
+      checkSnapshots(
+          ops,
+          tableName,
+          snapshotIds.subList(snapshotIds.size() - versionsToKeep, snapshotIds.size()));
     }
   }
 

From 1f11e110418fe6f5d71824ef2878c18bbecaacdd Mon Sep 17 00:00:00 2001
From: Will-Lo <wlo@apache.org>
Date: Mon, 23 Dec 2024 19:09:11 -0500
Subject: [PATCH 5/8] Improve comments/javadoc

---
 .../java/com/linkedin/openhouse/jobs/spark/Operations.java  | 6 +++---
 .../com/linkedin/openhouse/jobs/util/HistoryConfig.java     | 2 +-
 .../com/linkedin/openhouse/jobs/spark/OperationsTest.java   | 4 ++--
 3 files changed, 6 insertions(+), 6 deletions(-)

diff --git a/apps/spark/src/main/java/com/linkedin/openhouse/jobs/spark/Operations.java b/apps/spark/src/main/java/com/linkedin/openhouse/jobs/spark/Operations.java
index c13b3eedf..73e36abe9 100644
--- a/apps/spark/src/main/java/com/linkedin/openhouse/jobs/spark/Operations.java
+++ b/apps/spark/src/main/java/com/linkedin/openhouse/jobs/spark/Operations.java
@@ -208,9 +208,9 @@ public void expireSnapshots(String fqtn, int maxAge, String granularity, int ver
 
   /**
    * Expire snapshots on a given {@link Table}. If maxAge is provided, it will expire snapshots
-   * older than maxAge milliseconds. If versions is provided, it will retain the last versions
-   * snapshots. If both are provided, it will prioritize maxAge - only retain up to versions number
-   * of snapshots younger than the maxAge
+   * older than maxAge in granularity timeunit. If versions is provided, it will retain the last
+   * versions snapshots. If both are provided, it will prioritize maxAge; only retain up to versions
+   * number of snapshots younger than the maxAge
    */
   public void expireSnapshots(Table table, int maxAge, String granularity, int versions) {
     ExpireSnapshots expireSnapshotsCommand = table.expireSnapshots().cleanExpiredFiles(false);
diff --git a/apps/spark/src/main/java/com/linkedin/openhouse/jobs/util/HistoryConfig.java b/apps/spark/src/main/java/com/linkedin/openhouse/jobs/util/HistoryConfig.java
index aa27816f2..6d116d200 100644
--- a/apps/spark/src/main/java/com/linkedin/openhouse/jobs/util/HistoryConfig.java
+++ b/apps/spark/src/main/java/com/linkedin/openhouse/jobs/util/HistoryConfig.java
@@ -6,7 +6,7 @@
 import lombok.Getter;
 import lombok.ToString;
 
-/** Table retention config class. This is app side representation of /tables policies->retention */
+/** History Policy config class. This is app side representation of /tables policies->history */
 @Builder
 @Getter
 @EqualsAndHashCode
diff --git a/apps/spark/src/test/java/com/linkedin/openhouse/jobs/spark/OperationsTest.java b/apps/spark/src/test/java/com/linkedin/openhouse/jobs/spark/OperationsTest.java
index 648579893..1ad433f44 100644
--- a/apps/spark/src/test/java/com/linkedin/openhouse/jobs/spark/OperationsTest.java
+++ b/apps/spark/src/test/java/com/linkedin/openhouse/jobs/spark/OperationsTest.java
@@ -277,7 +277,7 @@ public void testSnapshotsExpirationMaxAge() throws Exception {
     final int numInserts = 3;
     final int maxAge = 0;
     // Not a realistic time setting that is accepted by the SQL, but tests that other snapshots are
-    // deleted
+    // deleted by the time history policy
     final String timeGranularity = "DAYS";
 
     List<Long> snapshotIds;
@@ -293,7 +293,7 @@ public void testSnapshotsExpirationMaxAge() throws Exception {
       log.info("Loaded table {}, location {}", table.name(), table.location());
 
       ops.expireSnapshots(table, maxAge, timeGranularity, 0);
-      // No snapshots should be cleaned up as they are all within the max age
+      // Only retain the last snapshot
       checkSnapshots(table, snapshotIds.subList(snapshotIds.size() - 1, snapshotIds.size()));
     }
     // restart the app to reload catalog cache

From 13c47e363c496113345edd8d5b4519dc6e8e66e9 Mon Sep 17 00:00:00 2001
From: Will-Lo <wlo@apache.org>
Date: Thu, 2 Jan 2025 18:09:00 -0500
Subject: [PATCH 6/8] Fix bug where retainLast would throw an exception if
 versions exceeded number of snapshots

---
 .../openhouse/jobs/spark/Operations.java      |  3 +-
 .../openhouse/jobs/spark/OperationsTest.java  | 69 +++++++++++++++++++
 2 files changed, 71 insertions(+), 1 deletion(-)

diff --git a/apps/spark/src/main/java/com/linkedin/openhouse/jobs/spark/Operations.java b/apps/spark/src/main/java/com/linkedin/openhouse/jobs/spark/Operations.java
index 73e36abe9..2b86ec38e 100644
--- a/apps/spark/src/main/java/com/linkedin/openhouse/jobs/spark/Operations.java
+++ b/apps/spark/src/main/java/com/linkedin/openhouse/jobs/spark/Operations.java
@@ -1,5 +1,6 @@
 package com.linkedin.openhouse.jobs.spark;
 
+import com.google.common.collect.Iterators;
 import com.google.common.collect.Lists;
 import com.linkedin.openhouse.common.stats.model.IcebergTableStats;
 import com.linkedin.openhouse.jobs.util.SparkJobUtil;
@@ -223,7 +224,7 @@ public void expireSnapshots(Table table, int maxAge, String granularity, int ver
       log.info("Expiring snapshots for table: {} older than {}ms", table, expireBeforeTimestampMs);
       expireSnapshotsCommand.expireOlderThan(expireBeforeTimestampMs).commit();
     }
-    if (versions > 0) {
+    if (versions > 0 && Iterators.size(table.snapshots().iterator()) > versions) {
       log.info("Expiring snapshots for table: {} retaining last {} versions", table, versions);
       // Note: retainLast keeps the last N snapshots that WOULD be expired, hence expireOlderThan
       // currentTime
diff --git a/apps/spark/src/test/java/com/linkedin/openhouse/jobs/spark/OperationsTest.java b/apps/spark/src/test/java/com/linkedin/openhouse/jobs/spark/OperationsTest.java
index 1ad433f44..5c4ac1999 100644
--- a/apps/spark/src/test/java/com/linkedin/openhouse/jobs/spark/OperationsTest.java
+++ b/apps/spark/src/test/java/com/linkedin/openhouse/jobs/spark/OperationsTest.java
@@ -334,6 +334,34 @@ public void testSnapshotsExpirationMaxAgeNoop() throws Exception {
     }
   }
 
+  @Test
+  public void testSnapshotsExpirationVersionsNoop() throws Exception {
+    final String tableName = "db.test_es_versions_noop_java";
+    final int numInserts = 3;
+    final int versionsToKeep = 5; // Should keep all versions given that there are fewer versions
+    List<Long> snapshotIds;
+    try (Operations ops = Operations.withCatalog(getSparkSession(), meter)) {
+      prepareTable(ops, tableName);
+      populateTable(ops, tableName, numInserts);
+      snapshotIds = getSnapshotIds(ops, tableName);
+      Assertions.assertEquals(
+          numInserts,
+          snapshotIds.size(),
+          String.format("There must be %d snapshot(s) after inserts", numInserts));
+      Table table = ops.getTable(tableName);
+      log.info("Loaded table {}, location {}", table.name(), table.location());
+
+      ops.expireSnapshots(table, 0, "", versionsToKeep);
+      // verify that table object snapshots are updated
+      checkSnapshots(table, snapshotIds);
+    }
+    // restart the app to reload catalog cache
+    try (Operations ops = Operations.withCatalog(getSparkSession(), meter)) {
+      // verify that new apps see snapshots correctly
+      checkSnapshots(ops, tableName, snapshotIds);
+    }
+  }
+
   @Test
   public void testSnapshotsExpirationVersions() throws Exception {
     final String tableName = "db.test_es_versions_java";
@@ -400,6 +428,47 @@ public void testSnapshotsExpirationBothAgeAndVersions() throws Exception {
     }
   }
 
+  @Test
+  public void testSnapshotsExpirationPrioritizeAge() throws Exception {
+    final String tableName = "db.test_es_age_prioritization_java";
+    final int numInserts = 3;
+    final int maxAge = 20;
+    final String timeGranularity =
+        "SECONDS"; // Not a realistic user configuration, for the sake of testing
+    final int versionsToKeep = 5;
+    final int versionsToKeepAfterExpiration = 2;
+    List<Long> snapshotIds;
+    try (Operations ops = Operations.withCatalog(getSparkSession(), meter)) {
+      prepareTable(ops, tableName);
+      populateTable(ops, tableName, numInserts);
+      Thread.sleep(30000); // Sleep for 30 seconds to expire all the snapshots
+      populateTable(ops, tableName, versionsToKeepAfterExpiration);
+      snapshotIds = getSnapshotIds(ops, tableName);
+      Assertions.assertEquals(
+          numInserts + versionsToKeepAfterExpiration,
+          snapshotIds.size(),
+          String.format("There must be %d snapshot(s) after inserts", numInserts));
+      Table table = ops.getTable(tableName);
+      log.info("Loaded table {}, location {}", table.name(), table.location());
+
+      ops.expireSnapshots(table, maxAge, timeGranularity, versionsToKeep);
+      // verify that only 2 snapshots are kept instead of 5 due to prioritizing maxAge
+      checkSnapshots(
+          table,
+          snapshotIds.subList(
+              snapshotIds.size() - versionsToKeepAfterExpiration, snapshotIds.size()));
+    }
+    // restart the app to reload catalog cache
+    try (Operations ops = Operations.withCatalog(getSparkSession(), meter)) {
+      // verify that new apps see snapshots correctly
+      checkSnapshots(
+          ops,
+          tableName,
+          snapshotIds.subList(
+              snapshotIds.size() - versionsToKeepAfterExpiration, snapshotIds.size()));
+    }
+  }
+
   @Test
   public void testStagedFilesDelete() throws Exception {
     final String tableName = "db.test_staged_delete";

From 2942b7746f6691f3b4dfa5bf1b9245f3f654f185 Mon Sep 17 00:00:00 2001
From: Will-Lo <wlo@apache.org>
Date: Fri, 3 Jan 2025 17:53:00 -0500
Subject: [PATCH 7/8] Fix table test names

---
 .../com/linkedin/openhouse/jobs/spark/OperationsTest.java     | 4 ++--
 1 file changed, 2 insertions(+), 2 deletions(-)

diff --git a/apps/spark/src/test/java/com/linkedin/openhouse/jobs/spark/OperationsTest.java b/apps/spark/src/test/java/com/linkedin/openhouse/jobs/spark/OperationsTest.java
index 5c4ac1999..b4156de62 100644
--- a/apps/spark/src/test/java/com/linkedin/openhouse/jobs/spark/OperationsTest.java
+++ b/apps/spark/src/test/java/com/linkedin/openhouse/jobs/spark/OperationsTest.java
@@ -273,7 +273,7 @@ public void testOrphanFilesDeletionNoStaging() throws Exception {
 
   @Test
   public void testSnapshotsExpirationMaxAge() throws Exception {
-    final String tableName = "db.test_es_maxage_noop_java";
+    final String tableName = "db.test_es_maxage_java";
     final int numInserts = 3;
     final int maxAge = 0;
     // Not a realistic time setting that is accepted by the SQL, but tests that other snapshots are
@@ -306,7 +306,7 @@ public void testSnapshotsExpirationMaxAge() throws Exception {
 
   @Test
   public void testSnapshotsExpirationMaxAgeNoop() throws Exception {
-    final String tableName = "db.test_es_maxage_java";
+    final String tableName = "db.test_es_maxage_noop_java";
     final int numInserts = 3;
     final int maxAge = 3;
     final String timeGranularity = "DAYS";

From 2f4de1be042ebd486dc54bcd59ffd918d660c06d Mon Sep 17 00:00:00 2001
From: Will-Lo <wlo@apache.org>
Date: Mon, 6 Jan 2025 13:13:44 -0500
Subject: [PATCH 8/8] Change string type of defaults to int

---
 .../openhouse/jobs/spark/SnapshotsExpirationSparkApp.java | 8 ++++----
 1 file changed, 4 insertions(+), 4 deletions(-)

diff --git a/apps/spark/src/main/java/com/linkedin/openhouse/jobs/spark/SnapshotsExpirationSparkApp.java b/apps/spark/src/main/java/com/linkedin/openhouse/jobs/spark/SnapshotsExpirationSparkApp.java
index b858db8d3..1d653858b 100644
--- a/apps/spark/src/main/java/com/linkedin/openhouse/jobs/spark/SnapshotsExpirationSparkApp.java
+++ b/apps/spark/src/main/java/com/linkedin/openhouse/jobs/spark/SnapshotsExpirationSparkApp.java
@@ -21,9 +21,9 @@ public class SnapshotsExpirationSparkApp extends BaseTableSparkApp {
   private final int versions;
 
   public static class DEFAULT_CONFIGURATION {
-    public static final String MAX_AGE = "3";
+    public static final int MAX_AGE = 3;
     public static final String GRANULARITY = "DAYS";
-    public static final String VERSIONS = "0";
+    public static final int VERSIONS = 0;
   }
 
   private static final String DEFAULT_GRANULARITY = "";
@@ -40,9 +40,9 @@ public SnapshotsExpirationSparkApp(
       int versions) {
     super(jobId, stateManager, fqtn);
     if (maxAge == 0 && versions == 0) {
-      this.maxAge = Integer.parseInt(DEFAULT_CONFIGURATION.MAX_AGE);
+      this.maxAge = DEFAULT_CONFIGURATION.MAX_AGE;
       this.granularity = DEFAULT_CONFIGURATION.GRANULARITY;
-      this.versions = Integer.parseInt(DEFAULT_CONFIGURATION.VERSIONS);
+      this.versions = DEFAULT_CONFIGURATION.VERSIONS;
     } else {
       this.granularity = granularity;
       this.maxAge = maxAge;