Kafka Connect missing dead letter queue safety guard

id: kafka-connect-missing-dlq-safety-guard
namespace: company.team
description: Give a tolerant Kafka Connect sink a dead letter topic before its
  next malformed record is discarded.

concurrency:
  limit: 1
  behavior: QUEUE

inputs:
  - id: connect_url
    type: STRING
    defaults: "{{ secret('KAFKA_CONNECT_URL') }}"
    description: Kafka Connect REST URL without a trailing slash.
  - id: bootstrap_servers
    type: STRING
    defaults: "{{ secret('KAFKA_BOOTSTRAP_SERVERS') }}"
    description: Brokers of the worker's own Kafka cluster.
  - id: connector_name
    type: STRING
    description: Exact connector to audit. Only a running tolerant sink without a
      DLQ can be changed.
  - id: dlq_replication_factor
    type: INT
    defaults: 3
    min: 1
    max: 32767
    description: Approved durability policy. Must fit the broker count. Use 1 only
      in a single-broker lab.
  - id: auto_remediate
    type: BOOL
    defaults: false
  - id: dry_run
    type: BOOL
    defaults: true
  - id: slack_webhook_url
    type: STRING
    defaults: "{{ secret('SLACK_WEBHOOK_URL') }}"

variables:
  dlq_topic: "kestra-dlq-{{ inputs.connector_name }}"

