Flink Job Restart Loop Sentinel

A Flink job that keeps failing and auto-restarting does not go quiet - it keeps consuming TaskManager slots and CPU on every retry, often for hours, while producing nothing useful. Flink's own fault-tolerance is deliberately persistent: that is exactly what makes a genuinely broken job look, from the outside, identical to a job that is about to recover on its own. This blueprint polls a job's state through Flink's own JobManager REST API, flags one stuck FAILED, FAILING, or RESTARTING, alerts Slack, and can stop it cleanly with a savepoint-and-cancel - preserving its state for a later resume instead of leaving it to retry indefinitely.

How it works

1. flink_job_health_schedule (io.kestra.plugin.core.trigger.Schedule) polls every 15 minutes, always forcing auto_remediate: "false" and dry_run: "true" through its own inputs: override regardless of this flow's defaults. Shipped disabled so you can validate a manual run first.
2. probe_job_status (io.kestra.plugin.core.http.Request) calls GET /jobs/:jobid on Flink's own JobManager REST API with options.allowFailed: true, so a missing job or unreachable cluster does not fail the task outright.
3. classify_job_health (io.kestra.plugin.core.debug.Return) classifies the result as UNREACHABLE (no HTTP 200), AT_RISK (state is FAILED, FAILING, or RESTARTING), or HEALTHY.
4. route_by_health (io.kestra.plugin.core.flow.Switch) branches on that classification:
   • HEALTHY: log_healthy records a clean check.
   • AT_RISK: a nested route_remediation (Switch on auto_remediate) either considers remediation ("true", gated again by dry_run through check_dry_run) or only logs that remediation is disabled ("false"). When both gates allow it, trigger_savepoint_and_cancel (io.kestra.plugin.flink.TriggerSavepoint, cancelJob: true) stops the job with a savepoint. alert_at_risk always fires to Slack regardless of which path ran.
   • UNREACHABLE: alert_unreachable pages Slack directly - a broken audit must never read as "job is fine".
5. log_audit_status always runs last, printing the classification regardless of which branch fired.
6. The errors block alerts Slack separately if the flow itself fails outside these handled branches.

What you get

• A direct, Flink-native answer to "is this job actually stuck right now", not an inference from dashboard CPU or checkpoint-lag graphs.
• Remediation that only ever stops the job cleanly with a preserved savepoint; nothing about the job's code, configuration, or data is touched.
• Two independent gates (auto_remediate, dry_run) between "stuck job detected" and "job actually stopped".
• A three-way classification (HEALTHY/AT_RISK/UNREACHABLE) and a final status log every run, so the execution history doubles as a job-health trend.

Who it's for

• Platform and data engineering teams running self-hosted Flink who need an early signal when a streaming job enters a restart loop, instead of discovering it from a resource-exhaustion alert hours later.
• SREs who want a pre-production gate confirming a newly submitted job stabilizes before declaring the deployment successful.
• Anyone evaluating the Flink plugin who wants a worked example of TriggerSavepoint used as a genuine operational safeguard rather than a routine checkpoint.

Why orchestrate this with Kestra

Flink's own REST API answers one request at a time and takes no action on its own; stopping a stuck job by hand means someone noticing, logging in, and running the stop command themselves. Kestra supplies the schedule, the three-way risk classification as a first-class branch, two independent authorization gates before anything is stopped, and an execution history that shows exactly when a job first got stuck and whether it was cleanly halted.

Prerequisites

• A reachable Flink JobManager REST endpoint and the job ID to audit.
• A savepoint target directory reachable by every TaskManager in the cluster, not just the JobManager.
• A Slack incoming webhook for alerts.

