Fix bug where azkaban prop.contains(object) is checking value not key as intended

Author: linweihs

gobblin-data-management/src/main/java/org/apache/gobblin/data/management/trash/TimeBasedSnapshotCleanupPolicy.java

@@ -21,7 +21,6 @@
 
 import org.apache.hadoop.fs.FileStatus;
 import org.joda.time.DateTime;
-import org.joda.time.DateTimeZone;
 
 
 /**
@@ -31,21 +30,17 @@ public class TimeBasedSnapshotCleanupPolicy implements SnapshotCleanupPolicy {
 
   public static final String SNAPSHOT_RETENTION_POLICY_MINUTES_KEY = "gobblin.trash.snapshot.retention.minutes";
   public static final int SNAPSHOT_RETENTION_POLICY_MINUTES_DEFAULT = 1440; // one day
-  public static final String RETENTION_SNAPSHOT_TIMEZONE = "gobblin.trash.snapshot.retention.timezone";
 
-  protected final int retentionMinutes;
-  protected final DateTimeZone retentionSnapshotTimezone;
+  private final int retentionMinutes;
 
   public TimeBasedSnapshotCleanupPolicy(Properties props) {
     this.retentionMinutes = Integer.parseInt(props.getProperty(SNAPSHOT_RETENTION_POLICY_MINUTES_KEY,
         Integer.toString(SNAPSHOT_RETENTION_POLICY_MINUTES_DEFAULT)));
-    this.retentionSnapshotTimezone = props.containsKey(RETENTION_SNAPSHOT_TIMEZONE) ? DateTimeZone.forID(props.getProperty(RETENTION_SNAPSHOT_TIMEZONE)) : DateTimeZone.getDefault();
   }
 
   @Override
   public boolean shouldDeleteSnapshot(FileStatus snapshot, Trash trash) {
     DateTime snapshotTime = Trash.TRASH_SNAPSHOT_NAME_FORMATTER.parseDateTime(snapshot.getPath().getName());
-    // Ensure that the comparison between snapshotTime and the current time can be done in the same time zone (UTC) with timezone specified in the propertie (Default to system time zone)
-    return snapshotTime.plusMinutes(this.retentionMinutes).isBefore(DateTime.now(this.retentionSnapshotTimezone));
+    return snapshotTime.plusMinutes(this.retentionMinutes).isBeforeNow();
   }
-}
+}

gobblin-data-management/src/main/java/org/apache/gobblin/data/management/trash/Trash.java

@@ -178,7 +178,7 @@ protected Trash(FileSystem fs, Properties props, String user) throws IOException
   }
 
   public static Trash getTrash(FileSystem fs, Properties props, String user) throws IOException {
-    if (props.contains(TRASH_CLASS_KEY)) {
+    if (props.containsKey(TRASH_CLASS_KEY)) {
       return GobblinConstructorUtils.invokeConstructor(Trash.class, props.getProperty(TRASH_CLASS_KEY), fs, props, user);
     } else {
       return new Trash(fs, props, user);

gobblin-data-management/src/test/java/org/apache/gobblin/data/management/trash/TimeBasedSnapshotCleanupPolicyTest.java

@@ -30,16 +30,15 @@
 
 public class TimeBasedSnapshotCleanupPolicyTest {
 
-    private MockTimeBasedSnapshotCleanupPolicy cleanupPolicy;
+    private TimeBasedSnapshotCleanupPolicy cleanupPolicy;
 
     @BeforeMethod
     public void setUp()  {
-        // Initialize the cleanup policy with a retention period (e.g., 1 day)
+        // Initialize the cleanup policy with a retention period of 600 minutes (10 hours)
         Properties properties = new Properties();
-        properties.setProperty(MockTimeBasedSnapshotCleanupPolicy.SNAPSHOT_RETENTION_POLICY_MINUTES_KEY, "1440"); // 1440 minutes = 1 day
-        properties.setProperty(MockTimeBasedSnapshotCleanupPolicy.RETENTION_SNAPSHOT_TIMEZONE, "UTC");
-        // Mock the cutoff time to be 2024-10-30 01:01:00 UTC
-        cleanupPolicy = new MockTimeBasedSnapshotCleanupPolicy(properties, new DateTime(2024, 10, 30, 1, 1));
+        properties.setProperty(TimeBasedSnapshotCleanupPolicy.SNAPSHOT_RETENTION_POLICY_MINUTES_KEY, "600"); // 600 minutes = 10 hours
+        cleanupPolicy = new TimeBasedSnapshotCleanupPolicy(properties);
+
     }
 
     @Test
@@ -49,42 +48,18 @@ public void testShouldDeleteSnapshot() throws IOException {
         TrashTestBase trashTestBase = new TrashTestBase(new Properties());
         Trash trash = trashTestBase.trash;
 
-        // Create dummy paths
-        FileStatus fs1 = new FileStatus(0, true, 0, 0, 0, 
-                                        new Path(trash.getTrashLocation(), new DateTime(2024, 10, 29, 1, 0, DateTimeZone.UTC).toString(Trash.TRASH_SNAPSHOT_NAME_FORMATTER)));
-        FileStatus fs2 = new FileStatus(0, true, 0, 0, 0, 
-                                        new Path(trash.getTrashLocation(), new DateTime(2024, 10, 29, 2, 0, DateTimeZone.UTC).toString(Trash.TRASH_SNAPSHOT_NAME_FORMATTER)));
+        // Get the current time
+        DateTime now = DateTime.now(DateTimeZone.UTC);
 
-        // Test old snapshot (should be deleted)
-        // 2024-10-29 01:00:00 UTC + 1440 minutes = 2024-10-30 01:00:00 UTC < Cutoff time a.k.a system current_time (2024-10-30 01:01:00 UTC)
-        Assert.assertTrue(cleanupPolicy.shouldDeleteSnapshot(fs1, trash), "Old snapshot should be deleted");
+        // Create dummy paths with timestamps between 11 and 9 hours ago in UTC
+        FileStatus fs1 = new FileStatus(0, true, 0, 0, 0,
+                new Path(trash.getTrashLocation(), now.minusHours(11).toString(Trash.TRASH_SNAPSHOT_NAME_FORMATTER)));
+        FileStatus fs2 = new FileStatus(0, true, 0, 0, 0,
+                new Path(trash.getTrashLocation(), now.minusHours(9).toString(Trash.TRASH_SNAPSHOT_NAME_FORMATTER)));           
 
-        // Test snapshot (should not be deleted)
-        // 2024-10-29 02:00:00 UTC + 1440 minutes = 2024-10-30 02:00:00 UTC > Cutoff time a.k.a system current_time (2024-10-30 01:01:00 UTC)
-        Assert.assertFalse(cleanupPolicy.shouldDeleteSnapshot(fs2, trash), "snapshot should not be deleted");
+        // Test snapshot (should be deleted)
+        Assert.assertTrue(cleanupPolicy.shouldDeleteSnapshot(fs1, trash),"Snapshot should be deleted");
+        // Test snapshot (should NOT be deleted)
+        Assert.assertFalse(cleanupPolicy.shouldDeleteSnapshot(fs2, trash),"Snapshot should NOT be deleted");
     }
-
-    /**
-     * Mock the TimeBasedSnapshotCleanupPolicy for testing purposes
-     * 
-     * In this class, the current time used in the comparison method isBefore() can be mocked 
-     * Why? The current time is used to determine if a snapshot is older than the retention period, 
-     * and given that the current time is always changing, it is difficult to test the method shouldDeleteSnapshot()
-     */
-    public class MockTimeBasedSnapshotCleanupPolicy extends TimeBasedSnapshotCleanupPolicy {
-
-        private final DateTime MOCK_CURRENT_TIME;
-
-        public MockTimeBasedSnapshotCleanupPolicy(Properties props, DateTime mockCurrentTime) {
-            super(props);
-            this.MOCK_CURRENT_TIME = mockCurrentTime;
-        }
-
-        @Override
-        public boolean shouldDeleteSnapshot(FileStatus snapshot, Trash trash) {
-            DateTime snapshotTime = Trash.TRASH_SNAPSHOT_NAME_FORMATTER.parseDateTime(snapshot.getPath().getName());
-            return snapshotTime.plusMinutes(this.retentionMinutes).isBefore(this.MOCK_CURRENT_TIME.withZoneRetainFields(this.retentionSnapshotTimezone));
-        }
-    }
-
 }

