Add data-availability triggering for TableTriggers

.github/workflows/integration-tests.yml

@@ -35,18 +35,12 @@ jobs:
         min-coverage-changed-files: 60
         pass-emoji: ':green_circle:'
         update-comment: true
-    - name: Create K8s Kind Cluster
+    - name: Install kind
       uses: helm/kind-action@v1.4.0
       with:
-        cluster_name: 'kind'
-        config: './etc/cluster.yaml'
-    - name: Load Docker Images
-      run: |
-        kind load docker-image hoptimator
-        kind load docker-image hoptimator-flink-runner
-        kind load docker-image hoptimator-flink-operator
+        install_only: true
     - name: Run Integration Tests
-      run: make integration-tests-kind
+      run: make integration-tests
     - name: Capture Integration Reports
       if: failure()
       uses: actions/upload-artifact@v6

Makefile

@@ -20,6 +20,11 @@ build:
 	docker build hoptimator-flink-runner -f hoptimator-flink-runner/Dockerfile-flink-runner -t hoptimator-flink-runner
 	docker build hoptimator-flink-runner -f hoptimator-flink-runner/Dockerfile-flink-operator -t hoptimator-flink-operator
 
+load-kind-cluster:
+	kind load docker-image hoptimator:latest --name hoptimator-kind
+	kind load docker-image hoptimator-flink-operator:latest --name hoptimator-kind
+	kind load docker-image hoptimator-flink-runner:latest --name hoptimator-kind
+
 bounce: build undeploy deploy
 
 clean:
@@ -64,7 +69,7 @@ deploy-flink: deploy
 	kubectl create namespace flink || echo "skipping"
 	kubectl create -f https://github.com/jetstack/cert-manager/releases/download/v1.8.2/cert-manager.yaml || echo "skipping"
 	helm repo add flink-operator-repo https://downloads.apache.org/flink/flink-kubernetes-operator-1.13.0/
-	helm upgrade --install --atomic --set webhook.create=false,image.pullPolicy=Never,image.repository=docker.io/library/hoptimator-flink-operator,image.tag=latest --set-json='watchNamespaces=["default","flink"]' flink-kubernetes-operator flink-operator-repo/flink-kubernetes-operator
+	helm upgrade --install --atomic --set webhook.create=false,image.pullPolicy=Never,image.repository=hoptimator-flink-operator,image.tag=latest --set-json='watchNamespaces=["default","flink"]' flink-kubernetes-operator flink-operator-repo/flink-kubernetes-operator
 	kubectl apply -f ./deploy/dev/flink-session-cluster.yaml
 	kubectl apply -f ./deploy/dev/flink-sql-gateway.yaml
 	kubectl apply -f ./deploy/samples/flink-template.yaml
@@ -124,24 +129,21 @@ deploy-logical: deploy deploy-flink deploy-kafka deploy-venice
 undeploy-logical:
 	kubectl delete -f ./deploy/samples/logicaldb.yaml || echo "skipping"
 
-deploy-dev-environment: deploy deploy-demo deploy-flink deploy-kafka deploy-mysql deploy-venice deploy-logical
+create-kind-cluster:
+	kind create cluster -n hoptimator-kind --config ./etc/cluster.yaml || echo "skipping"
+
+deploy-dev-environment: create-kind-cluster load-kind-cluster deploy deploy-demo deploy-flink deploy-kafka deploy-mysql deploy-venice deploy-logical
 
 undeploy-dev-environment: undeploy-logical undeploy-venice undeploy-mysql undeploy-kafka undeploy-flink undeploy-demo undeploy
 	kubectl delete -f ./deploy/dev || echo "skipping"
+	kind delete cluster -n hoptimator-kind || echo "skipping"
 
-# Integration test setup intended to be run locally
+# End-to-end integration tests, used both locally and in CI (the GitHub workflow installs kind, then
+# runs this target). create-kind-cluster provisions the kind cluster with ./etc/cluster.yaml, which
+# maps the Kafka bootstrap/broker and Flink SQL gateway node ports to localhost — so the tests reach
+# the cluster from the host with no manual port-forwarding.
 integration-tests: deploy-dev-environment
