Kafka Consumer Group Dead-Letter Rebalance and Offset Rewind Guard

id: kafka-consumer-group-dead-letter-rebalance
namespace: company.streaming

description: |
  Inspect Kafka consumer group partition lag and rebalance state via Kafka AdminClient or HTTP Kafka REST Proxy,
  detect deadlocked partitions where consumer offset has not advanced while topic end offset increases,
  isolate poison pill offsets to a dead-letter quarantine topic, safely rewind or skip committed offsets
  with dry-run verification, and persist recovery audit telemetry into Kestra KV.

inputs:
  - id: bootstrap_servers
    type: STRING
    displayName: Kafka Bootstrap Servers
    description: Comma-separated list of Kafka broker host and port endpoints.
    defaults: "localhost:9092"

  - id: kafka_rest_proxy_url
    type: STRING
    displayName: Kafka REST Proxy URL
    description: Optional HTTP Kafka REST Proxy endpoint for cluster and consumer
      lag telemetry.
    defaults: ""

  - id: consumer_group_id
    type: STRING
    displayName: Consumer Group ID
    description: Target Apache Kafka consumer group identifier to audit and guard.
    defaults: "orders-payment-consumer"

  - id: monitored_topics
    type: ARRAY
    itemType: STRING
    displayName: Monitored Kafka Topics
    description: Specific topic names to evaluate for partition deadlocks. Leave
      empty to audit all assigned topics.
    defaults:
      - "orders-v1"

  - id: deadlock_lag_threshold
    type: INT
    displayName: Deadlock Partition Lag Threshold
    description: Minimum message lag (endOffset - currentOffset) required before
      flagging a partition as deadlocked.
    defaults: 50

  - id: rebalance_grace_seconds
    type: INT
    displayName: Rebalance Grace Period (Seconds)
    description: Grace duration permitted for consumer group rebalance coordination
      before flagging deadlock loops.
    defaults: 15

  - id: remediation_action
    type: SELECT
    displayName: Remediation Action
    description: Automated recovery action to execute when a poison pill offset or
      deadlocked partition is identified.
    values:
      - SKIP_POISON_RECORD
      - REWIND_OFFSET
      - AUDIT_ONLY
    defaults: SKIP_POISON_RECORD

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

  - id: quarantine_topic
    type: STRING
    displayName: Dead-Letter Quarantine Topic
    description: Destination Kafka topic where dead-letter poison pill metadata and
      alert records are published.
    defaults: "orders-dead-letter-quarantine"

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

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

triggers:
  - id: periodic_deadlock_guard
    type: io.kestra.plugin.core.trigger.Schedule
    description: Run automated Kafka consumer group deadlock and rebalance audit
      every 5 minutes.
    cron: "*/5 * * * *"
    disabled: true

  - id: on_demand_deadlock_guard_webhook
    type: io.kestra.plugin.core.trigger.Webhook
    description: Authenticated HTTP endpoint invoked by Kafka alerting probes,
      Prometheus Alertmanager, or incident runbooks.
    key: "{{ secret('KAFKA_DEADLOCK_GUARD_WEBHOOK_KEY') }}"

