Kafka Connect Sink Poison Pill Offset Rewind and Skip Gate

id: kafka-connect-sink-poison-pill-offset-rewind-and-skip-gate
namespace: company.streaming

description: |
  Audit Kafka Connect sink connectors for poison pill task crashes and frozen committed offsets,
  quarantine diagnostic failure records to a dead-letter topic, safely rewind or advance committed
  partition offsets via KIP-875 with human approval, and verify recovery on the live cluster.

variables:
  connect_url: "http://localhost:8083"
  connector_name: "orders-jdbc-sink"
  bootstrap_servers: "localhost:9092"
  quarantine_topic: "connect-poison-pill-quarantine"

inputs:
  - id: connect_url
    type: STRING
    displayName: Kafka Connect REST URL
    description: Base URL of the Kafka Connect REST API worker cluster.
    defaults: "http://localhost:8083"

  - id: connector_name
    type: STRING
    displayName: Connector Name
    description: Kafka Connect sink or source connector name to audit, guard, and remediate.
    defaults: "orders-jdbc-sink"

  - id: bootstrap_servers
    type: STRING
    displayName: Kafka Bootstrap Servers
    description: Comma-separated list of Kafka broker endpoints for dead-letter
      quarantine publishing.
    defaults: "localhost:9092"

  - id: quarantine_topic
    type: STRING
    displayName: Dead-Letter Quarantine Topic
    description: Destination Kafka topic where isolated poison pill payloads and
      failure metadata are published.
    defaults: "connect-poison-pill-quarantine"

  - id: publish_to_dlq
    type: BOOL
    displayName: Publish to Dead-Letter Topic
    description: When true, publishes structured poison pill isolation records to
      the quarantine topic.
    defaults: true

  - id: remediation_action
    type: SELECT
    displayName: Remediation Action
    description: Corrective offset action to apply when a poison pill offset or
      stalled partition is identified.
    values:
      - SKIP_POISON_RECORD
      - REWIND_OFFSET
      - AUDIT_ONLY
    defaults: SKIP_POISON_RECORD

  - id: rewind_record_count
    type: INT
    displayName: Rewind Record Count
    description: Number of records to rewind when remediation_action is set to
      REWIND_OFFSET.
    defaults: 1

  - id: stalled_runs_threshold
    type: INT
    displayName: Stalled Runs Threshold
    description: Number of consecutive runs an offset must remain uncommitted with
      failing tasks before triggering remediation.
    defaults: 1

  - id: dry_run
    type: BOOL
    displayName: Dry Run Mode
    description: When true, generates the remediation plan and logs diagnostics
      without altering connector offsets.
    defaults: true

triggers:
  - id: on_connector_failure
    type: io.kestra.plugin.kafka.ConnectorStatusTrigger
    description: Event-based polling trigger that fires automatically when the
      connector or any worker task enters FAILED state.
    connectUrl: "{{ vars.connect_url }}"
    connectorName: "{{ vars.connector_name }}"
    targetState: FAILED
    interval: PT1M
    disabled: true

  - id: periodic_offset_guard
    type: io.kestra.plugin.core.trigger.Schedule
    description: Periodic scheduled audit monitoring connector offset progression
      and task health every 10 minutes.
    cron: "*/10 * * * *"
    disabled: true