-	kubectl port-forward -n kafka svc/one-kafka-external-bootstrap 9092 & echo $$! > port-forward.pid
-	kubectl port-forward -n flink svc/flink-sql-gateway 8083 & echo $$! > port-forward-2.pid
-	kubectl port-forward -n flink svc/basic-session-deployment-rest 8081 & echo $$! > port-forward-3.pid
-	./gradlew intTest --no-parallel || kill `cat port-forward.pid port-forward-2.pid port-forward-3.pid`
-	kill `cat port-forward.pid`
-	kill `cat port-forward-2.pid`
-	kill `cat port-forward-3.pid`
-
-# kind cluster used in github workflow needs to have different routing set up, avoiding the need to forward kafka ports
-integration-tests-kind: deploy-dev-environment
-	./gradlew intTest -i --no-parallel
+	./gradlew intTest --no-parallel
 
 generate-models:
 	./generate-models.sh
@@ -164,4 +166,4 @@ run-zeppelin: build-zeppelin
 	  --name hoptimator-zeppelin \
 	  hoptimator-zeppelin
 
-.PHONY: install test coverage build bounce clean quickstart deploy-config undeploy-config deploy undeploy deploy-demo undeploy-demo deploy-flink undeploy-flink deploy-kafka undeploy-kafka deploy-mysql undeploy-mysql deploy-venice undeploy-venice deploy-logical undeploy-logical build-zeppelin run-zeppelin integration-tests integration-tests-kind deploy-dev-environment undeploy-dev-environment generate-models release
+.PHONY: install test coverage build bounce clean quickstart deploy-config undeploy-config deploy undeploy deploy-demo undeploy-demo deploy-flink undeploy-flink deploy-kafka undeploy-kafka deploy-mysql undeploy-mysql deploy-venice undeploy-venice deploy-logical undeploy-logical build-zeppelin run-zeppelin integration-tests deploy-dev-environment undeploy-dev-environment generate-models release

deploy/dev/flink-session-cluster.yaml

@@ -4,7 +4,7 @@ metadata:
   name: basic-session-deployment
   namespace: flink
 spec:
-  image: docker.io/library/hoptimator-flink-runner
+  image: hoptimator-flink-runner:latest
   imagePullPolicy: Never
   flinkVersion: v1_18
   flinkConfiguration:

deploy/hoptimator-operator-deployment.yaml

@@ -17,8 +17,8 @@ spec:
       serviceAccountName: hoptimator-operator
       containers:
       - name: hoptimator-operator
-        image: docker.io/library/hoptimator
-        imagePullPolicy: Never
+        image: hoptimator:latest
+        imagePullPolicy: IfNotPresent
         command: ["./hoptimator-operator-integration/bin/hoptimator-operator-integration", "jdbc:calcite:model=/etc/config/model.yaml"]
         volumeMounts:
         - name: config-volume

deploy/samples/kafka-trigger.yaml

@@ -0,0 +1,36 @@
+# Sample: a TableTrigger that fires automatically when new records arrive on a Kafka topic.
+#
+# The KafkaInputWatermark provider (hoptimator-kafka-controller) reports the topic's latest record
+# timestamp as the input's data-time frontier. The generic TableTriggerReconciler advances this
+# trigger's cursor to that frontier and launches the Job below over the newly-available window
+# [watermark, timestamp]. No external scheduler/poker is needed — contrast
+# deploy/samples/crontrigger.yaml, which pokes a trigger's status on a cron.
+#
+# Requires the operator to know the Kafka cluster: set the bootstrap servers via the
+# `hoptimator.kafka.bootstrap.servers` system property or the `KAFKA_BOOTSTRAP_SERVERS` env var.
+apiVersion: hoptimator.linkedin.com/v1alpha1
+kind: TableTrigger
+metadata:
+  name: kafka-arrival-trigger
+spec:
+  schema: KAFKA
+  table: existing-topic-1
+  yaml: |
+    apiVersion: batch/v1
+    kind: Job
+    metadata:
+      name: kafka-arrival-trigger-job
+    spec:
+      template:
+        spec:
+          containers:
+          - name: hello
+            image: alpine/k8s:1.33.0
+            command:
+            - /bin/bash
+            - -c
+            - >-
+              echo "New records on {{table}}: processing window [{{watermark}}, {{timestamp}}]"
+          restartPolicy: Never
+      backoffLimit: 4
+      ttlSecondsAfterFinished: 90