tasks:
  - id: read_config
    type: io.kestra.plugin.kafka.ConnectorGetConfig
    connectUrl: "{{ inputs.connect_url }}"
    connectorName: "{{ inputs.connector_name }}"

  - id: read_plugins
    type: io.kestra.plugin.core.http.Request
    description: Identify sink classes from the worker's plugin inventory, including
      custom connectors.
    uri: "{{ inputs.connect_url }}/connector-plugins"
    method: GET

  - id: read_status
    type: io.kestra.plugin.kafka.ConnectorGetStatus
    connectUrl: "{{ inputs.connect_url }}"
    connectorName: "{{ inputs.connector_name }}"

  - id: list_topics
    type: io.kestra.plugin.kafka.TopicList
    properties:
      bootstrap.servers: "{{ inputs.bootstrap_servers }}"
    callTimeout: PT15S

  - id: assess
    type: io.kestra.plugin.core.output.OutputValues
    values:
      sink: "{{ (outputs.read_plugins.body | jq('map(select(.class == ' ~
        (outputs.read_config.config['connector.class'] | toJson) ~ ' and .type
        == \"sink\")) | length') | first) > 0 }}"
      tolerant: "{{ (outputs.read_config.config['errors.tolerance'] ?? 'none') ==
        'all' }}"
      missing_dlq: "{{
        (outputs.read_config.config['errors.deadletterqueue.topic.name'] ?? '')
        | trim == '' }}"
      running: "{{ outputs.read_status.connectorState == 'RUNNING' and
        outputs.read_status.tasks | length > 0 and (outputs.read_status.tasks |
        jq('map(select(.state != \"RUNNING\")) | length') | first) == 0 }}"
      topic_exists: "{{ outputs.list_topics.topics contains render(vars.dlq_topic) }}"

  - id: report
    type: io.kestra.plugin.core.log.Log
    message: >-
      {{ inputs.connector_name }}: sink={{ outputs.assess.values.sink }}
      errors.tolerance.all={{ outputs.assess.values.tolerant }} missing_dlq={{
      outputs.assess.values.missing_dlq }} running={{
      outputs.assess.values.running }}. Proposed topic={{ render(vars.dlq_topic)
      }} exists={{ outputs.assess.values.topic_exists }}. A configured DLQ is
      preserved. Skipped records cannot be recovered by changing this
      configuration.

  - id: unprotected_sink
    type: io.kestra.plugin.core.flow.If
    condition: "{{ (outputs.assess.values.sink | string) == 'true' and
      (outputs.assess.values.tolerant | string) == 'true' and
      (outputs.assess.values.missing_dlq | string) == 'true' and
      (outputs.assess.values.running | string) == 'true' }}"
    then:
      - id: notify_gap
        type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
        url: "{{ inputs.slack_webhook_url }}"
        payload: "{{ {'text': 'Kafka Connect sink ' ~ inputs.connector_name ~ '
          tolerates conversion errors without a DLQ. Future rejected records
          need a durable destination. Proposed topic: ' ~ render(vars.dlq_topic)
          ~ '. Execution: ' ~ execution.id} | toJson }}"

      - id: live_mode
        type: io.kestra.plugin.core.flow.If
        condition: "{{ inputs.auto_remediate and not inputs.dry_run }}"
        then:
          - id: approve
            type: io.kestra.plugin.core.flow.Pause
            description: >-
              Approve a task restart and a dedicated kestra-dlq topic. Confirm
              these brokers belong to this worker and the topic prefix is
              reserved for this flow. Conversion failures after the change will
              contain original record bytes. Restrict DLQ access and retention
              to match the source data's sensitivity.
            onResume:
              - id: approved
                type: BOOL
                defaults: false

          - id: approved_change
            type: io.kestra.plugin.core.flow.If
            condition: "{{ outputs.approve.onResume.approved ?? false }}"
            then:
              - id: fresh_config
                type: io.kestra.plugin.kafka.ConnectorGetConfig
                connectUrl: "{{ inputs.connect_url }}"
                connectorName: "{{ inputs.connector_name }}"

              - id: fresh_status
                type: io.kestra.plugin.kafka.ConnectorGetStatus
                connectUrl: "{{ inputs.connect_url }}"
                connectorName: "{{ inputs.connector_name }}"

              - id: reject_stale_plan
                type: io.kestra.plugin.core.flow.If
                condition: "{{ outputs.fresh_config.config != outputs.read_config.config or
                  outputs.fresh_status.connectorState != 'RUNNING' or
                  outputs.fresh_status.tasks | length == 0 or
                  (outputs.fresh_status.tasks | jq('map(select(.state !=
                  \"RUNNING\")) | length') | first) > 0 }}"
                then:
                  - id: stale_plan
                    type: io.kestra.plugin.core.execution.Fail
                    errorMessage: Connector config or task state changed during approval. Audit
                      again before writing.

              - id: provision_dlq
                type: io.kestra.plugin.kafka.TopicCreate
                properties:
                  bootstrap.servers: "{{ inputs.bootstrap_servers }}"
                callTimeout: PT15S
                topic: "{{ render(vars.dlq_topic) }}"
                partitions: 1
                replicationFactor: "{{ inputs.dlq_replication_factor }}"
                ifNotExists: true
                configs:
                  cleanup.policy: delete
                  retention.ms: "604800000"
                  retention.bytes: "-1"

              - id: inspect_dlq
                type: io.kestra.plugin.kafka.TopicDescribe
                properties:
                  bootstrap.servers: "{{ inputs.bootstrap_servers }}"
                callTimeout: PT15S
                topic: "{{ render(vars.dlq_topic) }}"

              - id: check_topic_shape
                type: io.kestra.plugin.core.flow.If
                condition: "{{ outputs.inspect_dlq.partitionCount != 1 or
                  outputs.inspect_dlq.replicationFactor !=
                  inputs.dlq_replication_factor or
                  outputs.inspect_dlq.configs['cleanup.policy'] != 'delete' or
                  outputs.inspect_dlq.configs['retention.ms'] != '604800000' or
                  outputs.inspect_dlq.configs['retention.bytes'] != '-1' }}"
                then:
                  - id: wrong_topic_shape
                    type: io.kestra.plugin.core.execution.Fail
                    errorMessage: Existing DLQ does not match the one-partition replication and
                      seven-day retention policy. It was left untouched.

              - id: before_update
                type: io.kestra.plugin.kafka.ConnectorGetConfig
                description: Topic creation may take time. Recheck the entire approved config
                  immediately before replacement.
                connectUrl: "{{ inputs.connect_url }}"
                connectorName: "{{ inputs.connector_name }}"

              - id: before_update_status
                type: io.kestra.plugin.kafka.ConnectorGetStatus
                connectUrl: "{{ inputs.connect_url }}"
                connectorName: "{{ inputs.connector_name }}"

              - id: refuse_concurrent_edit
                type: io.kestra.plugin.core.flow.If
                condition: "{{ outputs.before_update.config != outputs.read_config.config or
                  outputs.before_update_status.connectorState != 'RUNNING' or
                  outputs.before_update_status.tasks | length == 0 or
                  (outputs.before_update_status.tasks | jq('map(select(.state !=
                  \"RUNNING\")) | length') | first) > 0 }}"
                then:
                  - id: concurrent_edit
                    type: io.kestra.plugin.core.execution.Fail
                    errorMessage: Config changed while provisioning the DLQ. Topic remains for a
                      reconciled retry; config was not replaced.

              - id: install_dlq
                type: io.kestra.plugin.kafka.ConnectorUpdateConfig
                description: Preserve every existing key and add only the three DLQ settings.
                connectUrl: "{{ inputs.connect_url }}"
                connectorName: "{{ inputs.connector_name }}"
                config: "{{ outputs.before_update.config |
                  merge({'errors.deadletterqueue.topic.name':
                  render(vars.dlq_topic),
                  'errors.deadletterqueue.context.headers.enable': 'true',
                  'errors.deadletterqueue.topic.replication.factor':
                  inputs.dlq_replication_factor | string}) }}"

              - id: await_tasks
                type: io.kestra.plugin.core.flow.LoopUntil
                condition: "{{ outputs.poll_status.connectorState == 'RUNNING' and
                  outputs.poll_status.tasks | length > 0 and
                  (outputs.poll_status.tasks | jq('map(select(.state !=
                  \"RUNNING\")) | length') | first) == 0 }}"
                failOnMaxReached: true
                checkFrequency:
                  interval: PT5S
                  maxIterations: 12
                tasks:
                  - id: poll_status
                    type: io.kestra.plugin.kafka.ConnectorGetStatus
                    connectUrl: "{{ inputs.connect_url }}"
                    connectorName: "{{ inputs.connector_name }}"

              - id: verify_config
                type: io.kestra.plugin.kafka.ConnectorGetConfig
                connectUrl: "{{ inputs.connect_url }}"
                connectorName: "{{ inputs.connector_name }}"

              - id: verify_topics
                type: io.kestra.plugin.kafka.TopicList
                properties:
                  bootstrap.servers: "{{ inputs.bootstrap_servers }}"
                callTimeout: PT15S

              - id: verify_result
                type: io.kestra.plugin.core.flow.If
                condition: "{{ outputs.verify_config.config != (outputs.before_update.config |
                  merge({'errors.deadletterqueue.topic.name':
                  render(vars.dlq_topic),
                  'errors.deadletterqueue.context.headers.enable': 'true',
                  'errors.deadletterqueue.topic.replication.factor':
                  inputs.dlq_replication_factor | string})) or not
                  (outputs.verify_topics.topics contains render(vars.dlq_topic))
                  }}"
                then:
                  - id: verification_failed
                    type: io.kestra.plugin.core.execution.Fail
                    errorMessage: DLQ topic or exact expected config absent after update. Inspect
                      the connector before retrying.

              - id: repaired
                type: io.kestra.plugin.core.log.Log
                message: >-
                  {{ inputs.connector_name }} now names {{
                  render(vars.dlq_topic) }} with context headers. The config
                  readback matches and all tasks reported RUNNING. Validate
                  delivery with a malformed canary before closing the incident.

