throw exception incase no snapshot found for copying

Author: Blazer-007

gobblin-data-management/src/main/java/org/apache/gobblin/data/management/copy/iceberg/IcebergTable.java

@@ -239,7 +239,8 @@ protected void registerIcebergTable(TableMetadata srcMetadata, TableMetadata dst
   public List<DataFile> getPartitionSpecificDataFiles(Predicate<StructLike> icebergPartitionFilterPredicate)
       throws IOException {
     TableMetadata tableMetadata = accessTableMetadata();
-    Snapshot currentSnapshot = tableMetadata.currentSnapshot();
+    Snapshot currentSnapshot = Optional.ofNullable(tableMetadata.currentSnapshot())
+        .orElseThrow(() -> new IOException(String.format("~%s~ No snapshots found", tableId)));
     long currentSnapshotId = currentSnapshot.snapshotId();
     List<DataFile> knownDataFiles = new ArrayList<>();
     GrowthMilestoneTracker growthMilestoneTracker = new GrowthMilestoneTracker();

gobblin-data-management/src/test/java/org/apache/gobblin/data/management/copy/iceberg/IcebergTableTest.java

@@ -253,6 +253,18 @@ public void testGetPartitionSpecificDataFiles() throws IOException {
     Assert.assertEquals(icebergTable.getPartitionSpecificDataFiles(alwaysFalsePredicate).size(), 0);
   }
 
+  @Test
+  public void testGetPartitionSpecificDataFilesThrowsExceptionWhenNoSnapshotsFound() throws IOException {
+    IcebergTable icebergTable = new IcebergTable(tableId,
+        catalog.newTableOps(tableId),
+        catalogUri,
+        catalog.loadTable(tableId));
+    // Using AlwaysTrue Predicate to avoid mocking of predicate class
+    Predicate<StructLike> alwaysTruePredicate = partition -> true;
+    IOException exception = Assert.expectThrows(IOException.class, () -> icebergTable.getPartitionSpecificDataFiles(alwaysTruePredicate));
+    Assert.assertEquals(exception.getMessage(), String.format("~%s~ No snapshots found", tableId));
+  }
+
   /** Verify that overwritePartition replace data files belonging to given partition col and value */
   @Test
   public void testOverwritePartition() throws IOException {