tasks:
  - id: validate_inputs
    type: io.kestra.plugin.core.execution.Assert
    description: Assert operational inputs satisfy sanity and threshold constraints.
    conditions:
      - "{{ (inputs.connector_name ?? vars.connector_name) | length > 0 }}"
      - "{{ (inputs.connect_url ?? vars.connect_url) | length > 0 }}"
      - "{{ inputs.rewind_record_count >= 1 }}"
      - "{{ inputs.stalled_runs_threshold >= 1 }}"
    errorMessage: "Invalid configuration inputs. Provide a valid connector name,
      connect URL, and positive rewind/stalled thresholds."

  - id: get_connector_status
    type: io.kestra.plugin.kafka.ConnectorGetStatus
    description: Query Kafka Connect REST API for connector status, worker
      assignments, task states, and stack traces.
    connectUrl: "{{ inputs.connect_url ?? vars.connect_url }}"
    connectorName: "{{ inputs.connector_name ?? vars.connector_name }}"
    retry:
      type: exponential
      interval: PT2S
      maxInterval: PT15S
      maxAttempts: 3

  - id: get_connector_offsets
    type: io.kestra.plugin.kafka.ConnectorGetOffsets
    description: Retrieve current committed partition offsets from Kafka Connect via
      KIP-875 GET /connectors/{name}/offsets.
    connectUrl: "{{ inputs.connect_url ?? vars.connect_url }}"
    connectorName: "{{ inputs.connector_name ?? vars.connector_name }}"
    retry:
      type: exponential
      interval: PT2S
      maxInterval: PT15S
      maxAttempts: 3

  - id: load_previous_offset_state
    type: io.kestra.plugin.core.kv.Get
    description: Retrieve previous run offset snapshot and stalled partition history
      from Kestra KV store.
    key: "kafka_connect_offset_tracker_{{ inputs.connector_name ??
      vars.connector_name }}"
    errorOnMissing: false

  - id: triage_poison_pill_and_compute_offsets
    type: io.kestra.plugin.scripts.python.Script
    description: Analyze failed task stack traces, diff committed offsets against KV
      baseline, isolate poison pill records, and compute target offsets.
    containerImage: python:3.11-slim
    inputFiles:
      status.json: "{{ outputs.get_connector_status | toJson }}"
      offsets.json: "{{ (outputs.get_connector_offsets.offsets ?? []) | toJson }}"
      previous_kv.json: "{{ (outputs.load_previous_offset_state.value ?? {}) | toJson }}"
    env:
      REMEDIATION_ACTION: "{{ inputs.remediation_action }}"
      REWIND_RECORD_COUNT: "{{ inputs.rewind_record_count }}"
      STALLED_RUNS_THRESHOLD: "{{ inputs.stalled_runs_threshold }}"
      CONNECTOR_NAME: "{{ inputs.connector_name ?? vars.connector_name }}"
      EXECUTION_ID: "{{ execution.id }}"
    script: |
      import json
      import os
      import re
      from datetime import datetime, timezone

      try:
          from kestra import Kestra
      except ImportError:
          Kestra = None

      remediation_action = os.environ.get("REMEDIATION_ACTION", "SKIP_POISON_RECORD")
      rewind_record_count = int(os.environ.get("REWIND_RECORD_COUNT", "1"))
      stalled_runs_threshold = int(os.environ.get("STALLED_RUNS_THRESHOLD", "1"))
      connector_name = os.environ.get("CONNECTOR_NAME", "unknown-connector")
      execution_id = os.environ.get("EXECUTION_ID", "manual-run")

      def load_json_file(filename, fallback):
          try:
              if os.path.exists(filename):
                  with open(filename, "r", encoding="utf-8") as f:
                      content = f.read().strip()
                      return json.loads(content) if content else fallback
          except Exception:
              pass
          return fallback

      status_data = load_json_file("status.json", {})
      offsets_data = load_json_file("offsets.json", [])
      previous_kv_raw = load_json_file("previous_kv.json", {})

      previous_kv = previous_kv_raw if isinstance(previous_kv_raw, dict) else {}
      if isinstance(previous_kv, str):
          try:
              previous_kv = json.loads(previous_kv)
          except Exception:
              previous_kv = {}

      tasks = status_data.get("tasks", []) if isinstance(status_data, dict) else []
      connector_state = status_data.get("connectorState", "UNKNOWN")

      failed_tasks = [t for t in tasks if t.get("state") == "FAILED"]
      has_failures = len(failed_tasks) > 0 or connector_state == "FAILED"

      error_traces = []
      poison_hints = []
      for t in failed_tasks:
          trace = t.get("trace") or ""
          task_id = t.get("id", -1)
          error_traces.append(f"Task {task_id}: {trace[:300]}")
          matches = re.findall(r"([a-zA-Z0-9._-]+)[ -]partition[ :]+([0-9]+)[ -]offset[ :]+([0-9]+)", trace, re.IGNORECASE)
          for m in matches:
              poison_hints.append({"topic": m[0], "partition": int(m[1]), "offset": int(m[2])})

      updated_kv = {}
      target_offsets = []
      stalled_partitions = []
      remediation_required = False

      if isinstance(offsets_data, list):
          for entry in offsets_data:
              if not isinstance(entry, dict):
                  continue
              partition_map = entry.get("partition", {})
              offset_map = entry.get("offset", {})

              topic = partition_map.get("kafka_topic", partition_map.get("topic", partition_map.get("table", "default")))
              partition_id = partition_map.get("kafka_partition", partition_map.get("partition", 0))
              partition_key = f"{topic}-{partition_id}"

              current_offset = None
              offset_field_name = "kafka_offset"
              for field in ["kafka_offset", "offset", "position", "incrementing"]:
                  if field in offset_map and isinstance(offset_map[field], (int, float)):
                      current_offset = int(offset_map[field])
                      offset_field_name = field
                      break

              prev_info = previous_kv.get(partition_key, {})
              prev_offset = prev_info.get("last_offset") if isinstance(prev_info, dict) else None
              stalled_runs = prev_info.get("stalled_runs", 0) if isinstance(prev_info, dict) else 0

              is_stalled = False
              if has_failures:
                  if current_offset is not None and prev_offset is not None and current_offset == prev_offset:
                      stalled_runs += 1
                      is_stalled = True
                  else:
                      stalled_runs = 1
                      is_stalled = True
              else:
                  stalled_runs = 0

              updated_kv[partition_key] = {
                  "last_offset": current_offset,
                  "stalled_runs": stalled_runs,
                  "last_audit_timestamp": datetime.now(timezone.utc).isoformat(),
                  "status": "STALLED" if is_stalled else "HEALTHY"
              }

              if is_stalled and stalled_runs >= stalled_runs_threshold:
                  # Always record a stalled partition, regardless of remediation_action, so
                  # AUDIT_ONLY still reports what it found instead of looking identical to HEALTHY.
                  stalled_partitions.append({
                      "partition_key": partition_key,
                      "topic": topic,
                      "partition": partition_id,
                      "current_offset": current_offset,
                      "stalled_runs": stalled_runs
                  })

                  if remediation_action != "AUDIT_ONLY":
                      remediation_required = True

                      target_offset_value = current_offset if current_offset is not None else 0
                      if remediation_action == "SKIP_POISON_RECORD":
                          target_offset_value += 1
                      elif remediation_action == "REWIND_OFFSET":
                          target_offset_value = max(0, target_offset_value - rewind_record_count)

                      new_offset_map = dict(offset_map)
                      new_offset_map[offset_field_name] = target_offset_value
                      target_offsets.append({
                          "partition": partition_map,
                          "offset": new_offset_map
                      })

      quarantine_payload = {
          "eventType": "KAFKA_CONNECT_POISON_PILL_ISOLATION",
          "executionId": execution_id,
          "connectorName": connector_name,
          "connectorState": connector_state,
          "failedTaskCount": len(failed_tasks),
          "stalledPartitions": stalled_partitions,
          "targetOffsets": target_offsets,
          "remediationPolicy": remediation_action,
          "errorTraces": error_traces,
          "timestamp": datetime.now(timezone.utc).isoformat()
      }

      summary_lines = [
          f"*Connector:* `{connector_name}` | *State:* `{connector_state}`",
          f"*Failed Tasks:* {len(failed_tasks)} of {len(tasks)}",
          f"*Stalled Partitions:* {len(stalled_partitions)}",
          f"*Remediation Required:* `{remediation_required}` (Policy: `{remediation_action}`)"
      ]
      if stalled_partitions:
          summary_lines.append("*Partition Breakdown:*")
          for sp in stalled_partitions:
              summary_lines.append(f"- `{sp['partition_key']}`: current offset `{sp['current_offset']}`, stalled `{sp['stalled_runs']}` run(s)")

      output_data = {
          "remediation_required": remediation_required,
          "has_failed_tasks": has_failures,
          "failed_task_count": len(failed_tasks),
          "stalled_partition_count": len(stalled_partitions),
          "target_offsets": target_offsets,
          "quarantine_payload": quarantine_payload,
          "kv_payload": updated_kv,
          "summary_markdown": "\\n".join(summary_lines)
      }

      if Kestra:
          Kestra.outputs(output_data)
      else:
          print("::" + json.dumps({"outputs": output_data}) + "::")

  - id: persist_offset_tracker_kv
    type: io.kestra.plugin.core.kv.Set
    description: Persist updated connector offset snapshot and remediation telemetry
      into Kestra namespace KV store for historical trend tracking.
    key: "kafka_connect_offset_tracker_{{ inputs.connector_name ??
      vars.connector_name }}"
    kvType: JSON
    overwrite: true
    value: "{{ outputs.triage_poison_pill_and_compute_offsets.vars.kv_payload |
      toJson }}"

  - id: gate_remediation
    type: io.kestra.plugin.core.flow.If
    description: Evaluate if stalled partitions or poison pill records require
      quarantine publishing and offset remediation.
    condition: "{{
      outputs.triage_poison_pill_and_compute_offsets.vars.remediation_required
      }}"
    then:
      - id: gate_quarantine_publishing
        type: io.kestra.plugin.core.flow.If
        description: Publish isolated poison pill metadata to quarantine topic when enabled.
        condition: "{{ inputs.publish_to_dlq }}"
        then:
          - id: publish_poison_pill_to_dlq
            type: io.kestra.plugin.kafka.Produce
            description: Publish structured poison pill record metadata, failed task stack
              trace, and offset telemetry to dead-letter quarantine topic.
            properties:
              bootstrap.servers: "{{ inputs.bootstrap_servers ?? vars.bootstrap_servers }}"
            topic: "{{ inputs.quarantine_topic ?? vars.quarantine_topic }}"
            from:
              key: "{{ inputs.connector_name ?? vars.connector_name }}_{{ execution.id }}"
              value: "{{
                outputs.triage_poison_pill_and_compute_offsets.vars.quarantine_payload
                | toJson }}"

      - id: check_dry_run_mode
        type: io.kestra.plugin.core.flow.If
        description: Branch between dry-run audit logging and authorized live cluster
          offset modification.
        condition: "{{ inputs.dry_run }}"
        then:
          - id: log_dry_run_plan
            type: io.kestra.plugin.core.log.Log
            description: Log proposed offset modifications without altering connector
              offsets in dry-run mode.
            message: |
              [DRY RUN] Identified {{ outputs.triage_poison_pill_and_compute_offsets.vars.stalled_partition_count }} stalled partition(s) on {{ inputs.connector_name ?? vars.connector_name }}.
              Target offsets planned for KIP-875 alteration:
              {{ outputs.triage_poison_pill_and_compute_offsets.vars.target_offsets | toJson }}
              Set dry_run=false and approve the pause gate to apply offset modifications to the live cluster.
        else:
          - id: approval_gate
            type: io.kestra.plugin.core.flow.Pause
            description: Pause workflow execution for operator verification before altering
              committed Kafka Connect offsets. Resume to apply or kill to abort.
            onResume:
              - id: approved
                type: BOOL
                displayName: Approve Offset Remediation
                description: Confirm authorization to stop connector tasks and apply altered
                  committed offsets.
                defaults: true

          - id: stop_connector
            type: io.kestra.plugin.core.http.Request
            description: Send PUT /connectors/{name}/stop to place the connector in STOPPED
              state required by KIP-875 for offset alteration.
            uri: "{{ inputs.connect_url ?? vars.connect_url }}/connectors/{{
              inputs.connector_name ?? vars.connector_name }}/stop"
            method: PUT
            headers:
              Accept: "application/json"
              Content-Type: "application/json"
            options:
              allowFailed: true

          - id: alter_connector_offsets
            type: io.kestra.plugin.kafka.ConnectorAlterOffsets
            description: Overwrite committed Kafka Connect connector offsets with target
              remediation offsets via KIP-875 PATCH /connectors/{name}/offsets.
            connectUrl: "{{ inputs.connect_url ?? vars.connect_url }}"
            connectorName: "{{ inputs.connector_name ?? vars.connector_name }}"
            offsets: "{{ outputs.triage_poison_pill_and_compute_offsets.vars.target_offsets
              }}"

          - id: resume_connector
            type: io.kestra.plugin.kafka.ConnectorResume
            description: Resume connector and restart worker tasks from newly committed
              remediation offsets.
            connectUrl: "{{ inputs.connect_url ?? vars.connect_url }}"
            connectorName: "{{ inputs.connector_name ?? vars.connector_name }}"

          - id: verify_post_remediation_offsets
            type: io.kestra.plugin.kafka.ConnectorGetOffsets
            description: Post-remediation verification querying Connect REST API to confirm
              new offsets are committed.
            connectUrl: "{{ inputs.connect_url ?? vars.connect_url }}"
            connectorName: "{{ inputs.connector_name ?? vars.connector_name }}"

          - id: verify_post_remediation_status
            type: io.kestra.plugin.kafka.ConnectorGetStatus
            description: Query connector status post-resume to verify connectorState is
              RUNNING and tasks are active.
            connectUrl: "{{ inputs.connect_url ?? vars.connect_url }}"
            connectorName: "{{ inputs.connector_name ?? vars.connector_name }}"

          - id: assert_connector_running
            type: io.kestra.plugin.core.execution.Assert
            description: Assert that connector returned to RUNNING state after offset
              modification.
            conditions:
              - "{{ outputs.verify_post_remediation_status.connectorState ==
                'RUNNING' }}"
            errorMessage: "Connector failed to return to RUNNING state after offset
              alteration. Inspect worker logs."

          - id: log_remediation_success
            type: io.kestra.plugin.core.log.Log
            description: Log verified offset alterations and connector recovery telemetry.
            message: "Successfully applied KIP-875 offset remediation for {{
              inputs.connector_name ?? vars.connector_name }}. Verified
              committed offsets: {{
              outputs.verify_post_remediation_offsets.offsets | toJson }}."

  - id: notify_if_unhealthy
    type: io.kestra.plugin.core.flow.If
    description: Only post to Slack when the connector actually has failed tasks, so
      a healthy poll produces no notification noise.
    condition: "{{
      outputs.triage_poison_pill_and_compute_offsets.vars.has_failed_tasks }}"
    then:
      - id: dispatch_slack_notification
        type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
        description: Dispatch structured incident triage and remediation report to Slack.
        url: "{{ secret('SLACK_WEBHOOK_URL') }}"
        payload: >-
          {{ {
            "text": "*Kafka Connect Offset Remediation & Poison Pill Report*\n" ~ outputs.triage_poison_pill_and_compute_offsets.vars.summary_markdown ~ "\n\n*Execution Mode:* dry_run=" ~ inputs.dry_run ~ ", remediation_action=" ~ inputs.remediation_action ~ " (Execution: `" ~ execution.id ~ "`)"
          } | toJson }}

errors:
  - id: alert_on_flow_error
    type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
    description: Dispatch emergency alert to Slack if an unexpected execution error occurs.
    url: "{{ secret('SLACK_WEBHOOK_URL') }}"
    payload: >-
      {{ {
        "text": ":rotating_light: *Flow Execution Failed*: `kafka-connect-sink-poison-pill-offset-rewind-and-skip-gate` (Execution: `" ~ execution.id ~ "`). Failed to audit or remediate Kafka Connect connector `" ~ (inputs.connector_name ?? vars.connector_name) ~ "`. Check Connect worker reachability and network permissions."
      } | toJson }}