triggers:
  - id: daily_audit
    type: io.kestra.plugin.core.trigger.Schedule
    cron: "0 6 * * *"
    disabled: true
    inputs:
      connector_name: orders-sink
      auto_remediate: false
      dry_run: true

errors:
  - id: alert_failure
    type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
    url: "{{ inputs.slack_webhook_url }}"
    payload: "{{ {'text': 'Kafka Connect DLQ guard failed for ' ~
      inputs.connector_name ~ '. Execution ' ~ execution.id ~ '. A provisioned
      topic may remain; inspect actual config before retrying.'} | toJson }}"

Safety

Replacing a connector config restarts its tasks. DLQ records contain the original bytes and error context, which can expose sensitive data. Restrict both DLQ access and Kestra execution access: the saved config can contain credentials.

Guard	What it stops
One exact connector, sink type, errors.tolerance=all and empty DLQ	Changing sources, fail-fast sinks or an existing error-routing policy
RUNNING connector with nonempty RUNNING tasks	Restarting an intentional pause or acting on an already failed pipeline
Two mode inputs and explicit Pause approval	An audit creating a topic or restarting tasks
Full config and status rechecks after approval and before replacement	Overwriting a plan that changed while approval or provisioning waited
Reserved topic name, shape and retention checks	Adopting a topic that would immediately expire or compact rejected records
Exact config readback, topic existence and task health	Reporting an accepted but ineffective config update as a repair

