[HKWMETRICS-769] unscheduled delete expired metrics job and remove co…

…nfig (#933)
hawkular · Apr 9, 2018 · 47f1bfc · 47f1bfc
1 parent 8998b1d
commit 47f1bfc
Showing 1 changed file with 17 additions and 1 deletion.
diff --git a/job-scheduler/src/main/java/org/hawkular/metrics/scheduler/api/JobsManager.java b/job-scheduler/src/main/java/org/hawkular/metrics/scheduler/api/JobsManager.java
@@ -75,14 +75,15 @@ public List<JobDetails> installJobs() {
         List<JobDetails> backgroundJobs = new ArrayList<>();
 
         unscheduleCompressData();
+        unscheduleDeleteExpiredMetrics();
         maybeScheduleTableCreator(backgroundJobs);
         maybeScheduleTempDataCompressor(backgroundJobs);
 
         return backgroundJobs;
     }
 
     private void unscheduleCompressData() {
-        Configuration config = configurationService.load(COMPRESS_DATA_JOB).toBlocking()
+        Configuration config = configurationService.load(COMPRESS_DATA_CONFIG_ID).toBlocking()
                 .firstOrDefault(new Configuration(COMPRESS_DATA_CONFIG_ID, new HashMap<>()));
         String jobId = config.get("jobId");
 
@@ -105,6 +106,21 @@ private void unscheduleCompressData() {
         }
     }
 
+    private void unscheduleDeleteExpiredMetrics() {
+        String jobName = "DELETE_EXPIRED_METRICS";
+        String configId = "org.hawkular.metrics.jobs.DELETE_EXPIRED_METRICS";
+
+        // We load the configuration first so that delete is done only if it exists in order to avoid generating
+        // tombstones.
+        Completable deleteConfig = configurationService.load(configId)
+                .map(config -> configurationService.delete(configId))
+                .toCompletable();
+        // unscheduleJobByTypeAndName will not generate unnecessary tombstones as it does reads before writes
+        Completable unscheduleJob = scheduler.unscheduleJobByTypeAndName(jobName, jobName);
+
+        Completable.merge(deleteConfig, unscheduleJob).await();
+    }
+
     private void maybeScheduleTableCreator(List<JobDetails> backgroundJobs) {
         Configuration config = configurationService.load(TEMP_TABLE_CREATE_CONFIG_ID).toBlocking()
                 .firstOrDefault(new Configuration(TEMP_TABLE_CREATE_CONFIG_ID, new HashMap<>()));