deploy/samples/retl-job-template.yaml

@@ -21,8 +21,20 @@ spec:
                 value: "{{trigger}}"
               - name: SOURCE_TABLE_NAME
                 value: "{{path}}"
+              # Optional: only set when the offline tier feeds an online tier (reverse-ETL).
+              # Defaults to empty so an offline-only LogicalTable's implicit trigger still renders.
               - name: ONLINE_TABLE_NAME
-                value: "{{job.properties.online.table.name}}"
-            command: ["bash",  "-c", "echo {{name}}-trigger fired at `date`"]
+                value: "{{job.properties.online.table.name:}}"
+              # Fire window [WINDOW_START, WINDOW_END]. On an incremental fire this is
+              # [watermark, timestamp]; on a backfill (FIRE ... FROM ... TO ...) it is [FROM, TO].
+              - name: WINDOW_START
+                value: "{{watermark}}"
+              - name: WINDOW_END
+                value: "{{timestamp}}"
+              - name: WINDOW_START_DATE
+                value: "{{watermarkDate}}"
+              - name: WINDOW_END_DATE
+                value: "{{timestampDate}}"
+            command: ["bash",  "-c", "echo {{name}}-trigger processing window [$WINDOW_START, $WINDOW_END] at `date`"]
           restartPolicy: Never
       backoffLimit: 4

docs/extending/index.md

@@ -15,6 +15,7 @@ both — pick the layer that matches what you're doing.
 | Build a dependency graph from some backing store (e.g. K8s).                                        | A `GraphProvider`. The K8s-backed default ships in `hoptimator-k8s`. |
 | Render the dependency graph in a format other than the ones shipped (DOT, an interactive web view, …). | A `GraphRenderer`. Mermaid and JSON renderers ship in `hoptimator-graph`. |
 | Customize what gets deployed for an existing system.                                                | Just a `TableTemplate` or `JobTemplate` — no Java needed. See [Templates and configuration](../kubernetes/templates.md). |
+| Fire a `TableTrigger` when your source has new data (event-time).                                   | An `InputFrontierSource` on your driver's schema. See [Firing triggers on data availability](#firing-triggers-on-data-availability). |
 
 ## How extensions are loaded
 
@@ -86,6 +87,43 @@ pipeline state lives somewhere other than Kubernetes — the K8s
 implementation is the reference. Both register via `META-INF/services`
 like every other SPI here.
 
+### "I want a `TableTrigger` to fire when my source has new data"
+
+#### Firing triggers on data availability
+
+This one is **not** a `ServiceLoader` SPI. A `TableTrigger` fires when its input advances to a new
+data-time *frontier*; the source reports that frontier by having the Calcite `Schema` its JDBC
+driver already builds implement `com.linkedin.hoptimator.InputFrontierSource`:
+
+```java
+public interface InputFrontierSource {
+  Optional<Instant> frontier(String table);                        // latest data-time seen
+  default List<DataChange> changesSince(String table, Instant since) { … } // late/out-of-order repair
+}
+```
+
+The `frontier` is an **optimistic** signal — "data has appeared through here," not a guarantee that
+everything at or before it has arrived. **Repair is what licenses that optimism:** late or
+out-of-order writes that land behind the cursor are reported by `changesSince` and healed with
+one-off backfills, so completeness is achieved by *frontier + repair*. A source that does **not**
+implement `changesSince` has no safety net and **must** report a *conservative* frontier (a real
+watermark) or it will silently drop late data.
+
+Because the capability hangs off the schema — the object the driver constructs from *this*
+`Database`'s connection config — per-cluster configuration (which brokers to read, etc.) is
+inherent: there is no global config to reach for, and many clusters can each be their own
+`Database`. The source-agnostic `TableTriggerReconciler` resolves a trigger's `(catalog, schema)` to
+the `Database`'s schema via `HoptimatorJdbcSchema.inputFrontierSource()` (which walks the driver's
+inner schema and `unwrap`s this interface — exactly like the `LogicalSchemaMarker` one-bit marker,
+just with methods), then asks it about the specific `table`. A schema that doesn't implement the
+interface is simply not frontier-driven, and the trigger falls back to cron/manual `FIRE`.
+
+To participate, have your driver's inner schema `implements InputFrontierSource`. The Kafka
+`ClusterSchema` in `hoptimator-kafka` is the reference: it takes the conservative route — a bounded
+out-of-orderness watermark (per-partition min of the latest timestamp, minus a `frontier.lag.ms`
+lag, excluding partitions more than the lag behind the leader), so it's sound without any
+`changesSince` repair. No `META-INF/services` file is required.
+
 ## Register, then test
 
 After dropping a service file and a class, the standard verification path