How it works

read_config (io.kestra.plugin.kafka.ConnectorGetConfig), read_plugins (io.kestra.plugin.core.http.Request), read_status (ConnectorGetStatus) and list_topics (TopicList) audit the selected connector. assess requires a canonical sink class, tolerance all, an empty or whitespace-only DLQ name and running tasks.
unprotected_sink posts the gap to Slack. live_mode requires auto_remediate=true and dry_run=false; approve waits for approved=true. Declining ends without a write.
fresh_config and fresh_status reject a stale approval. provision_dlq (TopicCreate) creates kestra-dlq-<connector_name> idempotently. inspect_dlq (TopicDescribe) checks one partition, the requested replication factor, delete cleanup, seven-day retention and no byte retention cap. Existing topics are never altered.
before_update and before_update_status recheck eligibility. install_dlq (ConnectorUpdateConfig) preserves the current config and adds the topic name, context headers and replication setting. await_tasks polls for up to 12 iterations at five-second intervals. verify_config, verify_topics and verify_result require the exact expected config and a visible topic. Any error alerts Slack.

What you get

A visible warning for a silent-loss configuration, a dedicated retained DLQ and a verified config change. A failed partial pass leaves its topic available for a reconciled retry.

Who it's for

Platform teams responsible for Kafka Connect sinks where malformed input is tolerated so healthy traffic can continue.

Why orchestrate this with Kestra

The incident needs an audit, a human decision, broker provisioning and a connector restart across two APIs. Kestra keeps that approval and its snapshot with the execution, queues overlapping runs and alerts when the readback fails.

Tested end to end

Kestra 2.0.5, Kafka plugin 1.10.5, Slack plugin 2.1.0 and Confluent Kafka/Connect 7.9.0. FileStream sinks used JsonConverter without schemas. Slack was the existing local HTTP sink.

Before repair, kafka-consumer-groups --describe showed committed offset 5, end offset 5 and lag 0. The sink contained only the two valid records from five inputs; three malformed records had no DLQ. The connector and task stayed RUNNING. After repair, a new valid record reached the sink and a new malformed record appeared once in the DLQ. kafka-console-consumer read its original bytes with topic and offset headers; kafka-get-offsets reported DLQ end offset 1. Previously skipped records were not recovered.

Case	Execution	Result
Existing DLQ	`Qgc04DHgHKcJCLJZ0Dk9v`	SUCCESS; existing topic and header policy preserved
Dry run	`3QgqARsgaNKE1CkmWQB1Xr`	SUCCESS; auto_remediate true, dry_run true, no writes
Declined approval	`2X51aYrxoS54uhqGwt5IyV`	SUCCESS; config unchanged
Remediation	`6WiyDTd7V6PaNbgjjI6DcH`	SUCCESS; one valid row delivered and one malformed record retained
Repeat	`57aAXsFKTk4eHEWU3HS34K`	SUCCESS; no topic or config writes
Refused config PUT	`713tfKpHbTUP6VYJ6sEtrW`	FAILED as intended; created topic retained
Partial recovery	`4PcFCEGS382zoLkncWKJPQ`	SUCCESS; existing topic reused and delivery proved
Config edit during approval	`60nROK53WKdFxJRhwdCzmr`	FAILED as intended; operator edit preserved, no writes
Tolerant source	`376y2sbYCwmaoll7gUu9Ra`	SUCCESS; no writes
Fail-fast sink	`6dM4xp8qUanBmOHnJrjtV3`	SUCCESS; no writes
Intentional pause	`2E7IWfJmHVhRdw52YbugRY`	SUCCESS; no writes
Automation disabled	`6CHKGSAUeC7fXk6qzeg07v`	SUCCESS; dry_run false, auto_remediate false, no writes
Whitespace DLQ name	`3DWXtf4LJzTXm2nCGzRF5e`	SUCCESS; treated as missing
Wrong-retention lookalike	`2CHP666vpyin8blq66664v`	FAILED as intended; topic and connector untouched
Three replicas on one broker	`3UrHGTCStiZrByjrhizZZN`	FAILED as intended; no connector write
HTTP 200 without DLQ keys	`2xn6oajzprPvz8XQaJJ8fR`	FAILED as intended; config readback caught the ineffective update