Local testing: docker run -d --name flink-restart-sentinel -p 8082:8081 apache/flink:latest jobmanager docker network connect YOUR_KESTRA_NETWORK flink-restart-sentinel (only needed if the Kestra Worker runs in a separate Docker network than this container; set flink_rest_url to http://flink-restart-sentinel:8081 from inside that network, or http://localhost:8082 from the host). A JobManager alone has no TaskManager to run jobs on; submit a TaskManager container and a sample job per Flink's own standalone-cluster quickstart before exercising AT_RISK end to end.

Secrets

• SLACK_WEBHOOK_URL: Slack incoming webhook used by alert_at_risk, alert_unreachable, and the errors block.
• This blueprint's flink_rest_url carries no embedded credentials, matching Flink's own unauthenticated-by-default REST API. For a cluster behind an authentication proxy, add the required headers to probe_job_status and trigger_savepoint_and_cancel from {{ secret(...) }} values rather than hardcoding them.
• In Kestra OSS (no Enterprise secrets backend), secrets are supplied as environment variables prefixed SECRET_, base64-encoded, and read back in flows with {{ secret('NAME') }} - for example SECRET_SLACK_WEBHOOK_URL=$(echo -n 'https://hooks.slack.com/...' | base64). This keeps credentials out of the flow YAML but, per Kestra's own documentation, offers no encryption at rest or access control beyond the host environment; use the Enterprise secrets backend for stronger guarantees.

Inputs

• flink_rest_url (STRING, default http://flink-jobmanager:8081): JobManager REST base URL.
• job_id (STRING, no default): the job audited.
• savepoint_target_directory (STRING, default file:///tmp/flink-savepoints): where TriggerSavepoint writes the savepoint when remediation runs.
• auto_remediate (SELECT: "false", "true"; default "false"): must be "true" for the savepoint-and-cancel to even be considered.
• dry_run (SELECT: "true", "false"; default "true"): must be explicitly "false", together with auto_remediate: "true", for the job to actually be stopped.

Outputs

• outputs.probe_job_status.code / .body: the raw HTTP status and JSON body for every run.
• outputs.classify_job_health.value: HEALTHY, AT_RISK, or UNREACHABLE for every run.

Quick start

1. Start a Flink standalone cluster locally (JobManager plus at least one TaskManager) and submit a sample job, or point flink_rest_url/job_id at an existing cluster and job.
2. Add the SLACK_WEBHOOK_URL secret.
3. Run the flow manually against a healthy, RUNNING job and confirm log_healthy fires.
4. To exercise AT_RISK, submit a job that intentionally throws on startup (or stop its sole TaskManager while it runs) so it enters FAILED/RESTARTING, then re-run and confirm alert_at_risk fires.
5. Run once with auto_remediate: "true" and dry_run: "true" against that stuck job and confirm log_dry_run_remediation describes the savepoint-and-cancel without running it.
6. Run once with auto_remediate: "true" and dry_run: "false" and confirm the job transitions to CANCELED and a savepoint appears under savepoint_target_directory.
7. Enable flink_job_health_schedule once you trust the check; it always runs in the safe auto_remediate: "false" / dry_run: "true" mode regardless of what you leave the flow's own defaults set to.

How to extend

• Loop over several job_id values with io.kestra.plugin.core.flow.Loop (ForEach is deprecated) to sweep every running job on a cluster in one run, after first listing jobs via GET /jobs the same way probe_job_status queries a single one.
• Replace the fixed 15-minute Schedule with the plugin's own io.kestra.plugin.flink.MonitorJob trigger, which fires directly on Flink job state changes instead of polling.
• Add a check on restart-strategy-reported restart count (also available from GET /jobs/:jobid) to distinguish a job that has restarted once from one restarting in a tight loop, before escalating to remediation.
• Route the remediation branch through a Pause approval task instead of (or alongside) the static auto_remediate input, for a human-reviewed stop on a production job.

Pitfalls

• This plugin has no structured "get job status" task. Verified task list: Submit, SubmitSql, CancelJob, TriggerSavepoint, plus the MonitorJob trigger - none of them read and return a job's current state as a plain value. This blueprint's probe calls Flink's own GET /jobs/:jobid REST endpoint directly through the core io.kestra.plugin.core.http.Request task instead.
• options.allowFailed: true, not allowFailed: true at the task level. Verified from two existing in-repo flows (aikido-reachability-gated-triage.yaml, api-health-monitor.yaml); the flat task-level property does not exist on io.kestra.plugin.core.http.Request.
• A savepoint directory must be reachable by every TaskManager, not just the JobManager issuing the REST call. Verified from Flink's own operational documentation on savepoints; a file:// path that only exists on the JobManager's local disk will fail for jobs with tasks running on other nodes in a real multi-node cluster.
• cancelJob: true stops the job - it does not restart it. TriggerSavepoint with this flag is Flink's own documented pattern for a clean, stateful stop, not an in-place recovery; resuming the job from the written savepoint is a separate, deliberate Submit with restoreFromSavepoint set, not something this blueprint does automatically.
• auto_remediate and dry_run are independent gates, not a single boolean. Both must be "true"/"false" respectively for TriggerSavepoint to run; setting only one leaves the other still blocking.
• Switch case keys are quoted strings matching the classifier's exact output. "HEALTHY", "AT_RISK", and "UNREACHABLE" must match classify_job_health.value verbatim; the defaults: block exists specifically to catch any value that does not match one of those three cases.

Links

• plugin-flink documentation
• Apache Flink REST API reference
• Apache Flink savepoints reference
• Kestra secrets (OSS vs Enterprise)