Kafka Topic Dynamic Min In-Sync Replicas and Unclean Leader Election Durability Guard

id: kafka-topic-insync-replicas-durability-guard
namespace: company.streaming

description: |
  Audit Apache Kafka topic min.insync.replicas and unclean.leader.election.enable configurations,
  track persistent durability violations across runs in Kestra KV, gate configuration changes
  behind an operator approval Pause, safely apply durability hardening policies via Kafka AdminClient
  TopicUpdate, and verify post-remediation cluster state.

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

  - id: monitored_topics
    type: ARRAY
    itemType: STRING
    displayName: Monitored Topics
    description: Specific topics to audit for durability violations. Leave empty to
      automatically audit all non-internal topics.
    defaults: []

  - id: min_insync_replicas_floor
    type: INT
    displayName: Minimum In-Sync Replicas Floor
    description: Minimum in-sync replicas floor required for durable message
      acknowledgment. Must be at least 2 for multi-replica topics.
    defaults: 2

  - id: enforce_clean_leader_election
    type: BOOL
    displayName: Enforce Clean Leader Election
    description: When true, verifies that unclean.leader.election.enable is false to
      prevent out-of-sync replicas from becoming leader.
    defaults: true

  - id: min_consecutive_violation_runs
    type: INT
    displayName: Minimum Consecutive Violation Runs
    description: Number of consecutive audit runs a topic must remain in violation
      before automated remediation is eligible.
    defaults: 2

  - id: auto_remediate
    type: BOOL
    displayName: Enable Auto-Remediation
    description: When true and dry_run is false, enables the operator-approved
      write-path to update topic configurations.
    defaults: false

  - id: dry_run
    type: BOOL
    displayName: Dry Run Mode
    description: When true, audits topic configurations and calculates remediation
      plans without altering live Kafka topics.
    defaults: true

triggers:
  - id: periodic_durability_audit
    type: io.kestra.plugin.core.trigger.Schedule
    description: Run automated Kafka topic durability audit every 30 minutes.
    cron: "*/30 * * * *"
    disabled: true

  - id: on_demand_durability_webhook
    type: io.kestra.plugin.core.trigger.Webhook
    description: Authenticated HTTP endpoint invoked by Kafka monitoring alerts,
      Prometheus Alertmanager, or SRE runbooks.
    key: "{{ secret('KAFKA_DURABILITY_GUARD_WEBHOOK_KEY') }}"

