Kafka Producer Compression Ratio Sentinel

id: kafka-producer-compression-ratio-sentinel
namespace: company.team
description: |
  Automated Apache Kafka topic compression efficiency auditor. Samples message batches across high-throughput topics,
  calculates payload compressibility, and alerts when uncompressed streams cause network and broker storage bloat.

concurrency:
  limit: 1

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

  - id: topic_name
    type: STRING
    displayName: Kafka Topic
    defaults: "telemetry-events-raw"
    description: "Target Apache Kafka topic to sample and analyze for batch
      compression efficiency."

  - id: group_id
    type: STRING
    displayName: Consumer Group ID
    defaults: "kestra-compression-sentinel"
    description: "Consumer group ID used for non-intrusive metadata and payload sampling."

  - id: min_compression_ratio_threshold
    type: FLOAT
    displayName: Compression Threshold
    defaults: 1.5
    description: "Minimum expected compression ratio (uncompressed size / compressed
      size). Values below this threshold trigger an advisory alert."

  - id: slack_channel
    type: STRING
    displayName: Slack Channel
    defaults: "#data-platform"
    description: "Slack channel destination for topic bandwidth efficiency alerts."

triggers:
  - id: periodic_compression_audit
    type: io.kestra.plugin.core.trigger.Schedule
    description: "Audits topic compression efficiency every 4 hours."
    cron: "0 */4 * * *"

tasks:
  - id: sample_kafka_messages
    type: io.kestra.plugin.kafka.Consume
    description: "Samples a representative batch of records from the target Kafka topic."
    topic: "{{ inputs.topic_name }}"
    properties:
      bootstrap.servers: "{{ inputs.bootstrap_servers }}"
      group.id: "{{ inputs.group_id }}"
      auto.offset.reset: "latest"
      enable.auto.commit: "false"
    maxRecords: 50

  - id: evaluate_compression_efficiency
    type: io.kestra.plugin.scripts.python.Script
    description: "Measures raw payload bytes vs compressed zlib bytes to estimate
      producer batch compression efficiency."
    containerImage: python:3.11-slim
    taskRunner:
      type: io.kestra.plugin.scripts.runner.docker.Docker
    inputFiles:
      messages.json: "{{ outputs.sample_kafka_messages.messages | default('[]') | json }}"
    outputFiles:
      - compression_report.json
    script: |
      import json
      import zlib
      from kestra import Kestra

      with open("messages.json", "r", encoding="utf-8") as f:
          records = json.load(f)

      total_records = len(records)
      total_payload_bytes = 0
      total_zlib_compressed_bytes = 0

      for r in records:
          val = r.get("value", "")
          if isinstance(val, (dict, list)):
              raw_bytes = json.dumps(val).encode("utf-8")
          elif isinstance(val, str):
              raw_bytes = val.encode("utf-8")
          else:
              raw_bytes = str(val).encode("utf-8")

          raw_len = len(raw_bytes)
          total_payload_bytes += raw_len
          total_zlib_compressed_bytes += len(zlib.compress(raw_bytes))

      threshold = float("{{ inputs.min_compression_ratio_threshold }}")

      if total_zlib_compressed_bytes > 0:
          estimated_ratio = round(total_payload_bytes / total_zlib_compressed_bytes, 2)
      else:
          estimated_ratio = 1.0

      is_inefficient = (total_records > 0) and (estimated_ratio < threshold)

      report = {
          "topic": "{{ inputs.topic_name }}",
          "sampled_records": total_records,
          "total_uncompressed_bytes": total_payload_bytes,
          "estimated_compressed_bytes": total_zlib_compressed_bytes,
          "compression_ratio": estimated_ratio,
          "target_threshold": threshold,
          "is_inefficient": is_inefficient
      }

      with open("compression_report.json", "w", encoding="utf-8") as f:
          json.dump(report, f, indent=2)

      Kestra.outputs({
          "sampled_count": total_records,
          "compression_ratio": estimated_ratio,
          "is_inefficient": is_inefficient,
          "raw_bytes_kb": round(total_payload_bytes / 1024, 2)
      })

  - id: evaluate_inefficiency_condition
    type: io.kestra.plugin.core.flow.If
    description: "Evaluates whether topic throughput exhibits substandard
      compression ratios."
    condition: "{{ outputs.evaluate_compression_efficiency.vars.is_inefficient }}"
    then:
      - id: notify_slack_compression_gap
        type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
        description: "Dispatches bandwidth optimization guidance to data platform
          engineers via Slack."
        url: "{{ secret('SLACK_WEBHOOK_URL') }}"
        payload: |
          {
            "channel": "{{ inputs.slack_channel }}",
            "text": "*Apache Kafka Topic Compression Inefficiency Advisory*\n• *Topic Name*: `{{ inputs.topic_name }}`\n• *Broker Cluster*: `{{ inputs.bootstrap_servers }}`\n• *Measured Compression Ratio*: `{{ outputs.evaluate_compression_efficiency.vars.compression_ratio }}x` (Target Threshold: {{ inputs.min_compression_ratio_threshold }}x)\n• *Sampled Batch Volume*: {{ outputs.evaluate_compression_efficiency.vars.sampled_count }} records ({{ outputs.evaluate_compression_efficiency.vars.raw_bytes_kb }} KB)\n• *Operational Impact*: Uncompressed message streams waste cross-AZ network egress bandwidth and rapidly exhaust broker NVMe storage.\n• *Recommended Remediation*: Configure upstream producer applications with `compression.type=lz4` or `zstd` and tune `linger.ms=20` to enable multi-record batch compression."
          }

    else:
      - id: log_efficient_compression_status
        type: io.kestra.plugin.core.log.Log
        description: "Logs confirmation of healthy topic compression efficiency."
        message: "Kafka topic {{ inputs.topic_name }} compression audit compliant.
          Estimated ratio: {{
          outputs.evaluate_compression_efficiency.vars.compression_ratio }}x."

  - id: export_compression_manifest
    type: io.kestra.plugin.core.debug.Return
    description: "Outputs structured JSON compression manifest for cluster observability."
    format: |
      {
        "pipeline": "kafka-producer-compression-ratio-sentinel",
        "audited_at": "{{ execution.startDate }}",
        "topic": "{{ inputs.topic_name }}",
        "compression_ratio": {{ outputs.evaluate_compression_efficiency.vars.compression_ratio }},
        "status": "{{ outputs.evaluate_compression_efficiency.vars.is_inefficient ? 'INEFFICIENT' : 'OPTIMIZED' }}"
      }