Applies to VAST Cluster 5.5.
When a function keeps failing on an event, DataEngine retries it and then writes the event to your tenant's dead-letter topic, a topic on the tenant's VAST Event Broker. Nothing reads that topic unless you do. This page shows the pattern: a Kafka consumer that reads the dead-letter topic, decodes why each event failed, and either alerts on it or replays it.
This isn't a built-in DataEngine integration in 5.5. DataEngine writes dead-lettered events to the topic, and that's all: there's no dead-letter view, alert or replay command. You consume the topic with an ordinary Kafka client. How retries and dead-lettering work is in architecture, and how to set retries is in pipelines as code. Kafka clients, consumer groups and topic settings are covered in the VAST Event Broker section.
Commands use the environment variables from the DataEngine quickstart, plus KAFKA_BOOTSTRAP, the tenant broker's address as <host>:<port> (the broker.url in vastde triggers get <trigger> -o yaml). The scripts need a Kafka client: pip install kafka-python-ng. The outputs come from a lab cluster running VAST Cluster 5.5.0-SP6 with vastde v5.5.0-sp2 and kafka-python-ng 2.2.3, where the tenant's broker is the Event Broker view docs-kafka, the trigger topic is default and the dead-letter topic is default-dlt. Your names, GUIDs, addresses and times will differ.
Before you start
You need:
- A pipeline whose links set
retries. Without it, a server default applies that can keep a failing event retrying for a long time before it's dead-lettered. See set the retries. - The name of your tenant's dead-letter topic. Your platform team chose it when they enabled DataEngine. List the topics on the tenant's broker view to find it:
vastde topics list --database-name docs-kafka
Name Database
---------------------------------------------------------
default docs-kafka
default-dlt docs-kafka
- Network access from where your consumer runs to the broker, and the broker's authentication settings if it has any. See Event Broker security.
Every pipeline in the tenant shares the dead-letter topic. Your consumer sees other pipelines' failures too, so filter on the headers described below.
A function that fails
This handler raises on objects smaller than 10 bytes:
# picky/main.py
MIN_SIZE = 10
def init(ctx):
ctx.logger.info("picky ready")
def handler(ctx, event):
obj = event.get_data()["Records"][0]["s3"]["object"]
ctx.logger.info("event %s: %s (%d bytes)", event.id, obj["key"], obj["size"])
if obj["size"] < MIN_SIZE:
raise ValueError(f"{obj['key']} is {obj['size']} bytes, need at least {MIN_SIZE}")
ctx.logger.info("processed %s", obj["key"])
Build it and create it as picky, and create an element trigger checked-uploads for keys under checked/, as in the quickstart. The pipeline retries a failed event once:
name: size-check
description: fails on small objects; failed events go to the dead-letter topic
kubernetes_cluster_vrn: vast:dataengine:kubernetes-clusters:test-cluster
namespace: dataengine-ns-da47bae0-aefd-4e85-a5a9-2020d6a2cb91
manifest:
triggers:
- name: uploads
vrn: vast:dataengine:triggers:checked-uploads
function_deployments:
- name: picky
function_vrn: vast:dataengine:functions:picky
links:
- source: [uploads]
destination: [picky]
config:
retries: 1
vastde pipelines create --config @size-check.yaml --deploy
Pipeline created: size-check
Name: size-check
Description: fails on small objects; failed events go to the dead-letter topic
Tags: []
GUID: ea12f590-1e72-4b1d-ae5d-d935ceed6afc
Owner: [id: 4294967295, id-type: vid]
Created At: 2026-10-08T14:10:26Z
Updated At: 2026-10-08T14:10:26Z
VRN: vast:dataengine:pipelines:size-check
Deploying pipeline: size-check
Pipeline deployed successfully
Note the time before you upload, in milliseconds, so the consumer can start reading from it. Then upload a 4-byte object:
date +%s%3N > since.ms
printf tiny > tiny.txt; aws --endpoint-url "$S3_ENDPOINT" s3 cp tiny.txt "s3://$BUCKET/checked/tiny.txt" >/dev/null
sleep 20; vastde logs get size-check --since 2m --scope user
2026-10-08 14:10:54.76 [picky] [INFO] [user] picky ready
2026-10-08 14:11:24.93 [picky] [INFO] [user] event 87982405058579: checked/tiny.txt (4 bytes)
2026-10-08 14:11:26.94 [picky] [INFO] [user] event 87982405058579: checked/tiny.txt (4 bytes)
With retries: 1, the handler ran twice, with the same event ID, and then the event was dead-lettered. Both attempts are one trace with status ERROR:
vastde traces list size-check --since 5m
Trace ID Status Pipeline Function Start Time End Time
--------------------------------------------------------------------------------------------------------------------------------------
a2ffe8ac414a747bf8d6c613a6796e06 OK size-check picky 2026-10-08 14:10:54.766 2026-10-08 14:10:54.769
0000500500000013777f3b6d508f4c57 ERROR size-check picky 2026-10-08 14:11:24.930 2026-10-08 14:11:26.948
(The OK trace is the instance's init.)
Read and decode the dead-letter topic
This consumer reads every partition of the topic from a point in time, without a consumer group, and prints one line per dead-lettered event, with the error message decoded. With --raw, it prints each record's headers, key and value first:
"""Read the tenant dead-letter topic and print one line per failed event.
usage: python dlt_consumer.py <bootstrap> <dead-letter topic> [--since-ms N] [--raw]
"""
import base64
import sys
from kafka import KafkaConsumer, TopicPartition
bootstrap, topic = sys.argv[1], sys.argv[2]
since_ms = int(sys.argv[sys.argv.index("--since-ms") + 1]) if "--since-ms" in sys.argv else 0
raw = "--raw" in sys.argv
consumer = KafkaConsumer(bootstrap_servers=bootstrap, enable_auto_commit=False, consumer_timeout_ms=10000)
parts = [TopicPartition(topic, p) for p in consumer.partitions_for_topic(topic)]
consumer.assign(parts)
offsets = consumer.offsets_for_times({tp: since_ms for tp in parts})
for tp in parts:
if offsets[tp] is None:
consumer.seek_to_end(tp)
else:
consumer.seek(tp, offsets[tp].offset)
for rec in consumer:
h = {k: v.decode() for k, v in rec.headers}
if raw:
print(f"--- partition={rec.partition} offset={rec.offset} timestamp={rec.timestamp}")
for k, v in rec.headers:
print(f" header {k}: {v.decode(errors='replace')}")
print(f" key: {rec.key.decode() if rec.key else None}")
print(f" value: {rec.value.decode(errors='replace') if rec.value else None}")
error = base64.b64decode(h.get("ce_knativeerrordata", "")).decode(errors="replace")
print(f"{h.get('ce_time')} id={h.get('ce_id')} source={h.get('ce_source')} "
f"path={h.get('ce_elementpath')} code={h.get('ce_knativeerrorcode')} "
f"dest={h.get('ce_knativeerrordest')} error={error!r}")
python dlt_consumer.py "$KAFKA_BOOTSTRAP" default-dlt --since-ms $(cat since.ms) --raw
--- partition=7 offset=0 timestamp=1791468687032
header ce_specversion: 1.0
header ce_id: 87982405058579
header ce_source: vastdata.com:checked-uploads.f1c78fd2-63e7-408c-9e7b-0bd4a3fd3341
header ce_type: vastdata.com:Element.ElementCreated
header content-type: application/json
header ce_subject: vast-broker-docs-kafka.default
header ce_time: 2026-10-08T14:11:24.228703Z
header ce_knativeerrorcode: 500
header ce_elementsourcetype: vast:s3
header ce_knativeerrordata: Y2hlY2tlZC90aW55LnR4dCBpcyA0IGJ5dGVzLCBuZWVkIGF0IGxlYXN0IDEw
header ce_elementparenthandle: 411921858243579213
header ce_elementpath: docs-src/checked/tiny.txt
header ce_knativekafkapartition: 0
header ce_knativekafkaoffset: 16
header ce_knativeerrordest: http://picky-nbdpm6d3aaaaa.dataengine-ns-da47bae0-aefd-4e85-a5a9-2020d6a2cb91.svc.cluster.local
header ce_elementhandle: 4343675762408176142
key: None
value: {"Records": [{"eventVersion": "2.2", "eventSource": "vast:s3", "awsRegion": "<cluster name>", "eventTime": "2026-10-08T14:11:24.228703Z", "eventName": "ObjectCreated:Put", "userIdentity": {"principalId": "<user>"}, "requestParameters": {"sourceIPAddress": "<client IP>"}, "responseElements": {"x-amz-request-id": "0x50010005fff9", "x-amz-id-2": "0x50010005fff9"}, "s3": {"s3SchemaVersion": "1.0", "configurationId": "checked-uploads.f1c78fd2-63e7-408c-9e7b-0bd4a3fd3341", "bucket": {"name": "docs-src", "ownerIdentity": {"principalId": "<user>"}, "arn": "arn:aws:s3:::docs-src"}, "object": {"key": "checked/tiny.txt", "size": 4, "eTag": "d60cadf1a41c651e1f0ade50136bad43", "sequencer": "000400000000000f4241"} } } ] }
2026-10-08T14:11:24.228703Z id=87982405058579 source=vastdata.com:checked-uploads.f1c78fd2-63e7-408c-9e7b-0bd4a3fd3341 path=docs-src/checked/tiny.txt code=500 dest=http://picky-nbdpm6d3aaaaa.dataengine-ns-da47bae0-aefd-4e85-a5a9-2020d6a2cb91.svc.cluster.local error='checked/tiny.txt is 4 bytes, need at least 10'
One event that failed twice gave one dead-letter record.
The record
A dead-lettered record is the original event as a CloudEvent in Kafka binary form, with three error headers added:
| Part | What it holds | Use it to |
|---|---|---|
| Value | The event's original data, unchanged. For an element event, the S3-style {"Records": [...]} JSON. |
Replay the event, or see what the function was given |
| Key | None | (Don't rely on the key for partitioning or deduplication.) |
ce_id |
The event's ID. Retries, and replays, keep it. | Deduplicate; find the event's logs and trace |
ce_source |
The trigger, as vastdata.com:<trigger>.<trigger GUID> |
Tell the shared topic's pipelines apart |
ce_subject |
The topic the event came from, as vast-broker-<broker view>.<topic> |
Find the topic to replay to |
ce_type, ce_time, ce_elementpath, ce_element* |
The event's other attributes, as described in events | Find the object |
ce_knativekafkapartition, ce_knativekafkaoffset |
Where the event was in the topic it came from | Trace it back to the source topic |
ce_knativeerrorcode |
The HTTP status of the last failed attempt: 500 when the handler raised |
Tell handler errors from other failures |
ce_knativeerrordata |
The last failed attempt's response body, base64-encoded. When the handler raised, it's the exception's message. | Say why it failed |
ce_knativeerrordest |
The address of the function deployment that failed. It starts with the function deployment's name (picky here). |
Tell which function failed |
The record doesn't name the pipeline. If several pipelines use the same trigger, tell them apart by ce_knativeerrordest.
Alert on dead letters
To alert, run the same decoding in a long-running consumer and send each line to your alerting or chat system instead of printing it:
- Track your position with a consumer group of your own, so that a restarted consumer carries on where it stopped instead of seeking by time. See consume in a group.
- Filter on
ce_sourceorce_knativeerrordestto alert only on your own pipelines. - Alert before retention removes the records. Dead-lettered records are kept only as long as the topic's retention allows. Retention is a topic setting: see retention and ask your platform team what it's set to.
Replay a dead-lettered event
There are two ways to run a failed event again. Which one to use depends on what was wrong.
Fix the input and re-upload
If the input was wrong, fix it and write the object again. That's a new event, with a new ID and a new trace. Here the object is overwritten with a longer one:
echo "now long enough" > fixed.txt; aws --endpoint-url "$S3_ENDPOINT" s3 cp fixed.txt "s3://$BUCKET/checked/tiny.txt" >/dev/null
sleep 10; vastde logs get size-check --since 1m --scope user
2026-10-08 14:12:21.09 [picky] [INFO] [user] event 87982405058581: checked/tiny.txt (16 bytes)
2026-10-08 14:12:21.09 [picky] [INFO] [user] processed checked/tiny.txt
This is the safest way to replay: it uses only what DataEngine does with any upload. If the function was wrong instead, publish a fixed revision and deploy it first, then re-upload the object to fire the trigger again.
Re-produce the record to the trigger's topic
If the failure was transient (a service the function calls was down, say) or you've deployed a fixed function, you can re-produce the dead-lettered record to the topic it came from, so the original event is delivered again. This isn't a documented DataEngine feature: it's what we observed on VAST Cluster 5.5.0-SP6.
The topic is the last part of ce_subject (default here; it's also the Topic that vastde triggers get shows). Remove the ce_knativeerror* headers and keep the others, so the record looks like the original event:
"""Re-produce one dead-letter record to the trigger's topic, without the knativeerror* headers.
usage: python replay.py <bootstrap> <dead-letter topic> <partition> <offset> <target topic>
"""
import sys
from kafka import KafkaConsumer, KafkaProducer, TopicPartition
bootstrap, dlt, partition, offset, target = sys.argv[1], sys.argv[2], int(sys.argv[3]), int(sys.argv[4]), sys.argv[5]
consumer = KafkaConsumer(bootstrap_servers=bootstrap, enable_auto_commit=False, consumer_timeout_ms=10000)
tp = TopicPartition(dlt, partition)
consumer.assign([tp])
consumer.seek(tp, offset)
rec = next(iter(consumer))
headers = [(k, v) for k, v in rec.headers if not k.startswith("ce_knativeerror")]
producer = KafkaProducer(bootstrap_servers=bootstrap)
md = producer.send(target, key=rec.key, value=rec.value, headers=headers).get(timeout=30)
producer.flush()
print(f"re-produced {dlt}[{partition}]@{offset} to {md.topic}[{md.partition}]@{md.offset} "
f"with {len(headers)} headers: {', '.join(k for k, _ in headers)}")
The producer must use the settings Event Broker needs (no idempotence, no compression); kafka-python-ng's defaults meet them. For other clients, see required producer settings. Replay the record at partition 7, offset 0:
python replay.py "$KAFKA_BOOTSTRAP" default-dlt 7 0 default
re-produced default-dlt[7]@0 to default[5]@16 with 13 headers: ce_specversion, ce_id, ce_source, ce_type, content-type, ce_subject, ce_time, ce_elementsourcetype, ce_elementparenthandle, ce_elementpath, ce_knativekafkapartition, ce_knativekafkaoffset, ce_elementhandle
The function received the original event again, with its original ID and its original data:
sleep 15; vastde logs get size-check --since 1m --scope user
2026-10-08 14:12:21.09 [picky] [INFO] [user] event 87982405058581: checked/tiny.txt (16 bytes)
2026-10-08 14:12:21.09 [picky] [INFO] [user] processed checked/tiny.txt
2026-10-08 14:12:37.03 [picky] [INFO] [user] event 87982405058579: checked/tiny.txt (4 bytes)
2026-10-08 14:12:39.05 [picky] [INFO] [user] event 87982405058579: checked/tiny.txt (4 bytes)
The replayed event still says the object is 4 bytes, although the object had been overwritten by then, so it failed again, twice, and was dead-lettered again. Its attempts joined the original event's trace, whose end time moved from 14:11:26 to 14:12:39:
vastde traces list size-check --since 10m
Trace ID Status Pipeline Function Start Time End Time
--------------------------------------------------------------------------------------------------------------------------------------
a2ffe8ac414a747bf8d6c613a6796e06 OK size-check picky 2026-10-08 14:10:54.766 2026-10-08 14:10:54.769
0000500500000013777f3b6d508f4c57 ERROR size-check picky 2026-10-08 14:11:24.930 2026-10-08 14:12:39.053
0000500500000015777f3b6d508f4c57 OK size-check picky 2026-10-08 14:12:21.095 2026-10-08 14:12:21.096
And the dead-letter topic now has two identical records for the event:
python dlt_consumer.py "$KAFKA_BOOTSTRAP" default-dlt --since-ms $(cat since.ms)
2026-10-08T14:11:24.228703Z id=87982405058579 source=vastdata.com:checked-uploads.f1c78fd2-63e7-408c-9e7b-0bd4a3fd3341 path=docs-src/checked/tiny.txt code=500 dest=http://picky-nbdpm6d3aaaaa.dataengine-ns-da47bae0-aefd-4e85-a5a9-2020d6a2cb91.svc.cluster.local error='checked/tiny.txt is 4 bytes, need at least 10'
2026-10-08T14:11:24.228703Z id=87982405058579 source=vastdata.com:checked-uploads.f1c78fd2-63e7-408c-9e7b-0bd4a3fd3341 path=docs-src/checked/tiny.txt code=500 dest=http://picky-nbdpm6d3aaaaa.dataengine-ns-da47bae0-aefd-4e85-a5a9-2020d6a2cb91.svc.cluster.local error='checked/tiny.txt is 4 bytes, need at least 10'
What this means for a replay tool:
- A replay delivers the event as it was. It carries the data from when the event happened, not the object's current state. If the input was the problem, re-upload instead.
- A replay that fails is dead-lettered again, as a new record that looks the same as the first. Use the record's offset, or keep a list of the
ce_ids you've replayed, so that you don't replay the same event in a loop. - A replay keeps the event's ID and trace. The handler can't tell it from a retry, so it must be idempotent (see retries and the dead-letter topic).
- A replay reaches every pipeline that uses the trigger. Delivery is matched on the event's
ce_source, which names the trigger, not the pipeline, so every pipeline linked to that trigger gets the replayed event, not only the one that failed. If several pipelines share the trigger, re-upload instead, or give the failing pipeline its own trigger.
Clean up
# not-run
vastde pipelines delete size-check
vastde triggers delete checked-uploads
vastde functions delete picky
rm since.ms
The dead-lettered records expire with the topic's retention.
See also
- Architecture: retries and the dead-letter topic: which failures are retried and dead-lettered.
- Pipelines as code: set the retries
- Events: the attributes in the record's headers.
- VAST Event Broker: consuming, producing, topics and security.
- Integrations