tasks:
  - id: validate_inputs
    type: io.kestra.plugin.core.execution.Assert
    description: Assert operational inputs meet sanity and safety boundaries.
    conditions:
      - "{{ inputs.bootstrap_servers | length > 0 }}"
      - "{{ inputs.min_insync_replicas_floor >= 1 }}"
      - "{{ inputs.min_consecutive_violation_runs >= 1 }}"
    errorMessage: "Invalid input parameters. Provide non-empty bootstrap servers,
      min_insync_replicas_floor >= 1, and min_consecutive_violation_runs >= 1."

  - id: list_topics
    type: io.kestra.plugin.kafka.TopicList
    description: Query Kafka cluster for all existing topic names using AdminClient
      listTopics.
    properties:
      bootstrap.servers: "{{ inputs.bootstrap_servers }}"

  - id: filter_target_topics
    type: io.kestra.plugin.scripts.python.Script
    description: Filter out internal Kafka topics and resolve the active list of
      user topics to audit.
    containerImage: python:3.11-slim
    env:
      ALL_TOPICS: "{{ outputs.list_topics.topics | toJson }}"
      MONITORED_TOPICS: "{{ inputs.monitored_topics | toJson }}"
    script: |
      import os
      import sys
      import json

      all_topics_raw = os.getenv("ALL_TOPICS", "[]")
      monitored_topics_raw = os.getenv("MONITORED_TOPICS", "[]")

      try:
          all_topics = json.loads(all_topics_raw) if all_topics_raw.strip() else []
          monitored_topics = json.loads(monitored_topics_raw) if monitored_topics_raw.strip() else []
      except Exception as exc:
          print(f"Error parsing topic lists: {exc}", file=sys.stderr)
          sys.exit(1)

      internal_prefixes = ("_", ".", "__")
      all_user_topics = [t for t in all_topics if not any(t.startswith(p) for p in internal_prefixes)]

      if monitored_topics and len(monitored_topics) > 0:
          target_topics = [t for t in monitored_topics if t in all_topics]
          missing_topics = [t for t in monitored_topics if t not in all_topics]
          if missing_topics:
              print(f"Warning: Requested monitored topics not found on cluster: {missing_topics}")
      else:
          target_topics = all_user_topics

      print(f"Discovered {len(all_topics)} total topics on cluster. Selected {len(target_topics)} target user topics for durability audit: {target_topics}")

      print("::" + json.dumps({
          "outputs": {
              "target_topics": target_topics,
              "total_cluster_topics": len(all_topics),
              "user_topic_count": len(target_topics)
          }
      }) + "::")

  - id: describe_topics
    type: io.kestra.plugin.core.flow.Loop
    description: Describe live configuration and partition layout for each audited
      Kafka topic.
    values: "{{ outputs.filter_target_topics.vars.target_topics }}"
    concurrencyLimit: 4
    outputs:
      - id: live_topic
        type: STRING
        value: "{{ {'topic': item.value, 'replicationFactor':
          outputs.describe.replicationFactor, 'partitionCount':
          outputs.describe.partitionCount, 'configs': outputs.describe.configs,
          'partitions': outputs.describe.partitions} | toJson }}"
    tasks:
      - id: describe
        type: io.kestra.plugin.kafka.TopicDescribe
        topic: "{{ item.value }}"
        properties:
          bootstrap.servers: "{{ inputs.bootstrap_servers }}"

  - id: get_previous_durability_state
    type: io.kestra.plugin.core.kv.Get
    description: Retrieve historical topic durability compliance state from Kestra KV store.
    key: "kafka_topic_durability_guard_state"
    errorOnMissing: false

  - id: evaluate_durability_policy
    type: io.kestra.plugin.scripts.python.Script
    description: Compare live topic configs against durability standards, compute
      remediation plans, and track multi-run violation streaks.
    containerImage: python:3.11-slim
    env:
      LOOP_OUTPUTS: "{{ (outputs.describe_topics.outputs ?? []) | toJson }}"
      PREVIOUS_STATE: "{{ (outputs.get_previous_durability_state.value ?? {}) | toJson }}"
      MIN_ISR_FLOOR: "{{ inputs.min_insync_replicas_floor }}"
      ENFORCE_CLEAN_ELECTION: "{{ inputs.enforce_clean_leader_election }}"
      MIN_CONSECUTIVE_RUNS: "{{ inputs.min_consecutive_violation_runs }}"
      AUTO_REMEDIATE: "{{ inputs.auto_remediate }}"
      DRY_RUN: "{{ inputs.dry_run }}"
      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()

      loop_outputs_raw = os.getenv("LOOP_OUTPUTS", "[]")
      previous_state_raw = os.getenv("PREVIOUS_STATE", "{}")
      min_isr_floor = int(os.getenv("MIN_ISR_FLOOR", "2"))
      enforce_clean_election = os.getenv("ENFORCE_CLEAN_ELECTION", "true").lower() == "true"
      min_consecutive_runs = int(os.getenv("MIN_CONSECUTIVE_RUNS", "2"))
      auto_remediate = os.getenv("AUTO_REMEDIATE", "false").lower() == "true"
      dry_run = os.getenv("DRY_RUN", "true").lower() == "true"
      execution_id = os.getenv("EXECUTION_ID", "local-execution")

      try:
          loop_outputs = json.loads(loop_outputs_raw) if loop_outputs_raw.strip() else []
      except Exception as exc:
          print(f"Error parsing loop outputs: {exc}", file=sys.stderr)
          sys.exit(1)

      try:
          prev_state = json.loads(previous_state_raw) if previous_state_raw.strip() else {}
          if not isinstance(prev_state, dict):
              prev_state = {}
      except Exception:
          prev_state = {}

      prev_topic_states = prev_state.get("topics", {})

      topics_data = []
      for entry in loop_outputs:
          lt = entry.get("outputs", {}).get("live_topic")
          if lt:
              if isinstance(lt, str):
                  try:
                      topics_data.append(json.loads(lt))
                  except Exception:
                      pass
              elif isinstance(lt, dict):
                  topics_data.append(lt)

      topic_audit_results = []
      remediation_topics = []
      topics_to_remediate_names = []
      new_topic_states = {}

      total_topics = len(topics_data)
      violation_count = 0
      compliant_count = 0

      for topic_entry in topics_data:
          topic_name = topic_entry.get("topic", "")
          configs = topic_entry.get("configs", {}) or {}
          rf = int(topic_entry.get("replicationFactor", 1))
          partition_count = int(topic_entry.get("partitionCount", 0))

          live_min_isr_str = configs.get("min.insync.replicas", "1")
          try:
              live_min_isr = int(live_min_isr_str)
          except (ValueError, TypeError):
              live_min_isr = 1

          unclean_raw = str(configs.get("unclean.leader.election.enable", "false")).lower()
          live_unclean_election = unclean_raw == "true"

          # Determine target min.insync.replicas
          # Safe ceiling: cannot exceed replication factor (otherwise writes fail unconditionally)
          target_min_isr = min(min_isr_floor, rf)

          violations = []
          desired_configs = {}

          # Rule 1: min.insync.replicas check
          # Flag if topic has RF >= min_isr_floor but min.insync.replicas is below target
          if rf >= min_isr_floor and live_min_isr < target_min_isr:
              violations.append(f"min.insync.replicas is {live_min_isr} (expected >= {target_min_isr})")
              desired_configs["min.insync.replicas"] = str(target_min_isr)
          elif rf < min_isr_floor and live_min_isr < rf:
              violations.append(f"min.insync.replicas is {live_min_isr} for RF={rf} (expected {rf})")
              desired_configs["min.insync.replicas"] = str(rf)

          # Rule 2: unclean leader election check
          if enforce_clean_election and live_unclean_election:
              violations.append("unclean.leader.election.enable is true (risks partition divergence and message loss)")
              desired_configs["unclean.leader.election.enable"] = "false"

          has_violation = len(violations) > 0
          prev_info = prev_topic_states.get(topic_name, {})
          prev_consecutive = int(prev_info.get("consecutive_violation_runs", 0))

          if has_violation:
              consecutive_runs = prev_consecutive + 1
              violation_count += 1
          else:
              consecutive_runs = 0
              compliant_count += 1

          topic_remediation_eligible = has_violation and (consecutive_runs >= min_consecutive_runs or live_unclean_election)

          if has_violation and desired_configs:
              remediation_topics.append({
                  "topic": topic_name,
                  "configs": desired_configs
              })
              topics_to_remediate_names.append(topic_name)

          topic_audit_results.append({
              "topic": topic_name,
              "replication_factor": rf,
              "partition_count": partition_count,
              "live_min_isr": live_min_isr,
              "live_unclean_election": live_unclean_election,
              "has_violation": has_violation,
              "violations": violations,
              "consecutive_violation_runs": consecutive_runs,
              "remediation_eligible": topic_remediation_eligible,
              "desired_configs": desired_configs
          })

          new_topic_states[topic_name] = {
              "consecutive_violation_runs": consecutive_runs,
              "last_evaluated_min_isr": live_min_isr,
              "last_evaluated_unclean": live_unclean_election,
              "last_evaluated_at": now_iso
          }

      has_violations = violation_count > 0
      remediation_eligible = any(t["remediation_eligible"] for t in topic_audit_results)

      if any(t["live_unclean_election"] for t in topic_audit_results):
          overall_status = "CRITICAL"
      elif has_violations:
          overall_status = "WARNING"
      else:
          overall_status = "HEALTHY"

      summary_text = (
          f"Kafka Durability Guard: {overall_status} | Audited {total_topics} topics. "
          f"{compliant_count} compliant, {violation_count} in violation. "
          f"Remediation eligible: {remediation_eligible} (Dry Run: {dry_run}, Auto Remediate: {auto_remediate}). "
          f"Actionable topics: {topics_to_remediate_names}"
      )
      print(summary_text)

      new_state_snapshot = {
          "last_execution_id": execution_id,
          "last_evaluated_at": now_iso,
          "overall_status": overall_status,
          "total_topics": total_topics,
          "compliant_count": compliant_count,
          "violation_count": violation_count,
          "topics": new_topic_states
      }

      audit_manifest = {
          "execution_id": execution_id,
          "evaluated_at": now_iso,
          "overall_status": overall_status,
          "total_topics": total_topics,
          "compliant_count": compliant_count,
          "violation_count": violation_count,
          "min_isr_floor": min_isr_floor,
          "enforce_clean_election": enforce_clean_election,
          "min_consecutive_runs": min_consecutive_runs,
          "dry_run": dry_run,
          "auto_remediate": auto_remediate,
          "remediation_eligible": remediation_eligible,
          "remediation_topics": remediation_topics,
          "topics_to_remediate_names": topics_to_remediate_names,
          "topic_audit_results": topic_audit_results,
          "summary": summary_text
      }

      print("::" + json.dumps({
          "outputs": {
              "overall_status": overall_status,
              "has_violations": has_violations,
              "remediation_eligible": remediation_eligible,
              "violation_count": violation_count,
              "compliant_count": compliant_count,
              "total_topics": total_topics,
              "remediation_topics": remediation_topics,
              "topics_to_remediate_names": topics_to_remediate_names,
              "new_state_snapshot": new_state_snapshot,
              "audit_manifest": audit_manifest,
              "summary_text": summary_text
          }
      }) + "::")

  - id: save_durability_state
    type: io.kestra.plugin.core.kv.Set
    description: Persist updated topic compliance telemetry and consecutive
      violation counts in Kestra KV store.
    key: "kafka_topic_durability_guard_state"
    value: "{{ outputs.evaluate_durability_policy.vars.new_state_snapshot | toJson }}"
    kvType: JSON

  - id: evaluate_durability_findings
    type: io.kestra.plugin.core.flow.If
    description: Branch workflow execution based on whether durability policy
      violations were detected.
    condition: "{{ outputs.evaluate_durability_policy.vars.has_violations == true }}"
    then:
      - id: remediation_gate
        type: io.kestra.plugin.core.flow.If
        description: Execute automated topic configuration hardening when auto_remediate
          is true, dry_run is false, and violation threshold is reached.
        condition: "{{ not inputs.dry_run and inputs.auto_remediate and
          outputs.evaluate_durability_policy.vars.remediation_eligible == true
          }}"
        then:
          - id: pause_for_remediation_approval
            type: io.kestra.plugin.core.flow.Pause
            description: Pause workflow for operator review before modifying Kafka topic
              configurations. Resume to apply or kill to abort.

          - id: remediate_topics
            type: io.kestra.plugin.core.flow.Loop
            description: Apply hardened durability configuration to each eligible Kafka
              topic using TopicUpdate.
            values: "{{ outputs.evaluate_durability_policy.vars.remediation_topics }}"
            concurrencyLimit: 2
            tasks:
              - id: update_topic_configs
                type: io.kestra.plugin.kafka.TopicUpdate
                topic: "{{ fromJson(item.value).topic }}"
                configs: "{{ fromJson(item.value).configs }}"
                properties:
                  bootstrap.servers: "{{ inputs.bootstrap_servers }}"

          - id: verify_topics
            type: io.kestra.plugin.core.flow.Loop
            description: Post-remediation verification querying TopicDescribe for each
              updated topic.
            values: "{{ outputs.evaluate_durability_policy.vars.topics_to_remediate_names
              }}"
            concurrencyLimit: 2
            outputs:
              - id: verified_topic
                type: STRING
                value: "{{ {'topic': item.value, 'configs': outputs.recheck_topic.configs} |
                  toJson }}"
            tasks:
              - id: recheck_topic
                type: io.kestra.plugin.kafka.TopicDescribe
                topic: "{{ item.value }}"
                properties:
                  bootstrap.servers: "{{ inputs.bootstrap_servers }}"

          - id: verify_durability_outcome
            type: io.kestra.plugin.scripts.python.Script
            description: Verify that applied durability configs are active on Kafka and
              match target policies.
            containerImage: python:3.11-slim
            env:
              VERIFY_OUTPUTS: "{{ (outputs.verify_topics.outputs ?? []) | toJson }}"
              EXPECTED_REMEDIATIONS: "{{
                outputs.evaluate_durability_policy.vars.remediation_topics |
                toJson }}"
            script: |
              import os
              import sys
              import json

              verify_raw = os.getenv("VERIFY_OUTPUTS", "[]")
              expected_raw = os.getenv("EXPECTED_REMEDIATIONS", "[]")

              try:
                  verify_entries = json.loads(verify_raw) if verify_raw.strip() else []
                  expected_remediations = json.loads(expected_raw) if expected_raw.strip() else []
              except Exception as exc:
                  print(f"Failed to parse verification input JSON: {exc}", file=sys.stderr)
                  sys.exit(1)

              verified_topics_map = {}
              for entry in verify_entries:
                  vt = entry.get("outputs", {}).get("verified_topic")
                  if vt:
                      if isinstance(vt, str):
                          try:
                              data = json.loads(vt)
                              verified_topics_map[data.get("topic")] = data.get("configs", {})
                          except Exception:
                              pass
                      elif isinstance(vt, dict):
                          verified_topics_map[vt.get("topic")] = vt.get("configs", {})

              mismatches = []
              for item in expected_remediations:
                  t = item.get("topic")
                  target_configs = item.get("configs", {})
                  actual_configs = verified_topics_map.get(t, {})

                  for k, expected_v in target_configs.items():
                      actual_v = actual_configs.get(k)
                      if str(actual_v) != str(expected_v):
                          mismatches.append(f"Topic '{t}': key '{k}' expected '{expected_v}', got '{actual_v}'")

              if mismatches:
                  print("Post-remediation verification FAILED with mismatches:", file=sys.stderr)
                  for m in mismatches:
                      print(f"  - {m}", file=sys.stderr)
                  sys.exit(1)

              print(f"Post-remediation verification SUCCESSFUL. Confirmed {len(expected_remediations)} topic(s) updated on cluster.")

          - id: log_remediation_success
            type: io.kestra.plugin.core.log.Log
            description: Log successful durability configuration remediation.
            message: "Successfully applied and verified durability configurations on Kafka
              cluster for {{
              outputs.evaluate_durability_policy.vars.remediation_topics |
              length }} topic(s)."

        else:
          - id: log_dry_run_remediation
            type: io.kestra.plugin.core.log.Log
            description: Log proposed remediation plan in dry-run mode or pending
              consecutive violation threshold.
            message: "Remediation bypassed (dry_run: {{ inputs.dry_run }}, auto_remediate:
              {{ inputs.auto_remediate }}, eligible: {{
              outputs.evaluate_durability_policy.vars.remediation_eligible }}).
              Target topics requiring hardening: {{
              outputs.evaluate_durability_policy.vars.topics_to_remediate_names
              | join(', ') }}."

      - id: save_recovery_audit_kv
        type: io.kestra.plugin.core.kv.Set
        description: Record full audit manifest and forensics into Kestra KV store.
        key: "kafka_topic_durability_audit_report"
        value: "{{ outputs.evaluate_durability_policy.vars.audit_manifest | toJson }}"
        kvType: JSON

      - id: notify_slack_durability_alert
        type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
        description: Deliver actionable incident notification to Slack with durability
          violation details.
        url: "{{ secret('SLACK_WEBHOOK_URL') }}"
        payload: |
          {
            "attachments": [
              {
                "color": "{{ outputs.evaluate_durability_policy.vars.overall_status == 'CRITICAL' ? '#E01E5A' : '#ECB22E' }}",
                "title": ":warning: Kafka Topic In-Sync Replicas & Durability Guard Alert",
                "text": "{{ outputs.evaluate_durability_policy.vars.summary_text }}",
                "fields": [
                  {
                    "title": "Overall Status",
                    "value": "*{{ outputs.evaluate_durability_policy.vars.overall_status }}*",
                    "short": true
                  },
                  {
                    "title": "Violations / Total",
                    "value": "{{ outputs.evaluate_durability_policy.vars.violation_count }} / {{ outputs.evaluate_durability_policy.vars.total_topics }}",
                    "short": true
                  },
                  {
                    "title": "Min ISR Floor",
                    "value": "`{{ inputs.min_insync_replicas_floor }}`",
                    "short": true
                  },
                  {
                    "title": "Clean Election Enforced",
                    "value": "`{{ inputs.enforce_clean_leader_election }}`",
                    "short": true
                  },
                  {
                    "title": "Remediation Eligible",
                    "value": "`{{ outputs.evaluate_durability_policy.vars.remediation_eligible }}`",
                    "short": true
                  },
                  {
                    "title": "Mode",
                    "value": "Dry Run: `{{ inputs.dry_run }}` | Auto Remediate: `{{ inputs.auto_remediate }}`",
                    "short": true
                  }
                ]
              }
            ]
          }

    else:
      - id: log_compliant_state
        type: io.kestra.plugin.core.log.Log
        description: Log compliant topic status when all audited topics meet durability
          policy standards.
        message: "All {{ outputs.evaluate_durability_policy.vars.compliant_count }}
          audited Kafka topic(s) comply with durability standards
          (min.insync.replicas >= {{ inputs.min_insync_replicas_floor }},
          unclean.leader.election.enable: false)."

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 during durability audit.
    url: "{{ secret('SLACK_WEBHOOK_URL') }}"
    payload: |
      {
        "attachments": [
          {
            "color": "#E01E5A",
            "title": ":rotating_light: Kafka Topic Durability Guard Execution Failed",
            "text": "Failed execution `{{ execution.id }}` while auditing Kafka topic durability on `{{ inputs.bootstrap_servers }}`.\nCheck Kafka broker connectivity, cluster credentials, or worker logs."
          }
        ]
      }