tasks:
  - id: validate_inputs
    type: io.kestra.plugin.core.execution.Assert
    description: Assert operational inputs satisfy threshold and sanity constraints.
    conditions:
      - "{{ inputs.consumer_group_id | length > 0 }}"
      - "{{ inputs.deadlock_lag_threshold >= 0 }}"
      - "{{ inputs.rebalance_grace_seconds >= 0 }}"
      - "{{ inputs.rewind_record_count >= 1 }}"
    errorMessage: "Invalid configuration inputs. Provide a valid consumer group ID,
      non-negative thresholds, and rewind count >= 1."

  - id: describe_consumer_group
    type: io.kestra.plugin.kafka.ConsumerGroupDescribe
    description: Query Kafka AdminClient for consumer group state, coordinator,
      active members, and partition offsets.
    properties:
      bootstrap.servers: "{{ inputs.bootstrap_servers }}"
    groupIds:
      - "{{ inputs.consumer_group_id }}"
    allowFailure: true

  - id: inspect_deadlock_and_poison_pill
    type: io.kestra.plugin.scripts.python.Script
    description: Analyze partition lag progression against historical KV snapshot,
      detect poison pill offsets, and construct remediation plan.
    containerImage: python:3.11-slim
    beforeCommands:
      - pip install -q requests
    env:
      BOOTSTRAP_SERVERS: "{{ inputs.bootstrap_servers }}"
      REST_PROXY_URL: "{{ inputs.kafka_rest_proxy_url }}"
      CONSUMER_GROUP_ID: "{{ inputs.consumer_group_id }}"
      MONITORED_TOPICS: "{{ inputs.monitored_topics | toJson }}"
      DEADLOCK_LAG_THRESHOLD: "{{ inputs.deadlock_lag_threshold }}"
      REMEDIATION_ACTION: "{{ inputs.remediation_action }}"
      REWIND_RECORD_COUNT: "{{ inputs.rewind_record_count }}"
      DRY_RUN: "{{ inputs.dry_run }}"
      QUARANTINE_TOPIC: "{{ inputs.quarantine_topic }}"
      PREVIOUS_STATE_JSON: "{{ kv('kafka_cg_tracker_' ~ (inputs.consumer_group_id |
        slugify), errorOnMissing=false) ?? '{}' }}"
      DESCRIBE_OUTPUT_JSON: "{{ (outputs.describe_consumer_group.groups ?? []) | toJson }}"
      EXECUTION_ID: "{{ execution.id }}"
    script: |
      import os
      import sys
      import json
      from datetime import datetime, timezone

      now_utc = datetime.now(timezone.utc)
      now_iso = now_utc.isoformat()

      bootstrap_servers = os.environ.get("BOOTSTRAP_SERVERS", "localhost:9092")
      rest_proxy_url = os.environ.get("REST_PROXY_URL", "").strip().rstrip("/")
      consumer_group_id = os.environ.get("CONSUMER_GROUP_ID", "orders-payment-consumer").strip()
      deadlock_lag_threshold = int(os.environ.get("DEADLOCK_LAG_THRESHOLD", "50"))
      remediation_action = os.environ.get("REMEDIATION_ACTION", "SKIP_POISON_RECORD")
      rewind_record_count = int(os.environ.get("REWIND_RECORD_COUNT", "1"))
      dry_run = os.environ.get("DRY_RUN", "true").lower() == "true"
      quarantine_topic = os.environ.get("QUARANTINE_TOPIC", "orders-dead-letter-quarantine").strip()
      execution_id = os.environ.get("EXECUTION_ID", "local-test")

      monitored_topics_raw = os.environ.get("MONITORED_TOPICS", "[]")
      try:
          monitored_topics = json.loads(monitored_topics_raw) if monitored_topics_raw.strip() else []
      except Exception:
          monitored_topics = []

      prev_state_raw = os.environ.get("PREVIOUS_STATE_JSON", "{}")
      try:
          prev_state = json.loads(prev_state_raw) if prev_state_raw.strip() else {}
      except Exception:
          prev_state = {}

      describe_raw = os.environ.get("DESCRIBE_OUTPUT_JSON", "[]")
      try:
          groups_described = json.loads(describe_raw) if describe_raw.strip() else []
      except Exception:
          groups_described = []

      # Optional HTTP Kafka REST Proxy diagnostic check
      rest_proxy_status = "NOT_CONFIGURED"
      if rest_proxy_url:
          try:
              import requests
              headers = {"Accept": "application/vnd.kafka.v2+json, application/json"}
              resp = requests.get(f"{rest_proxy_url}/consumers/{consumer_group_id}", headers=headers, timeout=5)
              if resp.status_code == 200:
                  rest_proxy_status = "CONNECTED"
                  print(f"REST Proxy connected at {rest_proxy_url} for group {consumer_group_id}.")
              else:
                  rest_proxy_status = f"HTTP_{resp.status_code}"
                  print(f"REST Proxy returned HTTP {resp.status_code} for group {consumer_group_id}.")
          except Exception as exc:
              rest_proxy_status = f"ERROR: {str(exc)}"
              print(f"Note: REST Proxy connection bypassed ({exc}). Using AdminClient metadata.")

      # Locate target group from describe output
      target_group = None
      for g in groups_described:
          if g.get("groupId") == consumer_group_id:
              target_group = g
              break

      if not target_group and groups_described:
          target_group = groups_described[0]

      group_state = target_group.get("state", "UNKNOWN") if target_group else "UNKNOWN"
      members = target_group.get("members", []) if target_group else []
      offsets = target_group.get("offsets", []) if target_group else []

      is_rebalancing = group_state in ["PreparingRebalance", "CompletingRebalance"]

      # Map active partition assignments to consumer members
      partition_assignments = {}
      for m in members:
          m_id = m.get("memberId", "unknown")
          c_id = m.get("clientId", "unknown")
          for ap in m.get("assignedPartitions", []):
              t = ap.get("topic")
              p = ap.get("partition")
              if t is not None and p is not None:
                  partition_assignments[f"{t}-{p}"] = {"member_id": m_id, "client_id": c_id}

      total_partitions_audited = 0
      deadlocked_partitions = []
      dead_letter_records = []
      alter_offsets = []
      new_state_snapshot = {}

      for item in offsets:
          topic = item.get("topic")
          partition = item.get("partition")
          if topic is None or partition is None:
              continue

          if monitored_topics and topic not in monitored_topics:
              continue

          total_partitions_audited += 1
          pk = f"{topic}-{partition}"
          curr_offset = item.get("currentOffset", 0)
          curr_end = item.get("endOffset", 0)
          curr_lag = item.get("lag", max(0, curr_end - curr_offset))

          assignment = partition_assignments.get(pk, {"member_id": "unassigned", "client_id": "unassigned"})

          # Capture in current state snapshot for KV storage
          new_state_snapshot[pk] = {
              "current_offset": curr_offset,
              "end_offset": curr_end,
              "lag": curr_lag,
              "timestamp": now_iso
          }

          is_deadlocked = False
          detection_reason = ""

          # Differential comparison against previous execution sample
          if pk in prev_state:
              prev_info = prev_state[pk]
              prev_offset = prev_info.get("current_offset")
              prev_end = prev_info.get("end_offset")

              # Offset has not advanced while topic end offset has grown past lag threshold
              if curr_offset == prev_offset and curr_end > prev_end and curr_lag >= deadlock_lag_threshold:
                  is_deadlocked = True
                  detection_reason = (
                      f"Committed offset stalled at {curr_offset} while topic end offset increased "
                      f"from {prev_end} to {curr_end} (accumulated lag: {curr_lag} >= {deadlock_lag_threshold})."
                  )
              elif curr_offset == prev_offset and curr_lag >= deadlock_lag_threshold and is_rebalancing:
                  is_deadlocked = True
                  detection_reason = (
                      f"Committed offset pinned at {curr_offset} with lag {curr_lag} during active rebalance cycle."
                  )
          else:
              # Baseline snapshot: flag unassigned or stalled partitions exceeding deadlock lag threshold
              if curr_lag >= deadlock_lag_threshold and curr_offset < curr_end:
                  if is_rebalancing or assignment.get("member_id") == "unassigned":
                      is_deadlocked = True
                      detection_reason = (
                          f"Initial sample detected unassigned or rebalancing partition with critical lag "
                          f"({curr_lag} >= {deadlock_lag_threshold}) stalled at offset {curr_offset}."
                      )

          if is_deadlocked:
              poison_offset = curr_offset

              if remediation_action == "SKIP_POISON_RECORD":
                  target_offset = poison_offset + 1
                  action_desc = f"Skip poisoned record at offset {poison_offset} and advance committed offset to {target_offset}"
              elif remediation_action == "REWIND_OFFSET":
                  target_offset = max(0, poison_offset - rewind_record_count)
                  action_desc = f"Rewind committed offset from {poison_offset} to {target_offset} (rewind {rewind_record_count} messages)"
              else:  # AUDIT_ONLY
                  target_offset = poison_offset
                  action_desc = f"Audit only. Poison pill identified at offset {poison_offset}. Offsets unchanged"

              deadlocked_info = {
                  "topic": topic,
                  "partition": partition,
                  "poison_offset": poison_offset,
                  "current_offset": curr_offset,
                  "end_offset": curr_end,
                  "lag": curr_lag,
                  "target_offset": target_offset,
                  "remediation_action": remediation_action,
                  "action_description": action_desc,
                  "detection_reason": detection_reason,
                  "member_id": assignment.get("member_id"),
                  "client_id": assignment.get("client_id")
              }
              deadlocked_partitions.append(deadlocked_info)

              # Construct structured dead-letter isolation payload
              dlq_record = {
                  "event_type": "KAFKA_POISON_PILL_DEADLOCK",
                  "consumer_group_id": consumer_group_id,
                  "topic": topic,
                  "partition": partition,
                  "poison_offset": poison_offset,
                  "target_offset": target_offset,
                  "lag": curr_lag,
                  "end_offset": curr_end,
                  "group_state": group_state,
                  "remediation_action": remediation_action,
                  "assigned_member": assignment.get("member_id"),
                  "assigned_client": assignment.get("client_id"),
                  "detected_at": now_iso,
                  "quarantine_topic": quarantine_topic,
                  "execution_id": execution_id
              }
              dead_letter_records.append(dlq_record)

              if remediation_action in ["SKIP_POISON_RECORD", "REWIND_OFFSET"]:
                  alter_offsets.append({
                      "topic": topic,
                      "partition": partition,
                      "offset": target_offset
                  })

      has_deadlock = len(deadlocked_partitions) > 0
      overall_status = "CRITICAL" if has_deadlock else ("WARNING" if is_rebalancing else "HEALTHY")

      summary_text = (
          f"Kafka Deadlock Guard: {overall_status} | Group '{consumer_group_id}' (state: {group_state}). "
          f"Audited {total_partitions_audited} partition(s). Deadlocked partitions: {len(deadlocked_partitions)}. "
          f"Rebalance in progress: {is_rebalancing}. Remediation: {remediation_action} (Dry run: {dry_run})."
      )
      print(summary_text)

      audit_manifest = {
          "overall_status": overall_status,
          "consumer_group_id": consumer_group_id,
          "group_state": group_state,
          "is_rebalancing": is_rebalancing,
          "rest_proxy_status": rest_proxy_status,
          "total_partitions_audited": total_partitions_audited,
          "deadlocked_partitions_count": len(deadlocked_partitions),
          "deadlocked_partitions": deadlocked_partitions,
          "alter_offsets": alter_offsets,
          "remediation_action": remediation_action,
          "dry_run": dry_run,
          "quarantine_topic": quarantine_topic,
          "evaluated_at": now_iso,
          "execution_id": execution_id,
          "summary": summary_text
      }

      print("::" + json.dumps({
          "outputs": {
              "overall_status": overall_status,
              "has_deadlock": has_deadlock,
              "deadlock_count": len(deadlocked_partitions),
              "group_state": group_state,
              "is_rebalancing": is_rebalancing,
              "rest_proxy_status": rest_proxy_status,
              "total_partitions_audited": total_partitions_audited,
              "deadlocked_partitions": deadlocked_partitions,
              "dead_letter_records": dead_letter_records,
              "alter_offsets": alter_offsets,
              "new_state_snapshot": new_state_snapshot,
              "audit_manifest": audit_manifest,
              "summary": summary_text
          }
      }) + "::")

  - id: save_state_tracker_kv
    type: io.kestra.plugin.core.kv.Set
    description: Persist updated partition offset snapshot in Kestra KV store for
      sequential lag differential detection.
    key: "kafka_cg_tracker_{{ inputs.consumer_group_id | slugify }}"
    value: "{{ outputs.inspect_deadlock_and_poison_pill.vars.new_state_snapshot |
      toJson }}"
    kvType: JSON

  - id: evaluate_deadlock_findings
    type: io.kestra.plugin.core.flow.If
    description: Branch execution based on whether deadlocked partitions or poison
      pill records were detected.
    condition: "{{ outputs.inspect_deadlock_and_poison_pill.vars.has_deadlock == true }}"
    then:
      - id: route_quarantine_dlq
        type: io.kestra.plugin.core.flow.If
        description: Route poison pill diagnostics to Kafka dead-letter quarantine topic
          when enabled.
        condition: "{{ inputs.publish_to_dlq_topic and (inputs.quarantine_topic | length
          > 0) }}"
        then:
          - id: publish_dead_letters
            type: io.kestra.plugin.core.flow.Loop
            description: Publish dead-letter notification record for each deadlocked
              partition to Kafka quarantine topic.
            values: "{{ outputs.inspect_deadlock_and_poison_pill.vars.dead_letter_records
              }}"
            tasks:
              - id: produce_to_quarantine_topic
                type: io.kestra.plugin.kafka.Produce
                description: Publish dead-letter message with bad offset and partition details.
                properties:
                  bootstrap.servers: "{{ inputs.bootstrap_servers }}"
                topic: "{{ inputs.quarantine_topic }}"
                keySerializer: STRING
                valueSerializer: JSON
                from:
                  key: "{{ item.value.topic ?? (item.value | fromJson).topic }}-{{
                    item.value.partition ?? (item.value | fromJson).partition
                    }}-{{ item.value.poison_offset ?? (item.value |
                    fromJson).poison_offset }}"
                  value: "{{ item.value | toJson }}"

      - id: remediation_gate
        type: io.kestra.plugin.core.flow.If
        description: Apply consumer offset rewind or poison record skip when dry_run is
          disabled.
        condition: "{{ not inputs.dry_run and
          (outputs.inspect_deadlock_and_poison_pill.vars.alter_offsets | length
          > 0) }}"
        then:
          - id: pause_for_offset_alter_approval
            type: io.kestra.plugin.core.flow.Pause
            description: Pause execution for operator verification before modifying
              committed consumer group offsets. Resume to apply or kill to
              abort.

          - id: alter_consumer_group_offsets
            type: io.kestra.plugin.kafka.ConsumerGroupAlterOffsets
            description: Safely adjust Kafka consumer group committed offsets to target
              recovery offsets.
            properties:
              bootstrap.servers: "{{ inputs.bootstrap_servers }}"
            groupId: "{{ inputs.consumer_group_id }}"
            offsets: "{{ outputs.inspect_deadlock_and_poison_pill.vars.alter_offsets }}"

          - id: verify_altered_offsets
            type: io.kestra.plugin.kafka.ConsumerGroupDescribe
            description: Post-remediation verification verifying consumer group committed
              offsets match altered targets.
            properties:
              bootstrap.servers: "{{ inputs.bootstrap_servers }}"
            groupIds:
              - "{{ inputs.consumer_group_id }}"

          - id: log_remediation_success
            type: io.kestra.plugin.core.log.Log
            description: Log verified offset alterations and recovery status.
            message: "Successfully applied offset alterations for {{
              inputs.consumer_group_id }}. Altered offsets: {{
              outputs.inspect_deadlock_and_poison_pill.vars.alter_offsets |
              toJson }}."

        else:
          - id: log_dry_run_remediation
            type: io.kestra.plugin.core.log.Log
            description: Log planned alterations in dry-run mode without modifying cluster
              state.
            message: "Dry run mode active. No offsets altered. Planned alterations: {{
              outputs.inspect_deadlock_and_poison_pill.vars.alter_offsets |
              toJson }}."

      - id: save_recovery_audit_kv
        type: io.kestra.plugin.core.kv.Set
        description: Record complete recovery audit manifest and incident forensics into
          Kestra KV store.
        key: "kafka_deadlock_audit_{{ inputs.consumer_group_id | slugify }}"
        value: "{{ outputs.inspect_deadlock_and_poison_pill.vars.audit_manifest | toJson
          }}"
        kvType: JSON

      - id: notify_slack_deadlock_alert
        type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
        description: Deliver prioritized incident notification to Slack with affected
          partitions, bad offsets, and remediation status.
        url: "{{ secret('SLACK_WEBHOOK_URL') }}"
        payload: |
          {
            "attachments": [
              {
                "color": "danger",
                "title": ":warning: Kafka Consumer Group Deadlock & Poison Pill Alert",
                "text": "*Consumer Group:* `{{ inputs.consumer_group_id }}`\n*Group State:* `{{ outputs.inspect_deadlock_and_poison_pill.vars.group_state }}`\n*Deadlocked Partitions:* {{ outputs.inspect_deadlock_and_poison_pill.vars.deadlock_count }}\n*Remediation Action:* `{{ inputs.remediation_action }}` (Dry Run: `{{ inputs.dry_run }}`)\n*Quarantine DLQ Topic:* `{{ inputs.quarantine_topic }}`\n*Execution ID:* `{{ execution.id }}`\n\n*Summary:* {{ outputs.inspect_deadlock_and_poison_pill.vars.summary }}"
              }
            ]
          }

    else:
      - id: log_healthy_state
        type: io.kestra.plugin.core.log.Log
        description: Log healthy consumer group status when no deadlocks or poison pill
          records are detected.
        message: "Kafka consumer group {{ inputs.consumer_group_id }} is healthy. No
          partition deadlocks or poison pills detected across {{
          outputs.inspect_deadlock_and_poison_pill.vars.total_partitions_audited
          }} partition(s)."

errors:
  - id: alert_failure
    type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
    description: Deliver critical alert if Kafka deadlock guard workflow encounters
      unhandled execution errors.
    url: "{{ secret('SLACK_WEBHOOK_URL') }}"
    payload: |
      {
        "attachments": [
          {
            "color": "danger",
            "title": ":rotating_light: Kafka Consumer Group Deadlock Guard Execution Failed",
            "text": "Failed execution `{{ execution.id }}` while inspecting consumer group `{{ inputs.consumer_group_id }}` on `{{ inputs.bootstrap_servers }}`.\nCheck broker connectivity, cluster ACLs, or worker logs."
          }
        ]
      }