gobblin-data-management/src/test/java/org/apache/gobblin/data/management/trash/TrashTest.java

@@ -318,4 +318,27 @@ public Boolean answer(InvocationOnMock invocation)
     }
   }
 
+  /**
+   * Test Trash.getTrash() method to ensure that the correct Trash implementation is returned based on the properties evaluation
+   */
+  @Test
+  public void testGetTrashEvanluateExpected() throws IOException {
+    Properties properties = new Properties();
+    properties.setProperty(Trash.TRASH_CLASS_KEY, MockTrash.class.getCanonicalName());
+    properties.setProperty(TrashFactory.SIMULATE, "true");
+    properties.setProperty(TrashFactory.SIMULATE_USING_ACTUAL_TRASH, "true");
+    properties.setProperty(Trash.TRASH_LOCATION_KEY, "/trash/dir");
+    FileSystem mockFs = mock(FileSystem.class);
+
+    Trash trash = Trash.getTrash(mockFs, properties, "testUser");
+
+    // Assert if condition is evaluated and the correct trash is returned
+    Assert.assertTrue(trash instanceof MockTrash);
+
+    properties.remove(Trash.TRASH_CLASS_KEY);
+    Trash trash2 = Trash.getTrash(mockFs, properties, "testUser");
+    // Assert if the trash would default to Trash if the condition is not met
+    Assert.assertTrue(trash2 instanceof Trash);
+  }
+
 }