[GitHub] [incubator-pinot] jihaozh commented on a change in pull request #5056: [TE] [data-availability] Scheduling of data availability tasks & some clean up

[GitBox <gi...@apache.org>]

jihaozh commented on a change in pull request #5056: [TE] [data-availability] Scheduling of data availability tasks & some clean up
URL: https://github.com/apache/incubator-pinot/pull/5056#discussion_r376630723
 
 

 ##########
 File path: thirdeye/thirdeye-pinot/src/main/java/org/apache/pinot/thirdeye/scheduler/DetectionScheduler.java
 ##########
 @@ -120,34 +126,41 @@ public void run() {
           LOG.error("Error removing job key {}", jobKey);
         }
       }
-
     } catch (SchedulerException e) {
       LOG.error("Error while scheduling detection pipeline", e);
     }
   }
 
-  private boolean jobUpdated(DetectionConfigDTO config, JobKey key) throws SchedulerException {
-    List<Trigger> triggers = (List<Trigger>) scheduler.getTriggersOfJob(key);
-    CronTrigger cronTrigger = (CronTrigger) triggers.get(0);
-    String cronInSchedule = cronTrigger.getCronExpression();
+  @Override
+  public Set<JobKey> getScheduledJobs() throws SchedulerException {
+    return this.scheduler.getJobKeys(GroupMatcher.jobGroupEquals(TaskConstants.TaskType.DETECTION.toString()));
+  }
 
-    if (!config.getCron().equals(cronInSchedule)) {
-      LOG.info("Cron expression for detection pipeline {} has been changed from {}  to {}. " + "Restarting schedule",
-          config.getId(), cronInSchedule, config.getCron());
-      return true;
-    }
-    return false;
+  @Override
+  public void shutdown() throws SchedulerException {
+    AnomalyUtils.safelyShutdownExecutionService(executorService, this.getClass());
+    scheduler.shutdown();
   }
 
-  private void startJob(DetectionConfigDTO config, JobKey key) throws SchedulerException {
-    Trigger trigger =
-        TriggerBuilder.newTrigger().withSchedule(CronScheduleBuilder.cronSchedule(config.getCron())).build();
-    JobDetail job = JobBuilder.newJob(DetectionPipelineJob.class).withIdentity(key).build();
-    this.scheduler.scheduleJob(job, trigger);
-    LOG.info(String.format("scheduled detection pipeline job %s.", key.getName()));
+  @Override
+  public void startJob(AbstractBean config, JobKey key) throws SchedulerException {
+    Trigger trigger = TriggerBuilder.newTrigger().withSchedule(
+        CronScheduleBuilder.cronSchedule(((DetectionConfigBean) config).getCron())).build();
+    JobDetail detectionJob = JobBuilder.newJob(DetectionPipelineJob.class).withIdentity(key).build();
+
+    this.scheduler.scheduleJob(detectionJob, trigger);
+    LOG.info(String.format("scheduled detection pipeline job %s", detectionJob.getKey().getName()));
+
+    // Data availability alerts will be scheduled only when enabled by the user.
+    if (DetectionUtils.isDataAvailabilityCheckEnabled((DetectionConfigDTO) config)) {
+      JobDetail dataAvailabilityJob = JobBuilder.newJob(DetectionDataAvailabilityJob.class).withIdentity(key).build();
+      this.scheduler.scheduleJob(dataAvailabilityJob, trigger);
 
 Review comment:
   Does the availability task and the detection task share the same cron?

----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
 
For queries about this service, please contact Infrastructure at:
users@infra.apache.org


With regards,
Apache Git Services

---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org