Keeping Trigger Timestamp in Job Annotation Updated (#172)

Co-authored-by: Shrinand Thakkar <sthakkar@linkedin.com>

Author: srnand

hoptimator-operator/src/main/java/com/linkedin/hoptimator/operator/trigger/TableTriggerReconciler.java

@@ -7,6 +7,7 @@
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
+import java.util.Objects;
 import java.util.Optional;
 
 import org.slf4j.Logger;
@@ -60,9 +61,9 @@
  *
  */
 public final class TableTriggerReconciler implements Reconciler {
-  private final static Logger log = LoggerFactory.getLogger(TableTriggerReconciler.class);
-  private final static String TRIGGER_KEY = "trigger";
-  private final static String TRIGGER_TIMESTAMP_KEY = "triggerTimestamp";
+  private static final Logger log = LoggerFactory.getLogger(TableTriggerReconciler.class);
+  static final String TRIGGER_KEY = "trigger";
+  static final String TRIGGER_TIMESTAMP_KEY = "triggerTimestamp";
 
   private final K8sApi<V1alpha1TableTrigger, V1alpha1TableTriggerList> tableTriggerApi;
   private final K8sApi<V1Job, V1JobList> jobApi;
@@ -146,6 +147,7 @@ public Result reconcile(Request request) {
           jobApi.delete(job);
           return new Result(true);  // retry
         } else {
+          maybeUpdateJobAnnotation(job, status.getTimestamp());
           log.info("Job for TableTrigger {} still running from a previous trigger event.", name);
           return new Result(true, pendingRetryDuration());  // retry later
         }
@@ -195,5 +197,18 @@ public static Controller controller(K8sContext context) {
         .watch(x -> ControllerBuilder.controllerWatchBuilder(V1alpha1TableTrigger.class, x).build())
         .build();
   }
+
+  void maybeUpdateJobAnnotation(V1Job job, OffsetDateTime timestamp) throws SQLException {
+    Map<String, String> annotations = Objects.requireNonNull(job.getMetadata()).getAnnotations();
+    if (annotations != null) {
+      String existing = annotations.get(TRIGGER_TIMESTAMP_KEY);
+      if (existing != null && timestamp.isAfter(OffsetDateTime.parse(existing))) {
+        annotations.put(TRIGGER_TIMESTAMP_KEY, timestamp.toString());
+        job.getMetadata().setAnnotations(annotations);
+        jobApi.update(job);
+        log.info("Updated {} in Job {} annotation to {}", TRIGGER_TIMESTAMP_KEY, job.getMetadata().getName(), timestamp);
+      }
+    }
+  }
 }
 

hoptimator-operator/src/test/java/com/linkedin/hoptimator/operator/trigger/TestTableTriggerReconciler.java

@@ -6,6 +6,7 @@
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
+import java.util.Objects;
 
 import org.junit.jupiter.api.Assertions;
 import org.junit.jupiter.api.BeforeEach;
@@ -97,4 +98,28 @@ void updatesWatermark() {
     Assertions.assertTrue(result.isRequeue());
     Assertions.assertNotNull(trigger.getStatus().getWatermark(), "Watermark was not set");
   }
+
+  @Test
+  void reconcileTriggersAnnotationUpdate() {
+    OffsetDateTime oldTimestamp = OffsetDateTime.parse("2024-01-01T12:00:00Z");
+    OffsetDateTime newTimestamp = OffsetDateTime.parse("2024-01-03T12:00:00Z");
+    Map<String, String> annotations = new HashMap<>();
+    annotations.put(TableTriggerReconciler.TRIGGER_KEY, "table-trigger");
+    annotations.put(TableTriggerReconciler.TRIGGER_TIMESTAMP_KEY, oldTimestamp.toString());
+    V1Job job = new V1Job()
+        .metadata(new V1ObjectMeta().name("running-table-trigger-job").annotations(annotations))
+        .status(new V1JobStatus().addConditionsItem(new V1JobCondition().type("Running").status("True")));
+    V1alpha1TableTrigger trigger = new V1alpha1TableTrigger()
+      .metadata(new V1ObjectMeta().name("table-trigger"))
+      .spec(new V1alpha1TableTriggerSpec().yaml(Yaml.dump(job)))
+      .status(new V1alpha1TableTriggerStatus().timestamp(newTimestamp));
+    triggers.add(trigger);
+    jobs.add(job);
+    reconciler.reconcile(new Request("namespace", "table-trigger"));
+    Assertions.assertEquals(
+      newTimestamp.toString(),
+      Objects.requireNonNull(Objects.requireNonNull(job.getMetadata()).getAnnotations()).get(TableTriggerReconciler.TRIGGER_TIMESTAMP_KEY),
+      "The annotation should be updated through reconcile when job is still running and if the trigger timestamp has advanced."
+    );
+  }
 }