diff --git a/.github/workflows/integration-tests.yml b/.github/workflows/integration-tests.yml index 5f5129cfc..d207aadcd 100644 --- a/.github/workflows/integration-tests.yml +++ b/.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 diff --git a/Makefile b/Makefile index 80d76b22d..4d2393f1a 100644 --- a/Makefile +++ b/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 diff --git a/deploy/dev/flink-session-cluster.yaml b/deploy/dev/flink-session-cluster.yaml index 40f46dbf0..3ea6e1b48 100644 --- a/deploy/dev/flink-session-cluster.yaml +++ b/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: diff --git a/deploy/hoptimator-operator-deployment.yaml b/deploy/hoptimator-operator-deployment.yaml index 75e84b049..e25c75783 100644 --- a/deploy/hoptimator-operator-deployment.yaml +++ b/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 diff --git a/deploy/samples/kafka-trigger.yaml b/deploy/samples/kafka-trigger.yaml new file mode 100644 index 000000000..48d9a5ed4 --- /dev/null +++ b/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 diff --git a/deploy/samples/retl-job-template.yaml b/deploy/samples/retl-job-template.yaml index 912ab26b2..8abd6ece3 100644 --- a/deploy/samples/retl-job-template.yaml +++ b/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 \ No newline at end of file diff --git a/docs/extending/index.md b/docs/extending/index.md index ee82fc8d1..fb0955ca8 100644 --- a/docs/extending/index.md +++ b/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 frontier(String table); // latest data-time seen + default List 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 diff --git a/docs/kubernetes/crd-reference.md b/docs/kubernetes/crd-reference.md index 0d7342ba0..1af1458bf 100644 --- a/docs/kubernetes/crd-reference.md +++ b/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 ``: + +| Variable | Example | Meaning | +| -------- | ------- | ------- | +| `{{}}` | `2026-05-08T07:55Z` | ISO-8601 instant (e.g. `{{timestamp}}`). | +| `{{EpochMs}}` | `1778226900000` | Unix epoch milliseconds. | +| `{{Date}}` | `2026-05-08` | UTC calendar date (handy for `dt=`-style partitions). | +| `{{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 +>