[gobblin] branch master updated: [GOBBLIN-1880] handle invalid cron schedules (#3742)

[wl...@apache.org]

This is an automated email from the ASF dual-hosted git repository.

wlo pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/gobblin.git


The following commit(s) were added to refs/heads/master by this push:
     new 722744263 [GOBBLIN-1880] handle invalid cron schedules (#3742)
722744263 is described below

commit 7227442632a32583e75b36b3737a8723efab6486
Author: meethngala <me...@gmail.com>
AuthorDate: Mon Aug 28 18:23:02 2023 -0700

    [GOBBLIN-1880] handle invalid cron schedules (#3742)
    
    * handle invalid cron schedules
    
    * address PR comments
    
    ---------
    
    Co-authored-by: Meeth Gala <mg...@linkedin.com>
---
 .../modules/scheduler/GobblinServiceJobScheduler.java     | 15 ++++++++++++---
 .../modules/scheduler/GobblinServiceJobSchedulerTest.java |  5 +++++
 2 files changed, 17 insertions(+), 3 deletions(-)

diff --git a/gobblin-service/src/main/java/org/apache/gobblin/service/modules/scheduler/GobblinServiceJobScheduler.java b/gobblin-service/src/main/java/org/apache/gobblin/service/modules/scheduler/GobblinServiceJobScheduler.java
index f23bf0bf8..367d3c96f 100644
--- a/gobblin-service/src/main/java/org/apache/gobblin/service/modules/scheduler/GobblinServiceJobScheduler.java
+++ b/gobblin-service/src/main/java/org/apache/gobblin/service/modules/scheduler/GobblinServiceJobScheduler.java
@@ -291,6 +291,10 @@ public class GobblinServiceJobScheduler extends JobScheduler implements SpecCata
    */
   @VisibleForTesting
   public static boolean isWithinRange(String cronExpression, int maxNumDaysToScheduleWithin) {
+    if (cronExpression.trim().isEmpty()) {
+      // If the cron expression is empty return true to capture adhoc flows
+      return true;
+    }
     CronExpression cron = null;
     Calendar cal = Calendar.getInstance(TimeZone.getTimeZone("UTC"));
     double numMillisInADay = 86400000;
@@ -299,17 +303,22 @@ public class GobblinServiceJobScheduler extends JobScheduler implements SpecCata
       cron.setTimeZone(TimeZone.getTimeZone("UTC"));
       Date nextValidTimeAfter = cron.getNextValidTimeAfter(new Date());
       if (nextValidTimeAfter == null) {
-        log.warn("Calculation issue for next valid time for expression: {}. Will default to true for within range",
+        log.warn("Next valid time doesn't exist since it's out of range for expression: {}. ",
             cronExpression);
-        return true;
+        // nextValidTimeAfter will be null in cases only when CronExpression is outdated for a given range
+        // this will cause NullPointerException while scheduling FlowSpecs from FlowCatalog
+        // Hence, returning false to avoid expired flows from being scheduled
+        return false;
       }
       cal.setTime(nextValidTimeAfter);
       long diff = cal.getTimeInMillis() - System.currentTimeMillis();
       return (int) Math.round(diff / numMillisInADay) < maxNumDaysToScheduleWithin;
     } catch (ParseException e) {
       e.printStackTrace();
+      // Return false when a parsing exception occurs due to invalid cron
+      return false;
     }
-    return true;
+
   }
 
   /**
diff --git a/gobblin-service/src/test/java/org/apache/gobblin/service/modules/scheduler/GobblinServiceJobSchedulerTest.java b/gobblin-service/src/test/java/org/apache/gobblin/service/modules/scheduler/GobblinServiceJobSchedulerTest.java
index 6db4a83e3..9949c97d5 100644
--- a/gobblin-service/src/test/java/org/apache/gobblin/service/modules/scheduler/GobblinServiceJobSchedulerTest.java
+++ b/gobblin-service/src/test/java/org/apache/gobblin/service/modules/scheduler/GobblinServiceJobSchedulerTest.java
@@ -95,6 +95,11 @@ public class GobblinServiceJobSchedulerTest {
     Assert.assertTrue(GobblinServiceJobScheduler.isWithinRange("0 * 4 ? * 1,2", thresholdToSkipScheduling));
     // For adhoc flows schedule is empty string
     Assert.assertTrue(GobblinServiceJobScheduler.isWithinRange("", thresholdToSkipScheduling));
+    // Schedule for midnight of the current day which is in the past if threshold is set to zero (today)
+    Assert.assertFalse(GobblinServiceJobScheduler.isWithinRange("0 0 0 * * ?", 0));
+    // Capture invalid schedules in the past
+    Assert.assertFalse(GobblinServiceJobScheduler.isWithinRange("0 0 0 ? * 6L 2022", thresholdToSkipScheduling));
+
   }
 
   /**