docs/kubernetes/crd-reference.md

@@ -274,6 +274,48 @@ spec:
 | `schedule`      | string  |          | Cron schedule. If set, the trigger fires on a schedule. If null, it fires on status patches. |
 | `paused`        | boolean |          | When true, the trigger does not fire (status updates are ignored).                         |
 
+### Input frontiers
+
+A trigger fires as its input advances to a new data-time **frontier**. When the input's `Database`
+schema exposes an `InputFrontierSource` (see [extending](../extending/index.md)), the operator
+advances the trigger's cursor to that source's reported frontier and launches the job over the
+newly-available window. The frontier is optimistic ("data has appeared through here"); sources that
+can report late changes have writes behind the cursor healed separately via one-off backfills, while
+sources that cannot must report a conservative frontier instead. Sources without any frontier
+capability fire on their cron `schedule` or on manual `FIRE`. Either way, the job is handed its
+output window and reads it directly:
+
+| Template variable | Meaning                                                         |
+| ----------------- | -------------------------------------------------------------- |
+| `{{watermark}}`   | Start of the window (the previous cursor position).            |
+| `{{timestamp}}`   | End of the window (the confirmed frontier / cron tick).        |
+
+The very first fire has no prior cursor, so `{{watermark}}` (and its derived forms) render as an
+empty string rather than skipping the job — the job decides how to treat that open lower bound.
+
+A job that needs a wider *trailing* read (e.g. to absorb late data) expresses that in its own SQL —
+that late-data tolerance is a per-job concern, handled by the job rather than by any platform-level
+schedule adjustment.
+
+Each window endpoint — `watermark` and `timestamp` — is also exported in a few convenience formats so
+a job can consume a value without parsing the ISO string. For base name `<v>`:
+
+| Variable | Example | Meaning |
+| -------- | ------- | ------- |
+| `{{<v>}}` | `2026-05-08T07:55Z` | ISO-8601 instant (e.g. `{{timestamp}}`). |
+| `{{<v>EpochMs}}` | `1778226900000` | Unix epoch milliseconds. |
+| `{{<v>Date}}` | `2026-05-08` | UTC calendar date (handy for `dt=`-style partitions). |
+| `{{<v>Hour}}` | `07` | UTC hour-of-day, zero-padded. |
+
+So `{{timestampDate}}`, `{{timestampEpochMs}}`, `{{watermarkHour}}`, etc. are all available.
+
+> **Window vars in a `JobTemplate`.** A `JobTemplate` referenced by `CREATE TRIGGER ... as
+> <template>` is rendered twice: once at `CREATE` (to fix the static vars — `{{name}}`, `{{table}}`,
+> `{{path}}`, `job.properties.*`, …) and again per-fire (to fill the window). The window vars above
+> are automatically deferred through the first render, so they reach the launched job intact. Any
+> *other* unresolved `{{variable}}` still fails the `CREATE` loudly — a `JobTemplate` that renders to
+> nothing is rejected rather than stored empty.
+
 ### Status fields
 
 | Field       | Type      | Description                                                                  |