A second fresh broker/worker pair reproduced the baseline and repeated healthy 4EiF4kgXZkUbFdpiUiql63, remediation 5MmjGFiigKtlZevfo7PXgO, race_or_partial qGOzz1iLo9rQEp5Ruwkzc on the same runtime bytes. The no-effect test forwarded a genuine config PUT with the DLQ keys stripped: Connect returned HTTP 200, then the real config readback failed verification. All readbacks came from the real worker.

Prerequisites

Kafka and a Kafka Connect worker on that same cluster, the Kafka plugin and a Slack incoming webhook. Reserve the kestra-dlq- prefix. Connector names must form valid Kafka topic names after the prefix is added. connector.class must use the canonical class name listed by /connector-plugins.

Kestra needs Connect config/status reads, plugin inventory reads and config update permission on the named connector. Its Kafka principal needs Describe, DescribeConfigs and Create for the reserved DLQ names. The worker's principal needs Write and Describe on the DLQ. Configure authentication/TLS properties on all relevant Kafka tasks and credentials on both Connect tasks and the HTTP inventory request for a secured deployment.

Secrets

KAFKA_CONNECT_URL: worker REST base URL.
KAFKA_BOOTSTRAP_SERVERS: brokers used by that worker.
SLACK_WEBHOOK_URL: incoming webhook for findings and failures.

Inputs

Input	Default	Purpose
`connect_url`	Secret	Connect REST URL without a trailing slash
`bootstrap_servers`	Secret	Worker cluster's brokers
`connector_name`	Required	One exact connector
`dlq_replication_factor`	`3`	Approved replica policy; `1` only for a single-broker lab
`auto_remediate`	`false`	Permit a change to reach approval
`dry_run`	`true`	Keep every write disabled
`slack_webhook_url`	Secret	Webhook; permits a local sink for testing

Quick start

Add the secrets and reserve the DLQ topic prefix with suitable ACLs. Confirm the broker list belongs to the worker.
Set connector_name and run with the defaults. Check the sink, tolerance, missing-DLQ and running flags in report.
Set a replication factor that fits the cluster. Run with auto_remediate=true and dry_run=false, then approve the task restart with approved=true.
Send a controlled malformed record and read it from the DLQ with its context headers. Confirm that a valid record still reaches the destination before closing the incident.
To schedule audits, replace the trigger's example connector name. The daily Schedule ships disabled and forces both safe-mode inputs.

How to extend

Change retention only by updating both provision_dlq.configs and check_topic_shape. Existing topics with another policy are refused rather than modified. Add destination-specific canaries before treating this as proof for a different sink implementation.

Limits: one connector per execution. Existing nonempty DLQ names are preserved even when their topics are absent; Connect normally attempts to create a configured missing DLQ itself. This flow prevents future converter-error loss and does not replay skipped records. Concurrency is one for this deployed flow only. Kafka Connect has no conditional config PUT or stable incarnation ID, so coordinate external config writers and connector deletion throughout live runs; the last-read-to-write interval cannot be made atomic here. The flow verifies config, topic policy and reported task health. Canary delivery was proved independently for FileStream/JsonConverter. Authenticated clusters, replica failures, other sinks and real Slack delivery were not exercised.

Prevent Kafka Connect sinks from dropping malformed records

Safety

How it works

What you get

Who it's for

Why orchestrate this with Kestra

Tested end to end

Prerequisites

Secrets

Inputs

Quick start

How to extend

Links

Tasks

Orchestrate with Kestra

Share this Blueprint

Create a Blueprint

New to Kestra?

Prevent Kafka Connect sinks from dropping malformed records

Safety

How it works

What you get

Who it's for

Why orchestrate this with Kestra

Tested end to end

Prerequisites

Secrets

Inputs

Quick start

How to extend

Links

Tasks

Orchestrate with Kestra

Share this Blueprint

Create a Blueprint

More Data Blueprints

NocoDB Database Platform Health, Storage Engine, and Project Sentinel

Detect and Resume Stuck Paused NSQ Channels with Accumulating Backlog

Object Storage Integrity and Streaming Replica Parity Audit

Stop a Bad Monthly Load with SLA Assertions on NYC Taxi Data in DuckDB

Automated OpenLineage Metadata Extraction and Marquez Governance Pipeline

Write-Audit-Publish (WAP) and Atomic Merge Gate for Apache Iceberg with Project Nessie

New to Kestra?