Flink Job Terminal State Health Gate

A Flink job that fails mid-stream does not page anyone on its own; the JobManager UI shows the state if someone opens it, and nothing in Flink itself compares that state against what you expected or notifies a channel. This blueprint polls the job's own REST status and treats reaching a terminal state outside your expected list as a governance event worth paging on, not a dashboard someone has to remember to check.

This blueprint was created by Mohit Jeswani.

How it works

1. job_terminal_watch (io.kestra.plugin.flink.MonitorJob) polls the JobManager REST API every 30 seconds and fires exactly one execution the moment the job reaches a terminal state (FINISHED, CANCELED, or FAILED). failOnError: false keeps the trigger firing an execution on FAILED instead of failing the trigger itself before evaluate can run.
2. evaluate (io.kestra.plugin.core.output.OutputValues) classifies the trigger's own success flag and finalState: a state in expectedTerminalStates is OK, FAILED is JOB_FAILED, anything else unexpected (for example an operator-initiated CANCELED that was not expected) is UNEXPECTED_TERMINATION.
3. route (io.kestra.plugin.core.flow.If):
   • compliant: logs the clean terminal state.
   • alert: posts the job id, terminal state, and stateDetails to Slack.
4. summary logs the final state and severity, closing the governance cycle.
5. errors alerts Slack if the orchestration flow itself fails, separate from a Flink job failure, which is handled by alert above.

What you get

• Detection built on the job's own REST-reported state, not an inferred heartbeat or log scrape, so a failure is caught the moment the JobManager itself records it.
• A failed job and an unexpected-but-not-failed termination (for example a manual cancel nobody flagged as expected) both surface, with distinct severities.
• A read-only check: this flow only polls status, it never cancels, resubmits, or otherwise touches the job.

Who it's for

• Streaming platform and data engineering teams running Flink jobs who want a failure paged the moment the JobManager reports it, not discovered when a downstream consumer notices stale data.
• Teams evaluating the Flink plugin who want a worked example of MonitorJob used as a standing governance trigger, distinct from this repo's existing Flink blueprints, which use Submit, SubmitSql, and TriggerSavepoint rather than terminal-state monitoring.

Why orchestrate this with Kestra

Flink's REST API reports job state on request; it does not decide which terminal states are expected, compare the actual state against that list, or notify a Slack channel when they diverge. Kestra keeps expectedTerminalStates as trigger configuration, turns a state mismatch into a Slack page naming the exact state and details, and keeps a full execution history of every terminal event.

Prerequisites

• A running Flink cluster with its JobManager REST API reachable from the Kestra worker.
• The jobId of a job already submitted to that cluster (for example via this repo's flink-stateful-job-savepoint-restart-drill blueprint, or flink run).
• A Slack incoming webhook URL.

This flow polls the Flink REST API directly; a local Flink cluster can be run via the official flink:1.20 Docker image for an end-to-end local test, no managed Flink service required.

Secrets

• SLACK_WEBHOOK_URL: used by alert and the errors block. In Kestra OSS, provide it as the base64-encoded environment variable SECRET_SLACK_WEBHOOK_URL.

Inputs

This blueprint has no flow inputs; restUrl, jobId, and expectedTerminalStates are trigger configuration, set directly on job_terminal_watch since a polling trigger has no execution inputs to render from.

Outputs

• {{ trigger.jobId }} / .finalState / .stateDetails / .success: the terminal state MonitorJob observed, available to every downstream task as trigger variables.
• {{ outputs.evaluate.values.severity }}: OK, JOB_FAILED, or UNEXPECTED_TERMINATION.

Pitfalls

• restUrl and jobId are trigger-level configuration, not flow inputs; editing them means editing the trigger block itself, not passing a different value at execution time.
• expectedTerminalStates defaults to ['FINISHED'] only; a job that is expected to be manually CANCELED as part of normal operation needs CANCELED added to that list or every such cancellation reports UNEXPECTED_TERMINATION.
• This trigger fires once, when the job reaches a terminal state; it does not re-arm for the same job afterward, since a terminal Flink job does not restart on its own. Submit a new job (and a new execution of a monitoring flow pointed at its new jobId) for the next run.

Quick start

1. Add the SLACK_WEBHOOK_URL secret.
2. Run a Flink job (locally via the flink:1.20 Docker image, or against a real cluster) and note its jobId from the JobManager UI or REST API.
3. Set restUrl and jobId on job_terminal_watch to that cluster and job.
4. Enable the trigger and let the job finish; confirm compliant or alert fires matching the job's actual outcome.

How to extend

• Add CANCELED to expectedTerminalStates for jobs that are routinely stopped manually as part of a deploy, so a planned cancellation does not page as UNEXPECTED_TERMINATION.
• Chain io.kestra.plugin.flink.Submit in the alert branch to automatically resubmit the job from its last savepoint, gated behind an auto_restart input so the first version of this flow stays alert-only.
• Route JOB_FAILED to an on-call paging channel and UNEXPECTED_TERMINATION to a lower- urgency channel, since an unexpected cancel and an actual failure usually have different owners.

Links

• Flink plugin: https://kestra.io/plugins/plugin-flink
• Flink REST API reference: https://nightlies.apache.org/flink/flink-docs-stable/docs/ops/rest_api/
• Slack incoming webhooks: https://api.slack.com/messaging/webhooks