Drill a Flink streaming job restart without losing state with Kestra

id: flink-stateful-job-savepoint-restart-drill
namespace: company.team

description: |
  Find the running Flink streaming job, take a savepoint of its state and in remediate mode stop it
  with a savepoint then relaunch it from that savepoint so a restart or an upgrade keeps the state. The
  flow reads the Flink REST API again after the relaunch and fails the run unless a job is running from
  the savepoint and its input record count is still climbing.

inputs:
  - id: rest_url
    type: STRING
    displayName: Flink JobManager REST URL
    description: Base URL of the Flink cluster REST API, reachable from Kestra, for
      example http://flink-jobmanager:8081.
    defaults: "http://flink-jobmanager:8081"

  - id: job_name
    type: STRING
    displayName: Target job name
    description: Name of the streaming job to drill. Empty picks the first running
      job the cluster reports.
    defaults: ""

  - id: jar_url
    type: STRING
    displayName: Job JAR URL
    description: Location of the job JAR, fetched into Kestra storage then
      relaunched from the savepoint. Needed only for a restore.
    defaults: ""

  - id: entry_class
    type: STRING
    displayName: Job main class
    description: Fully qualified main class inside the JAR, used when the job is relaunched.
    defaults: ""

  - id: parallelism
    type: INT
    displayName: Restore parallelism
    description: Parallelism for the relaunched job.
    defaults: 2

  - id: savepoint_dir
    type: STRING
    displayName: Savepoint directory
    description: A directory every JobManager and TaskManager can read and write,
      where savepoints are written.
    defaults: "file:///flink-savepoints"
  - id: drain
    type: BOOL
    displayName: Drain on stop
    description: Drain streaming input before the stop with savepoint. Leave false
      for a plain restart.
    defaults: false

  - id: allow_non_restored_state
    type: BOOL
    displayName: Allow non restored state
    description: Let the relaunch skip savepoint state that no longer maps to the
      job graph, for an upgraded job.
    defaults: false

  - id: savepoint_path_override
    type: STRING
    displayName: Savepoint path override
    description: Relaunch from this exact savepoint path instead of the one the stop
      produced, to recover from an earlier savepoint.
    defaults: ""

  - id: min_records_delta
    type: INT
    displayName: Minimum record delta
    description: The relaunched job must read at least this many more records
      between two samples to count as consuming.
    defaults: 1

  - id: poll_seconds
    type: INT
    displayName: Operation timeout seconds
    description: How long to wait for a savepoint or a stop with savepoint to complete.
    defaults: 180

  - id: sample_seconds
    type: INT
    displayName: Sample gap seconds
    description: Seconds between the two record count samples that prove the
      relaunched job is consuming.
    defaults: 12

  - id: auto_remediate
    type: BOOL
    displayName: Allow the restart
    description: Must be true, with dry_run false and an approval, before the job is
      stopped and relaunched.
    defaults: false

  - id: dry_run
    type: BOOL
    displayName: Dry run
    description: When true the flow only takes a savepoint and leaves the job running.
    defaults: true

