Kafka Connect Paused Connector Resume Guard

id: kafka-connect-paused-connector-resume-guard
namespace: company.team

description: |
  Find Kafka Connect connectors that are paused or stopped instead of failed, hold them until
  they have been silent for a whole grace window, then resume only those behind an approval gate
  and re-read the Connect REST API to prove each one is running again. A paused connector is not
  a failed connector. It raises no error, fills no error log, trips no FAILED check. It just
  stops moving records. Connectors on an expected-paused allowlist are never resumed, connectors
  silent for less than the grace window are only reported. Task states are read as well as
  the connector state because a connector can read RUNNING while every task under it is paused.

inputs:
  - id: connect_url
    type: STRING
    displayName: Kafka Connect REST endpoint
    description: Base URL of the Connect worker REST API, for example
      http://connect:8083. Held as a secret because it is an internal admin
      endpoint.
    defaults: "{{ secret('KAFKA_CONNECT_URL') }}"

  - id: expected_paused
    type: ARRAY
    itemType: STRING
    displayName: Connectors that are allowed to stay paused
    description: Connector names paused on purpose, for example one held during a
      migration or a backfill that is being released by hand. These are reported
      on every run but never resumed.
    defaults: []

  - id: min_paused_minutes
    type: INT
    displayName: Minutes a connector must stay silent before it can be resumed
    description: Measured from the first run that saw the connector silent in its
      current state. Anything newer is reported, never resumed, so a maintenance
      window that is still running is left alone.
    defaults: 30

  - id: verify_delay
    type: DURATION
    displayName: Wait before the re-read
    description: A resumed connector needs a moment for the worker to hand its tasks
      back out, so the re-read waits this long before deciding whether the
      resume landed.
    defaults: PT10S

  - id: auto_remediate
    type: BOOL
    displayName: Auto remediate
    defaults: false

  - id: dry_run
    type: BOOL
    displayName: Dry run
    description: Must be false, together with auto_remediate, for ConnectorResume to run.
    defaults: true

variables:
  kv_key: "kafka_connect_paused_{{ inputs.connect_url | slugify }}"

