id: kafka-clickhouse-dead-letter-quarantine-remediation
namespace: company.team
description: |
Realtime Kafka stream consumer subscribing to a dead-letter queue (DLQ) topic.
Inspects poisoned messages, classifies error categories using flow-control Switch,
persists deduplicated audit records into ClickHouse via ReplacingMergeTree, and
notifies Slack on error bursts with automated KV cooldown suppression.
inputs:
- id: kafka_dlq_topic
type: STRING
displayName: Kafka DLQ Topic
description: Kafka dead-letter queue topic to consume poisoned records from.
defaults: "orders-dlq"
- id: kafka_group_id
type: STRING
displayName: Kafka Consumer Group ID
description: Consumer group identifier for the realtime trigger subscriber.
defaults: "kestra-dlq-quarantine-group"
- id: clickhouse_url
type: STRING
displayName: ClickHouse JDBC URL
description: ClickHouse JDBC endpoint URL.
defaults: "jdbc:clickhouse://clickhouse:8123/default"
- id: clickhouse_table
type: STRING
displayName: ClickHouse Quarantine Table
description: Target ClickHouse table storing deduped dead-letter records.
defaults: "kafka_dlq_quarantine"
- id: burst_window_minutes
type: INT
displayName: Burst Window (Minutes)
description: Rolling lookback window in minutes for evaluating error burst frequency.
defaults: 15
- id: burst_error_threshold
type: INT
displayName: Burst Error Threshold
description: Number of DLQ messages in the rolling window required to trigger a
Slack alert.
defaults: 10
- id: alert_cooldown_minutes
type: INT
displayName: Alert Cooldown (Minutes)
description: Minimum duration in minutes between consecutive Slack burst alerts.
defaults: 30
- id: slack_channel
type: STRING
displayName: Slack Notification Channel
description: Target Slack channel receiving error burst notifications.
defaults: "#data-ops-alerts"
variables:
orig_topic: "{{ trigger.value.original_topic ??
trigger.headers['x-original-topic'] ?? trigger.topic ?? 'unknown-topic' }}"
orig_partition: "{{ trigger.value.original_partition ?? trigger.partition ?? 0 }}"
orig_offset: "{{ trigger.value.original_offset ?? trigger.offset ?? 0 }}"
error_class: "{{ trigger.value.error_class ??
trigger.headers['x-exception-fqcn'] ?? (trigger.value.error ??
'UnknownException') }}"
error_message: "{{ trigger.value.error_message ??
trigger.headers['x-exception-message'] ?? (trigger.value.message ?? 'No
error message provided') }}"
stack_trace: "{{ trigger.value.stack_trace ?? trigger.value.trace ?? '' }}"
error_category: >-
{%- set err = (trigger.value.error_class ??
trigger.headers['x-exception-fqcn'] ?? trigger.value.error ?? '') ~ ' ' ~
(trigger.value.error_message ?? trigger.headers['x-exception-message'] ??
trigger.value.message ?? '') | lower -%} {%- if 'serialization' in err or
'jsonparse' in err or 'malformed' in err or 'unrecognized token' in err -%}
deserialization {%- elseif 'schema' in err or 'avro' in err or
'incompatible' in err or 'missing field' in err or 'type mismatch' in err
-%} schema_mismatch {%- elseif 'timeout' in err or 'connection' in err or
'timed out' in err or 'unreachable' in err or '503' in err -%}
downstream_timeout {%- else -%} unknown {%- endif -%}
triggers:
- id: dlq_stream_consumer
type: io.kestra.plugin.kafka.RealtimeTrigger
description: Realtime Kafka consumer subscribing to the dead-letter topic and
firing one execution per poisoned message.
topic: "{{ inputs.kafka_dlq_topic }}"
groupId: "{{ inputs.kafka_group_id }}"
properties:
bootstrap.servers: "{{ secret('KAFKA_BOOTSTRAP_SERVERS') }}"
auto.offset.reset: "latest"
valueDeserializer: JSON
tasks:
- id: ensure_triage_table
type: io.kestra.plugin.jdbc.clickhouse.Query
description: Idempotently create the ClickHouse dead-letter quarantine table
using ReplacingMergeTree for deduplication
url: "{{ inputs.clickhouse_url }}"
username: "{{ secret('CLICKHOUSE_USER') }}"
password: "{{ secret('CLICKHOUSE_PASSWORD') }}"
sql: |
CREATE TABLE IF NOT EXISTS {{ inputs.clickhouse_table }} (
original_topic String,
original_partition Int32,
original_offset Int64,
dlq_topic String,
dlq_partition Int32,
dlq_offset Int64,
error_category LowCardinality(String),
error_class String,
error_message String,
stack_trace String,
severity LowCardinality(String),
remediation_hint String,
payload String,
quarantined_at DateTime64(3) DEFAULT now64(3),
execution_id String
)
ENGINE = ReplacingMergeTree(quarantined_at)
PRIMARY KEY (original_topic, original_partition, original_offset)
ORDER BY (original_topic, original_partition, original_offset);
retry:
type: exponential
interval: PT2S
maxInterval: PT10S
maxAttempts: 3
- id: classify_remediation
type: io.kestra.plugin.core.flow.Switch
description: Evaluate the derived error category to assign operational severity
and targeted remediation instructions
value: "{{ vars.error_category }}"
cases:
deserialization:
- id: set_deserialization_remediation
type: io.kestra.plugin.core.output.OutputValues
description: Assign high severity and payload sanitization instructions for
syntax errors
values:
severity: "HIGH"
remediation_hint: "Verify upstream producer serializer format and inspect
payload for unescaped characters or encoding corruption."
schema_mismatch:
- id: set_schema_remediation
type: io.kestra.plugin.core.output.OutputValues
description: Assign critical severity and schema compatibility instructions for
contract violations
values:
severity: "CRITICAL"
remediation_hint: "Inspect schema evolution rules and verify Schema Registry
compatibility modes between producer and consumer."
downstream_timeout:
- id: set_timeout_remediation
type: io.kestra.plugin.core.output.OutputValues
description: Assign medium severity and retry guidance for transient network or
service unavailability
values:
severity: "MEDIUM"
remediation_hint: "Check downstream service health, connection pools, and
latency metrics; replay is safe once downstream recovers."
defaults:
- id: set_unknown_remediation
type: io.kestra.plugin.core.output.OutputValues
description: Assign warning severity and triage guidance for unclassified exceptions
values:
severity: "WARN"
remediation_hint: "Inspect full payload stack trace in ClickHouse triage table
to diagnose unclassified failure."
- id: record_quarantine_event
type: io.kestra.plugin.jdbc.clickhouse.Query
description: Insert poisoned record into ClickHouse with base64 decoding for
guaranteed SQL safety and redelivery idempotency
url: "{{ inputs.clickhouse_url }}"
username: "{{ secret('CLICKHOUSE_USER') }}"
password: "{{ secret('CLICKHOUSE_PASSWORD') }}"
sql: |
INSERT INTO {{ inputs.clickhouse_table }} (
original_topic, original_partition, original_offset,
dlq_topic, dlq_partition, dlq_offset,
error_category, error_class, error_message, stack_trace,
severity, remediation_hint, payload, execution_id
)
VALUES (
base64Decode('{{ vars.orig_topic | base64encode }}'),
{{ vars.orig_partition }},
{{ vars.orig_offset }},
base64Decode('{{ (trigger.topic ?? inputs.kafka_dlq_topic) | base64encode }}'),
{{ trigger.partition ?? 0 }},
{{ trigger.offset ?? 0 }},
base64Decode('{{ vars.error_category | base64encode }}'),
base64Decode('{{ vars.error_class | base64encode }}'),
base64Decode('{{ vars.error_message | base64encode }}'),
base64Decode('{{ vars.stack_trace | base64encode }}'),
base64Decode('{{ (outputs.set_deserialization_remediation.values.severity ?? outputs.set_schema_remediation.values.severity ?? outputs.set_timeout_remediation.values.severity ?? outputs.set_unknown_remediation.values.severity) | base64encode }}'),
base64Decode('{{ (outputs.set_deserialization_remediation.values.remediation_hint ?? outputs.set_schema_remediation.values.remediation_hint ?? outputs.set_timeout_remediation.values.remediation_hint ?? outputs.set_unknown_remediation.values.remediation_hint) | base64encode }}'),
base64Decode('{{ (trigger.value ?? {}) | toJson | base64encode }}'),
base64Decode('{{ execution.id | base64encode }}')
);
retry:
type: exponential
interval: PT2S
maxInterval: PT10S
maxAttempts: 3
- id: query_window_errors
type: io.kestra.plugin.jdbc.clickhouse.Query
description: Calculate total errors in rolling window, identify top error
classes, and extract sample offsets for triage
url: "{{ inputs.clickhouse_url }}"
username: "{{ secret('CLICKHOUSE_USER') }}"
password: "{{ secret('CLICKHOUSE_PASSWORD') }}"
fetchOne: true
sql: |
SELECT
count(*) AS window_error_count,
topK(3)(error_class) AS top_error_classes,
groupArray(5)(original_offset) AS sample_offsets
FROM {{ inputs.clickhouse_table }}
WHERE quarantined_at >= now64(3) - INTERVAL {{ inputs.burst_window_minutes }} MINUTE;
retry:
type: exponential
interval: PT2S
maxInterval: PT10S
maxAttempts: 3
- id: evaluate_burst_alert
type: io.kestra.plugin.core.flow.If
description: Determine if recent DLQ message frequency exceeds the burst threshold
condition: "{{ outputs.query_window_errors.row.window_error_count >=
inputs.burst_error_threshold }}"
then:
- id: check_cooldown_gate
type: io.kestra.plugin.core.flow.If
description: Verify that no burst alert was sent within the active cooldown
window via KV store
condition: "{{ kv('kafka_dlq_burst_cooldown', errorOnMissing=false) == null }}"
then:
- id: alert_slack_burst
type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
description: Send Slack alert detailing the error burst frequency, top error
classes, and sample offsets
url: "{{ secret('SLACK_WEBHOOK_URL') }}"
payload: |
{
"channel": "{{ inputs.slack_channel }}",
"text": "🚨 *Kafka DLQ Error Burst Alert*\n*Topic:* {{ trigger.topic ?? inputs.kafka_dlq_topic }}\n*Window Errors:* {{ outputs.query_window_errors.row.window_error_count }} messages in the last {{ inputs.burst_window_minutes }} minutes (threshold: {{ inputs.burst_error_threshold }})\n*Top Error Classes:* {{ outputs.query_window_errors.row.top_error_classes | join(', ') }}\n*Sample Offsets:* {{ outputs.query_window_errors.row.sample_offsets | join(', ') }}\n*Latest Category:* {{ vars.error_category }}\n*Severity:* {{ (outputs.set_deserialization_remediation.values.severity ?? outputs.set_schema_remediation.values.severity ?? outputs.set_timeout_remediation.values.severity ?? outputs.set_unknown_remediation.values.severity) }}\n*Remediation:* {{ (outputs.set_deserialization_remediation.values.remediation_hint ?? outputs.set_schema_remediation.values.remediation_hint ?? outputs.set_timeout_remediation.values.remediation_hint ?? outputs.set_unknown_remediation.values.remediation_hint) }}\n*Execution:* `{{ execution.id }}`"
}
retry:
type: exponential
interval: PT2S
maxInterval: PT10S
maxAttempts: 3
- id: set_alert_cooldown
type: io.kestra.plugin.core.kv.Set
description: Record cooldown state in Kestra KV store with TTL to prevent alert
flooding
key: kafka_dlq_burst_cooldown
value: "{{ execution.id }}"
ttl: "PT{{ inputs.alert_cooldown_minutes }}M"
else:
- id: log_cooldown_suppressed
type: io.kestra.plugin.core.log.Log
description: Record that burst alert was suppressed due to active cooldown
window
message: |
ℹ️ DLQ error burst threshold reached ({{ outputs.query_window_errors.row.window_error_count }} errors in {{ inputs.burst_window_minutes }}m), but Slack notification was suppressed by active cooldown window (cooldown: {{ inputs.alert_cooldown_minutes }}m).
else:
- id: log_quarantine_metrics
type: io.kestra.plugin.core.log.Log
description: Log normal quarantine metrics when error frequency is below burst
threshold
message: |
✅ Quarantined DLQ message from {{ vars.orig_topic }} [p:{{ vars.orig_partition }}, o:{{ vars.orig_offset }}].
Category: {{ vars.error_category }} | Severity: {{ (outputs.set_deserialization_remediation.values.severity ?? outputs.set_schema_remediation.values.severity ?? outputs.set_timeout_remediation.values.severity ?? outputs.set_unknown_remediation.values.severity) }}
Rolling window error count: {{ outputs.query_window_errors.row.window_error_count }} (threshold: {{ inputs.burst_error_threshold }}).
Execution: {{ execution.id }}