tasks:
  - id: detect
    type: io.kestra.plugin.core.http.Request
    description: Read the running jobs from the Flink REST API.
    uri: "{{ inputs.rest_url }}/jobs/overview"
    method: GET

  - id: assess
    type: io.kestra.plugin.core.output.OutputValues
    description: Keep the running jobs, apply the name filter and pick the one job to drill.
    values:
      r: >-
        {{ {'jobs': fromJson(outputs.detect.body).jobs, 'name': inputs.job_name}
        | jq('
           .name as $n
           | [ .jobs[] | select(.state == "RUNNING") | {jid: .jid, name: .name} ] as $running
           | ($running | if $n == "" then . else map(select(.name == $n)) end) as $match
           | { running: $running, running_count: ($running | length),
               match_count: ($match | length), target: ($match | first),
               found: (($match | length) > 0) }') | first | toJson }}

  - id: findings_alert
    type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
    description: Name the job the drill is about to savepoint, because it touches a
      live streaming job.
    runIf: "{{ fromJson(outputs.assess.values.r).found }}"
    url: "{{ secret('SLACK_WEBHOOK_URL') }}"
    payload: |
      {{ {'text': 'Flink savepoint restart drill on ' ~ inputs.rest_url ~ ': target ' ~ fromJson(outputs.assess.values.r).target.name ~ ' (' ~ fromJson(outputs.assess.values.r).target.jid ~ '), ' ~ fromJson(outputs.assess.values.r).running_count ~ ' running. dry_run ' ~ inputs.dry_run ~ ', auto_remediate ' ~ inputs.auto_remediate ~ '.'} | toJson }}

  - id: drill
    type: io.kestra.plugin.core.flow.If
    description: Only run the drill when a target job exists.
    condition: "{{ fromJson(outputs.assess.values.r).found }}"
    then:
      - id: describe
        type: io.kestra.plugin.core.http.Request
        description: Read the target job detail for the record count before the savepoint.
        uri: "{{ inputs.rest_url }}/jobs/{{ fromJson(outputs.assess.values.r).target.jid
          }}"
        method: GET

      - id: baseline
        type: io.kestra.plugin.core.output.OutputValues
        description: Sum the records read across the job vertices as the before number.
        values:
          b: >-
            {{ {'v': fromJson(outputs.describe.body)} | jq('
               .v as $v
               | { jid: $v.jid, name: $v.name, state: $v.state,
                   records_in: ([ $v.vertices[] | .metrics["read-records"] // 0 ] | add) }') | first | toJson }}

      - id: report
        type: io.kestra.plugin.core.log.Log
        description: One line naming the target job and its record count before the
          savepoint.
        message: >-
          Flink drill target {{ fromJson(outputs.baseline.values.b).name }} ({{
          fromJson(outputs.baseline.values.b).jid }}) state {{
          fromJson(outputs.baseline.values.b).state }}, records read so far {{
          fromJson(outputs.baseline.values.b).records_in }}, {{
          fromJson(outputs.assess.values.r).running_count }} running jobs on {{
          inputs.rest_url }}.

      - id: trigger_savepoint
        type: io.kestra.plugin.core.http.Request
        description: Ask Flink for a savepoint of the running job without stopping it.
        uri: "{{ inputs.rest_url }}/jobs/{{ fromJson(outputs.assess.values.r).target.jid
          }}/savepoints"
        method: POST
        contentType: application/json
        body: "{{ {'target-directory': inputs.savepoint_dir, 'cancel-job': false} |
          toJson }}"

      - id: wait_savepoint
        type: io.kestra.plugin.core.flow.LoopUntil
        description: Poll the savepoint request until it completes.
        condition: "{{ fromJson(outputs.poll_savepoint.body).status.id == 'COMPLETED' }}"
        checkFrequency:
          interval: PT2S
          maxDuration: "PT{{ inputs.poll_seconds }}S"
        failOnMaxReached: true
        tasks:
          - id: poll_savepoint
            type: io.kestra.plugin.core.http.Request
            uri: "{{ inputs.rest_url }}/jobs/{{ fromJson(outputs.assess.values.r).target.jid
              }}/savepoints/{{
              fromJson(outputs.trigger_savepoint.body)['request-id'] }}"
            method: GET
      - id: read_savepoint
        type: io.kestra.plugin.core.http.Request
        description: Read the completed savepoint location.
        uri: "{{ inputs.rest_url }}/jobs/{{ fromJson(outputs.assess.values.r).target.jid
          }}/savepoints/{{
          fromJson(outputs.trigger_savepoint.body)['request-id'] }}"
        method: GET

      - id: confirm_still_running
        type: io.kestra.plugin.core.http.Request
        description: The savepoint must not have disturbed the job, so read its state again.
        uri: "{{ inputs.rest_url }}/jobs/{{ fromJson(outputs.assess.values.r).target.jid
          }}"
        method: GET

      - id: savepoint_report
        type: io.kestra.plugin.core.output.OutputValues
        description: Record the savepoint path and that the job is still running.
        values:
          s: >-
            {{ {'sp': fromJson(outputs.read_savepoint.body), 'job':
            fromJson(outputs.confirm_still_running.body)} | jq('
               { savepoint_path: .sp.operation.location, status: .sp.status.id, job_state: .job.state }') | first | toJson }}

      - id: savepoint_log
        type: io.kestra.plugin.core.log.Log
        description: Report the savepoint and whether the job kept running.
        message: >-
          Savepoint {{ fromJson(outputs.savepoint_report.values.s).status }} at
          {{ fromJson(outputs.savepoint_report.values.s).savepoint_path }}. Job
          state after the savepoint is {{
          fromJson(outputs.savepoint_report.values.s).job_state }}.

      - id: savepoint_guard
        type: io.kestra.plugin.core.flow.If
        description: A savepoint only run must leave a completed savepoint and a still
          running job.
        condition: "{{ fromJson(outputs.savepoint_report.values.s).status != 'COMPLETED'
          or fromJson(outputs.savepoint_report.values.s).job_state != 'RUNNING'
          }}"
        then:
          - id: savepoint_failed
            type: io.kestra.plugin.core.execution.Fail
            errorMessage: >-
              Savepoint drill did not leave a clean state on {{ inputs.rest_url
              }}. Savepoint status {{
              fromJson(outputs.savepoint_report.values.s).status }}, job state
              {{ fromJson(outputs.savepoint_report.values.s).job_state }}.
      - id: remediate_gate
        type: io.kestra.plugin.core.flow.If
        description: Stopping and relaunching the job needs auto_remediate true and
          dry_run false.
        condition: "{{ inputs.auto_remediate and not inputs.dry_run }}"
        then:
          - id: approve_restart
            type: io.kestra.plugin.core.flow.Pause
            description: A restart stops a live streaming job, so a human has to approve it.
            pauseDuration: PT2H
            behavior: FAIL
            onResume:
              - id: approved
                type: BOOL
                defaults: false
                description: Set true to stop the job with a savepoint and relaunch it from that
                  savepoint.
              - id: reason
                type: STRING
                defaults: "planned restart"
                description: Recorded in the execution so the decision is auditable.

          - id: restart
            type: io.kestra.plugin.core.flow.If
            condition: "{{ (outputs.approve_restart.onResume is defined) and
              outputs.approve_restart.onResume.approved }}"
            then:
              - id: stop_with_savepoint
                type: io.kestra.plugin.core.http.Request
                description: Stop the job and write a fresh savepoint to restart from.
                uri: "{{ inputs.rest_url }}/jobs/{{ fromJson(outputs.assess.values.r).target.jid
                  }}/stop"
                method: POST
                contentType: application/json
                body: "{{ {'targetDirectory': inputs.savepoint_dir, 'drain': inputs.drain} |
                  toJson }}"

              - id: wait_stop
                type: io.kestra.plugin.core.flow.LoopUntil
                description: Poll the stop request until the savepoint completes.
                condition: "{{ fromJson(outputs.poll_stop.body).status.id == 'COMPLETED' }}"
                checkFrequency:
                  interval: PT2S
                  maxDuration: "PT{{ inputs.poll_seconds }}S"
                failOnMaxReached: true
                tasks:
                  - id: poll_stop
                    type: io.kestra.plugin.core.http.Request
                    uri: "{{ inputs.rest_url }}/jobs/{{ fromJson(outputs.assess.values.r).target.jid
                      }}/savepoints/{{
                      fromJson(outputs.stop_with_savepoint.body)['request-id']
                      }}"
                    method: GET
              - id: read_stop_savepoint
                type: io.kestra.plugin.core.http.Request
                description: Read the savepoint the stop wrote.
                uri: "{{ inputs.rest_url }}/jobs/{{ fromJson(outputs.assess.values.r).target.jid
                  }}/savepoints/{{
                  fromJson(outputs.stop_with_savepoint.body)['request-id'] }}"
                method: GET

              - id: choose_savepoint
                type: io.kestra.plugin.core.output.OutputValues
                description: Relaunch from the override when set, otherwise from the savepoint
                  the stop produced.
                values:
                  path: >-
                    {{ {'ovr': inputs.savepoint_path_override, 'sp':
                    fromJson(outputs.read_stop_savepoint.body).operation.location}
                    | jq('if .ovr == "" then .sp else .ovr end') | first }}

              - id: download_jar
                type: io.kestra.plugin.core.http.Download
                description: Fetch the job JAR into Kestra storage so the plugin can upload it
                  to Flink.
                uri: "{{ inputs.jar_url }}"

              - id: restore
                type: io.kestra.plugin.flink.Submit
                description: Relaunch the job from the savepoint, which is the state preserving
                  restart.
                restUrl: "{{ inputs.rest_url }}"
                jarUri: "{{ outputs.download_jar.uri }}"
                entryClass: "{{ inputs.entry_class }}"
                parallelism: "{{ inputs.parallelism }}"
                restoreFromSavepoint: "{{ outputs.choose_savepoint.values.path }}"
                allowNonRestoredState: "{{ inputs.allow_non_restored_state }}"
                allowFailure: true

              - id: restored
                type: io.kestra.plugin.core.output.OutputValues
                description: Capture the relaunched job id, empty when the relaunch was
                  rejected.
                values:
                  jid: "{{ outputs.restore.jobId ?? '' }}"

              - id: verify_restart
                type: io.kestra.plugin.core.flow.If
                description: With a relaunched job id, re-read the cluster. Without one, the
                  restore was rejected.
                condition: "{{ outputs.restored.values.jid != '' }}"
                then:
                  - id: settle_1
                    type: io.kestra.plugin.core.flow.Sleep
                    duration: "PT{{ inputs.sample_seconds }}S"

                  - id: sample_1
                    type: io.kestra.plugin.core.http.Request
                    description: First read of the relaunched job detail.
                    uri: "{{ inputs.rest_url }}/jobs/{{ outputs.restored.values.jid }}"
                    method: GET

                  - id: checkpoints
                    type: io.kestra.plugin.core.http.Request
                    description: Read the restore record that names the savepoint the job started
                      from.
                    uri: "{{ inputs.rest_url }}/jobs/{{ outputs.restored.values.jid }}/checkpoints"
                    method: GET

                  - id: settle_2
                    type: io.kestra.plugin.core.flow.Sleep
                    duration: "PT{{ inputs.sample_seconds }}S"

                  - id: sample_2
                    type: io.kestra.plugin.core.http.Request
                    description: Second read of the relaunched job detail, to prove the record count
                      advanced.
                    uri: "{{ inputs.rest_url }}/jobs/{{ outputs.restored.values.jid }}"
                    method: GET

                  - id: verdict
                    type: io.kestra.plugin.core.output.OutputValues
                    description: The relaunched job must be running, resumed from the savepoint and
                      reading more records.
                    values:
                      v: >-
                        {{ {'s1': fromJson(outputs.sample_1.body), 's2':
                        fromJson(outputs.sample_2.body), 'cp':
                        fromJson(outputs.checkpoints.body), 'want':
                        inputs.min_records_delta, 'sp':
                        outputs.choose_savepoint.values.path} | jq('
                           .want as $want | .sp as $sp
                           | ([ .s1.vertices[] | .metrics["read-records"] // 0 ] | add) as $r1
                           | ([ .s2.vertices[] | .metrics["read-records"] // 0 ] | add) as $r2
                           | (.cp.latest.restored // {}) as $rest
                           | { state: .s2.state, records_before: $r1, records_after: $r2, delta: ($r2 - $r1),
                               restored_path: ($rest.external_path // ""), is_savepoint: ($rest.is_savepoint // false),
                               expected_path: $sp, running: (.s2.state == "RUNNING"),
                               resumed: ((($rest.is_savepoint // false) == true) and (($rest.external_path // "") == $sp)),
                               consuming: (($r2 - $r1) >= $want) }
                           | . + { ok: (.running and .resumed and .consuming) }') | first | toJson }}
                  - id: verify
                    type: io.kestra.plugin.core.flow.If
                    description: Fail the run unless the relaunched job is running, resumed from the
                      savepoint and consuming.
                    condition: "{{ not fromJson(outputs.verdict.values.v).ok }}"
                    then:
                      - id: restart_not_verified
                        type: io.kestra.plugin.core.execution.Fail
                        errorMessage: >-
                          Flink restart did not verify on {{ inputs.rest_url }}.
                          Job state {{ fromJson(outputs.verdict.values.v).state
                          }}, running {{
                          fromJson(outputs.verdict.values.v).running }}, resumed
                          from savepoint {{
                          fromJson(outputs.verdict.values.v).resumed }}
                          (restored path {{
                          fromJson(outputs.verdict.values.v).restored_path }},
                          expected {{
                          fromJson(outputs.verdict.values.v).expected_path }}),
                          records {{
                          fromJson(outputs.verdict.values.v).records_before }}
                          then {{
                          fromJson(outputs.verdict.values.v).records_after }}
                          (delta {{ fromJson(outputs.verdict.values.v).delta }},
                          needed {{ inputs.min_records_delta }}).
                    else:
                      - id: restart_verified
                        type: io.kestra.plugin.core.log.Log
                        message: >-
                          Flink job relaunched and verified on {{
                          inputs.rest_url }}. State {{
                          fromJson(outputs.verdict.values.v).state }}, resumed
                          from {{
                          fromJson(outputs.verdict.values.v).restored_path }},
                          records read {{
                          fromJson(outputs.verdict.values.v).records_before }}
                          then {{
                          fromJson(outputs.verdict.values.v).records_after }}
                          (delta {{ fromJson(outputs.verdict.values.v).delta
                          }}). Approved with reason "{{
                          outputs.approve_restart.onResume.reason }}".
                else:
                  - id: restore_did_not_start
                    type: io.kestra.plugin.core.execution.Fail
                    errorMessage: >-
                      The relaunch did not start a job on {{ inputs.rest_url }}.
                      The stop savepoint was {{
                      fromJson(outputs.read_stop_savepoint.body).operation.location
                      }} and the restore from {{
                      outputs.choose_savepoint.values.path }} returned no job
                      id, so the state was not restored.
            else:
              - id: not_approved
                type: io.kestra.plugin.core.log.Log
                message: "Restart on {{ inputs.rest_url }} was not approved, so the job was left
                  running after only a savepoint."

        else:
          - id: no_restart
            type: io.kestra.plugin.core.log.Log
            message: >-
              No restart on {{ inputs.rest_url }}. dry_run {{ inputs.dry_run }},
              auto_remediate {{ inputs.auto_remediate }}. A savepoint was taken
              and the job left running.
    else:
      - id: no_target
        type: io.kestra.plugin.core.log.Log
        message: >-
          No running job to drill on {{ inputs.rest_url }} (job_name filter "{{
          inputs.job_name }}", {{
          fromJson(outputs.assess.values.r).running_count }} running).

errors:
  - id: alert_flow_failure
    type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
    url: "{{ secret('SLACK_WEBHOOK_URL') }}"
    payload: |
      {{ {'text': 'Flink savepoint restart drill failed on ' ~ inputs.rest_url ~ ' (execution ' ~ execution.id ~ '). The streaming job state may not be safe to restart until this run succeeds.'} | toJson }}

triggers:
  - id: daily
    type: io.kestra.plugin.core.trigger.Schedule
    description: Shipped disabled and in safe mode. Enable it once the inputs match
      your cluster.
    cron: "30 2 * * *"
    disabled: true
    inputs:
      dry_run: true
      auto_remediate: false

A long running Flink streaming job holds state in memory and in its checkpoints. Restarting it for a config change, a version bump or a node drain means stopping it and starting it again. A plain stop and start throws that state away. The job then reprocesses from the start or from the latest offsets with empty operator state, which breaks exactly the aggregations and the deduplication the job exists for. The safe path is a savepoint: snapshot the state, stop from that snapshot, start the new run from it.

This blueprint runs that savepoint and restart as a drill against a real Flink cluster. It finds the running job through the REST API, takes a savepoint and in remediate mode stops the job with a savepoint then relaunches it from that savepoint with the io.kestra.plugin.flink.Submit task. It then reads the cluster again and fails the run unless a job is running from the savepoint and its input record count is still climbing. A savepoint only dry run proves the state can be captured without touching the live job.

This blueprint was created by zkasuran.

Safety

The restart is destructive: it stops a live streaming job. It runs only when auto_remediate is true and dry_run is false, then only after a human approves the approve_restart pause. The default inputs and the schedule leave the flow in savepoint only mode.

The savepoint and the stop with savepoint are driven through the Flink REST API with io.kestra.plugin.core.http.Request. The dedicated TriggerSavepoint and CancelJob tasks in the bundled plugin build their status poller with a retry interval equal to its own ceiling, which Kestra rejects before the first poll, so they cannot complete a savepoint on this version. The relaunch uses the Submit task, which works. The REST calls hit the same endpoints those tasks would.

How it works

detect (io.kestra.plugin.core.http.Request) reads GET /jobs/overview from the Flink REST API.
assess (io.kestra.plugin.core.output.OutputValues) keeps the RUNNING jobs, applies the job_name filter and picks the target. findings_alert posts to Slack naming the job the drill will touch. drill (io.kestra.plugin.core.flow.If) runs the rest only when a target exists.
describe and baseline read the target job and sum read-records across its vertices as the before number.
trigger_savepoint posts POST /jobs/{id}/savepoints with cancel-job false. wait_savepoint (io.kestra.plugin.core.flow.LoopUntil) polls the request until it is COMPLETED. read_savepoint reads the savepoint path. savepoint_guard fails the run if the savepoint did not complete or the job did not stay RUNNING, so a dry run that disturbed the job is caught.
remediate_gate requires auto_remediate true and dry_run false. approve_restart (io.kestra.plugin.core.flow.Pause) holds for a human.
On approval, stop_with_savepoint posts POST /jobs/{id}/stop, wait_stop polls it to COMPLETED and read_stop_savepoint reads the new savepoint path. choose_savepoint uses savepoint_path_override when set, otherwise that path.
download_jar (io.kestra.plugin.core.http.Download) fetches the job JAR into Kestra storage. restore (io.kestra.plugin.flink.Submit) relaunches the job with restoreFromSavepoint set to the chosen path.
verify_restart samples the relaunched job twice, reads GET /jobs/{id}/checkpoints for the restore record, then verify fails the run through io.kestra.plugin.core.execution.Fail unless the job is RUNNING, its latest.restored.external_path is the savepoint it was told to use and its read-records advanced by at least min_records_delta. A relaunch that returned no job id fails too.
errors alerts Slack when the flow fails, for example when the REST API is unreachable.

What you get

A savepoint of the live job on demand, with the path it was written to, taken without stopping the job.
A full restart drill that stops the job with a savepoint and brings it back from that savepoint, so you know the restart keeps state before you need it during an incident.
A verification that reads the cluster after the relaunch and fails the run unless the job came back from the savepoint and is reading records again. A restart that silently started from scratch does not pass.
A Slack alert on the job the drill targets and on any failure.

Who it's for

Teams running stateful Flink streaming jobs that cannot lose state across a restart or an upgrade.
Platform engineers who want a rehearsed, approved savepoint and restart path instead of a manual sequence of REST calls under pressure.
Anyone preparing a Flink version bump or a node drain who wants proof the state survives first.

Why orchestrate this with Kestra

A safe restart is several ordered REST calls with waits in between: trigger a savepoint, poll it to completion, stop with a savepoint, poll again, relaunch from the exact path, then read the cluster back. Kestra adds the approval gate before anything stops, the record count comparison that decides whether the run passes or fails, the Slack alerts and an execution history that shows every drill and the savepoint path it used. The relaunch goes through the typed io.kestra.plugin.flink.Submit task.

Tested end to end

On Kestra 2.0.5 OSS against a real Flink 1.20.5 standalone cluster (one JobManager, one TaskManager) on the same Docker network, with a local HTTP sink standing in for Slack. The JobManager and the TaskManager shared one savepoint and checkpoint volume, with execution.checkpointing.interval set to 10s. The job under drill was the Flink StateMachineExample, an unbounded keyed streaming job that reads a generated event stream. Every number below was confirmed against the Flink REST API directly (job state, the savepoint path on the shared volume and the read-records metric), not only through Kestra's log.

Run	Inputs	Result
detect, no target	defaults, no job running	execution `7KODBYMFuON8426yK1NxBQ` succeeded with state no target. Nothing was touched
dry run savepoint	`dry_run` true against a running `StateMachineExample`	execution `6RuNAVKNCuP1jpSSx7CWy9` took savepoint `savepoint-2e2b34-d07eb4f31d58`, its `_metadata` file was present on the shared volume and the job stayed `RUNNING`
restart	`auto_remediate` true, `dry_run` false, approved	execution `7bEi59j7wjGR4IayiGAn4c` stopped the job with savepoint `savepoint-2e2b34-9144bca657b7` then relaunched job `a9b023fcddc920f55565a4ff65fc5baa`. The relaunched job reported `latest.restored.external_path` equal to that savepoint with `is_savepoint` true, so it resumed from the savepoint. Its `read-records` went from 0 to 24413 across the two in run samples. A later REST sample read 39522014 then 39580463 over 14s, so it kept consuming
failure, bad savepoint	`savepoint_path_override` set to a path that does not exist, approved	execution `3QgQHKQJwsczE1NcMUH7Nz` stopped the job with savepoint `savepoint-a9b023-ba802c3b6e04`, the relaunch returned HTTP 400 with no job id, `verify` failed the run and the `errors` block posted to Slack. No job was left running, which is the loss the drill is built to catch
REST API down	`rest_url` at a host that does not resolve	execution `6claRATZ8ZZF2Sto5h8DfZ` failed at `detect` and the `errors` block posted to Slack

The savepoint paths and the record counts above were read from the Flink REST API and the shared volume directly, not only from Kestra's log. The relaunched job restored keyed state from the savepoint while its read-records counter started again at 0, which is correct: the counter is per run while the state is what the savepoint carries. The restore record is the proof it resumed.

Prerequisites

A Flink cluster whose REST API is reachable from Kestra at rest_url.
A savepoint directory every JobManager and TaskManager can read and write (one shared volume or object storage such as s3://). A savepoint written on one node and read on another fails when the path is local to a single container.
Checkpointing or at least a configured state backend on the cluster, so savepoints can be written.
For a restore, the job JAR at jar_url reachable from Kestra and its entry_class. The flow downloads the JAR into Kestra storage, because the Submit task reads the JAR from Kestra storage rather than an arbitrary URL.

Secrets

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

Inputs

Input	Default	Purpose
`rest_url`	`http://flink-jobmanager:8081`	Flink JobManager REST base URL
`job_name`	empty	Name filter, empty picks the first running job
`jar_url`	empty	Job JAR URL, fetched into Kestra storage for the relaunch
`entry_class`	empty	Fully qualified main class used on the relaunch
`parallelism`	`2`	Parallelism for the relaunched job
`savepoint_dir`	`file:///flink-savepoints`	Shared directory where savepoints are written
`drain`	`false`	Drain streaming input before the stop with savepoint
`allow_non_restored_state`	`false`	Skip savepoint state that no longer maps to the job graph
`savepoint_path_override`	empty	Relaunch from this exact path instead of the stop savepoint
`min_records_delta`	`1`	Records the relaunched job must gain between the two samples
`poll_seconds`	`180`	Timeout for a savepoint or a stop with savepoint
`sample_seconds`	`12`	Seconds between the two record count samples
`auto_remediate`	`false`	Must be true before a restart
`dry_run`	`true`	Must be false before a restart

Quick start

Add the SLACK_WEBHOOK_URL secret.
Set rest_url to your Flink JobManager REST URL and run with the defaults. The flow takes a savepoint of the running job and leaves it running. Read the savepoint path in the logs.
Set job_name if more than one job runs and you want a specific one.
To rehearse the restart, set jar_url and entry_class for the job, set auto_remediate to true and dry_run to false, run it and approve the paused execution. The flow stops the job with a savepoint, relaunches it from that savepoint and fails if it did not come back consuming.
Enable the daily trigger once the inputs match your cluster. It stays in savepoint only mode.

How to extend

Loop over several jobs and drill each one in a single execution.
Store the savepoint path in a KV pair so a later run can relaunch from a known good savepoint with savepoint_path_override.
Point savepoint_dir at object storage so the savepoint survives the loss of any single node.
Add a step that bumps the JAR to a new version on relaunch, with allow_non_restored_state true, to turn the drill into an upgrade.
Compare read-records against the before number over a longer window to assert throughput, not just that the count moved.

Flink Stateful Job Savepoint and Restart Drill

Safety

How it works

What you get

Who it's for

Why orchestrate this with Kestra

Tested end to end

Prerequisites

Secrets

Inputs

Quick start

How to extend

Links

Tasks

Orchestrate with Kestra

Share this Blueprint

Create a Blueprint

New to Kestra?

Flink Stateful Job Savepoint and Restart Drill

Safety

How it works

What you get

Who it's for

Why orchestrate this with Kestra

Tested end to end

Prerequisites

Secrets

Inputs

Quick start

How to extend

Links

Tasks

Orchestrate with Kestra

Share this Blueprint

Create a Blueprint

More Data Blueprints

Audit Couchbase GSI Secondary Indexes and Detect Offline States

Cancel runaway Couchbase N1QL queries that starve the query service

Publish CSV snapshots to PostgreSQL after bounded parallel staging

Run scheduled sales analytics on CSV data with DuckDB and post results to Slack

Extract Data from Couchbase and Transform It with Pandas for Analytics

Automated LLM Web Documentation Harvester with Crawl4AI

More Infrastructure Blueprints

Autonomous Multi-Agent Software Engineering and Self-Healing Pipeline

Govern Security Exception Requests End to End, from Developer Request to an Auditable Aikido Snooze

Aikido Domain Scan Staleness Compliance Gate

Gate Releases on New Licence Risk with an Aikido SBOM Diff Instead of a Full Backlog Scan

Autonomous React Component Generator with Deep Learning Visual QA

Autonomous React PR Reviewer with Playwright Visual Regression Diff

New to Kestra?