tasks:
  - id: freeze
    type: io.kestra.plugin.core.output.OutputValues
    description: One timestamp for the whole run, so the detect, the cursor and the
      age check agree.
    values:
      now_iso: >-
        {{ execution.startDate | date("yyyy-MM-dd'T'HH:mm:ss.SSS'Z'",
        timeZone="UTC") }}
      cutoff_iso: >-
        {{ execution.startDate | dateAdd(-1 * inputs.min_paused_minutes,
        'MINUTES') | date("yyyy-MM-dd'T'HH:mm:ss.SSS'Z'", timeZone="UTC") }}

  - id: list_connectors
    type: io.kestra.plugin.kafka.ConnectorList
    description: >-
      Read-only. Every connector on the worker with its connector state and the
      state of each of its tasks, which is the only call that shows a paused
      connector at all.
    connectUrl: "{{ inputs.connect_url }}"
    expandStatus: true
    retry:
      type: constant
      interval: PT10S
      maxAttempts: 3

  - id: analyze
    type: io.kestra.plugin.core.output.OutputValues
    description: >-
      A connector counts as silent when its own state is PAUSED or STOPPED, as
      well as when it reads otherwise while one of its tasks is PAUSED.
      Allowlisted names are split out here so they can never reach the resume
      step.
    values:
      result: >-
        {{ {'c': (outputs.list_connectors.connectors ?? []), 'allow':
        inputs.expected_paused} | jq('.c as $c | .allow as $allow | [$c[] |
        {name: .connectorName, state: (.connectorState // "UNKNOWN"), tasks:
        ((.tasks // []) | length), paused_tasks: ((.tasks // []) |
        map(select(.state == "PAUSED")) | length), failed_tasks: ((.tasks // [])
        | map(select(.state == "FAILED")) | length)} | . + {sig: (.state + "/" +
        (.paused_tasks | tostring) + "/" + (.tasks | tostring)), reason: (if
        .state == "PAUSED" then "connector PAUSED" elif .state == "STOPPED" then
        "connector STOPPED" else "connector " + .state + " with " +
        (.paused_tasks | tostring) + " of " + (.tasks | tostring) + " tasks
        PAUSED" end)}] as $all | [$all[] | select(.state == "PAUSED" or .state
        == "STOPPED" or (.tasks > 0 and .paused_tasks > 0))] as $silent |
        {total: ($all | length), healthy: ([$all[] | select(.state == "RUNNING"
        and .paused_tasks == 0 and .failed_tasks == 0)] | length), failed:
        ([$all[] | select(.state == "FAILED" or .failed_tasks > 0)] | length),
        allowlisted: ([$silent[] | select(((([.name] - $allow)) | length) == 0)]
        | map(.name) | sort), candidates: ([$silent[] | select(((([.name] -
        $allow)) | length) > 0)] | sort_by(.name)), candidate_names: ([$silent[]
        | select(((([.name] - $allow)) | length) > 0)] | map(.name) | sort),
        inventory: ([$all[] | .name + " state=" + .state + " tasks=" + ((.tasks
        - .paused_tasks) | tostring) + "/" + (.tasks | tostring) + " running"] |
        sort | join(" | "))}') | first | toJson }}

  - id: route
    type: io.kestra.plugin.core.flow.If
    condition: "{{ fromJson(outputs.analyze.values.result).candidates | length > 0 }}"
    then:
      - id: cursor_merge
        type: io.kestra.plugin.core.output.OutputValues
        description: >-
          Keep the first time each candidate was seen in its current silent
          state. A candidate that is new gets now. So does one whose state
          moved, so a connector that was stopped then paused starts its clock
          again rather than inheriting an old date.
        values:
          merged: >-
            {{ {'cur': (kv(render(vars.kv_key), errorOnMissing=false) ?? ''),
            'cand': fromJson(outputs.analyze.values.result).candidates, 'now':
            outputs.freeze.values.now_iso} | jq('(.cur | if type=="string" then
            (if . == "" then {} else fromjson end) else . end) as $c | .now as
            $n | reduce .cand[] as $x ({}; . + {($x.name): (if ($c[$x.name] !=
            null and $c[$x.name].sig == $x.sig) then ($x + {first_seen:
            $c[$x.name].first_seen}) else ($x + {first_seen: $n}) end)})') |
            first | toJson }}

      - id: save_cursor
        type: io.kestra.plugin.core.kv.Set
        description: No ttl. This is a first-seen record that must not expire out from
          under the age check.
        key: "{{ render(vars.kv_key) }}"
        kvType: JSON
        value: "{{ outputs.cursor_merge.values.merged }}"

      - id: split
        type: io.kestra.plugin.core.output.OutputValues
        description: Candidates past the grace window become targets. The rest are held
          and reported.
        values:
          target_names: >-
            {{ {'cur': fromJson(outputs.cursor_merge.values.merged), 'cutoff':
            outputs.freeze.values.cutoff_iso} | jq('.cutoff as $cut | [.cur |
            to_entries[] | select(.value.first_seen <= $cut) | .key] | sort') |
            first | toJson }}
          held_names: >-
            {{ {'cur': fromJson(outputs.cursor_merge.values.merged), 'cutoff':
            outputs.freeze.values.cutoff_iso} | jq('.cutoff as $cut | [.cur |
            to_entries[] | select(.value.first_seen > $cut) | .key] | sort') |
            first | toJson }}
          detail: >-
            {{ {'cur': fromJson(outputs.cursor_merge.values.merged)} | jq('[.cur
            | to_entries[] | .value.name + " (" + .value.reason + ", silent
            since " + .value.first_seen + ")"] | sort | join(" | ")') | first }}

      - id: log_findings
        type: io.kestra.plugin.core.log.Log
        message: >-
          Paused connector scan on the Connect worker: {{
          fromJson(outputs.analyze.values.result).total }} connectors, {{
          fromJson(outputs.analyze.values.result).healthy }} running with every
          task running, {{ fromJson(outputs.analyze.values.result).failed }}
          failed (left to a restart monitor). {{
          fromJson(outputs.analyze.values.result).candidates | length }} silent
          without being on the allowlist: {{ outputs.split.values.detail }}. {{
          fromJson(outputs.split.values.target_names) | length }} past the {{
          inputs.min_paused_minutes }} minute grace window ({{
          outputs.split.values.target_names | jq('join(", ")') | first }}). {{
          fromJson(outputs.split.values.held_names) | length }} held as too
          recent ({{ outputs.split.values.held_names | jq('join(", ")') | first
          }}). {{ fromJson(outputs.analyze.values.result).allowlisted | length
          }} expected paused, never resumed ({{ outputs.analyze.values.result |
          jq('.allowlisted | join(", ")') | first }}). Inventory: {{
          fromJson(outputs.analyze.values.result).inventory }}.
          auto_remediate={{ inputs.auto_remediate }} dry_run={{ inputs.dry_run
          }}.

      - id: alert
        type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
        url: "{{ secret('SLACK_WEBHOOK_URL') }}"
        payload: >-
          {{ {'text': 'Kafka Connect: ' ~
          (fromJson(outputs.analyze.values.result).candidates | length) ~ '
          connector(s) paused or stopped while nothing is failing, so no FAILED
          check will see them. ' ~ (fromJson(outputs.split.values.target_names)
          | length) ~ ' past the ' ~ inputs.min_paused_minutes ~ ' minute grace
          window (' ~ (outputs.split.values.target_names | jq('join(", ")') |
          first) ~ '). Detail: ' ~ outputs.split.values.detail ~ '.
          auto_remediate=' ~ inputs.auto_remediate ~ ' dry_run=' ~
          inputs.dry_run ~ '.'} | toJson }}

      - id: remediate_gate
        type: io.kestra.plugin.core.flow.If
        condition: "{{ inputs.auto_remediate and not inputs.dry_run and
          (fromJson(outputs.split.values.target_names) | length > 0) }}"
        then:
          - id: approve
            type: io.kestra.plugin.core.flow.Pause
            description: >-
              Resuming a connector restarts data movement. If somebody paused it
              on purpose for a migration, a backfill or a decommission, resuming
              it writes records they stopped on purpose. Read the target list in
              log_findings, check that none of those names belongs on the
              expected_paused list, then resume with approved=true. Resume with
              no answer or kill the execution to cancel.
            onResume:
              - id: approved
                type: BOOL
                displayName: Approve resuming the listed connectors
                required: false

          - id: do_resume
            type: io.kestra.plugin.core.flow.If
            condition: "{{ outputs.approve.onResume.approved ?? false }}"
            then:
              - id: remediate_resume
                type: io.kestra.plugin.core.flow.Loop
                description: >-
                  One connector per iteration and one iteration at a time. The
                  worker rebalances task assignments on every resume, so
                  parallel resumes queue behind each other anyway and a refusal
                  stays attributable to a single connector.
                values: "{{ outputs.split.values.target_names }}"
                concurrencyLimit: 1
                tasks:
                  - id: resume_one
                    type: io.kestra.plugin.kafka.ConnectorResume
                    description: allowFailure keeps the re-read as the authority on what actually
                      resumed.
                    allowFailure: true
                    connectUrl: "{{ inputs.connect_url }}"
                    connectorName: "{{ item.value }}"

              - id: settle
                type: io.kestra.plugin.core.flow.Sleep
                description: The worker accepts a resume, then reassigns tasks. Give it that
                  long before reading back.
                duration: "{{ inputs.verify_delay }}"

              - id: reverify
                type: io.kestra.plugin.kafka.ConnectorList
                description: Read-only re-read of the Connect REST API after the resume. This is
                  the gate.
                connectUrl: "{{ inputs.connect_url }}"
                expandStatus: true

              - id: verify_eval
                type: io.kestra.plugin.core.output.OutputValues
                description: >-
                  A resume counts only when the connector reads RUNNING and
                  every one of its tasks reads RUNNING. A connector that came
                  back with no tasks, with a paused task left behind or that is
                  missing from the list is a failure.
                values:
                  result: >-
                    {{ {'now': (outputs.reverify.connectors ?? []), 'targets':
                    fromJson(outputs.split.values.target_names)} | jq('.targets
                    as $t | [.now[] | select(((([.connectorName] - $t)) |
                    length) == 0) | {name: .connectorName, state:
                    (.connectorState // "UNKNOWN"), tasks: ((.tasks // []) |
                    length), running_tasks: ((.tasks // []) | map(select(.state
                    == "RUNNING")) | length)}] as $after | ([$after[] | .name +
                    " state=" + .state + " running_tasks=" + (.running_tasks |
                    tostring) + "/" + (.tasks | tostring)] | sort) as $report |
                    ([$after[] | select(.state != "RUNNING" or .tasks == 0 or
                    .running_tasks != .tasks) | .name + " state=" + .state + "
                    running_tasks=" + (.running_tasks | tostring) + "/" +
                    (.tasks | tostring)] | sort) as $bad | ([$after[] | .name])
                    as $seen | ([$t[] | select(((([.]) - $seen) | length) > 0) |
                    . + " is no longer on the worker"] | sort) as $gone |
                    {after: ($report | join("; ")), failures: (($bad + $gone) |
                    join("; ")), failure_count: (($bad | length) + ($gone |
                    length))}') | first | toJson }}

              - id: verify_result
                type: io.kestra.plugin.core.flow.If
                condition: "{{ fromJson(outputs.verify_eval.values.result).failure_count > 0 }}"
                then:
                  - id: resume_failed
                    type: io.kestra.plugin.core.execution.Fail
                    errorMessage: >-
                      ConnectorResume ran, but a re-read of the Connect REST API
                      shows a target that is not running. Still wrong: {{
                      fromJson(outputs.verify_eval.values.result).failures }}.
                      Full read back: {{
                      fromJson(outputs.verify_eval.values.result).after }}.
                      Either the worker refused the resume, the connector
                      resumed into a failure (a bad config, a source that is
                      gone, credentials that expired while it sat paused).
                      Another possibility is that somebody deleted it while the
                      execution was waiting for approval.
                else:
                  - id: prune_cursor
                    type: io.kestra.plugin.core.kv.Set
                    description: >-
                      Drop the resumed connectors from the first-seen record, so
                      one paused again later starts a fresh grace window instead
                      of inheriting the old date.
                    key: "{{ render(vars.kv_key) }}"
                    kvType: JSON
                    value: >-
                      {{ {'cur': fromJson(outputs.cursor_merge.values.merged),
                      'done': fromJson(outputs.split.values.target_names)} |
                      jq('.done as $d | (.cur | with_entries(select(((([.key] -
                      $d)) | length) > 0)))') | first | toJson }}

                  - id: resumed_ok
                    type: io.kestra.plugin.core.log.Log
                    message: >-
                      Resumed {{ fromJson(outputs.split.values.target_names) |
                      length }} connector(s) ({{
                      outputs.split.values.target_names | jq('join(", ")') |
                      first }}). A re-read of the Connect REST API confirms each
                      one is RUNNING with every task RUNNING: {{
                      fromJson(outputs.verify_eval.values.result).after }}.
            else:
              - id: not_approved
                type: io.kestra.plugin.core.log.Log
                message: "Pause resumed without approval or killed. No connector was resumed."
        else:
          - id: report_only
            type: io.kestra.plugin.core.log.Log
            message: >-
              Report only, nothing resumed. {{
              fromJson(outputs.split.values.target_names) | length }}
              connector(s) would be resumed with auto_remediate=true and
              dry_run=false: {{ outputs.split.values.target_names | jq('join(",
              ")') | first }}. Current inputs are auto_remediate={{
              inputs.auto_remediate }} dry_run={{ inputs.dry_run }}.
    else:
      - id: clear_cursor
        type: io.kestra.plugin.core.kv.Set
        description: Nothing silent off the allowlist, so reset the first-seen record.
        key: "{{ render(vars.kv_key) }}"
        kvType: JSON
        value: "{}"

      - id: all_moving
        type: io.kestra.plugin.core.log.Log
        message: >-
          No unexpected pause on the Connect worker. {{
          fromJson(outputs.analyze.values.result).total }} connectors checked,
          {{ fromJson(outputs.analyze.values.result).healthy }} running with
          every task running, {{ fromJson(outputs.analyze.values.result).failed
          }} failed (left to a restart monitor), {{
          fromJson(outputs.analyze.values.result).allowlisted | length }} paused
          on purpose and on the allowlist ({{ outputs.analyze.values.result |
          jq('.allowlisted | join(", ")') | first }}). Inventory: {{
          fromJson(outputs.analyze.values.result).inventory }}.

errors:
  - id: alert_failure
    type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
    url: "{{ secret('SLACK_WEBHOOK_URL') }}"
    payload: >-
      {{ {'text': 'Kafka Connect paused connector resume guard failed (execution
      ' ~ execution.id ~ '). No connector was resumed unless a log line says
      resumed. Check the logs.'} | toJson }}

triggers:
  - id: hourly_scan
    type: io.kestra.plugin.core.trigger.Schedule
    description: Shipped disabled and always in safe mode.
    cron: "17 * * * *"
    disabled: true
    inputs:
      auto_remediate: false
      dry_run: true