--- title: "Consume the dead-letter topic" slug: "dev-dataengine-integrations-dead-letter-topic" description: "Learn how to handle dead-letter topics in DataEngine, retry failed events, and replay them using Kafka consumers effectively." tags: ["DataEngine"] updated: 2026-10-28T02:14:59Z published: 2026-10-28T02:14:59Z canonical: "docs.vastdata.com/dev-dataengine-integrations-dead-letter-topic" --- > ## Documentation Index > Fetch the complete documentation index at: https://docs.vastdata.com/llms.txt > Use this file to discover all available pages before exploring further. # Consume the dead-letter topic *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](/develop/docs/dev-dataengine-concepts-architecture#retries-and-the-dead-letter-topic), and how to set `retries` is in [pipelines as code](/develop/docs/dev-dataengine-guides-pipelines-as-code#retries-and-the-dead-letter-topic). Kafka clients, consumer groups and topic settings are covered in the [VAST Event Broker section](/develop/docs/dev-event-broker-guides-consuming). Commands use the environment variables from the [DataEngine quickstart](/develop/docs/dev-dataengine-quickstart#before-you-start), 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](/develop/docs/dev-dataengine-guides-pipelines-as-code#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](/develop/docs/dev-event-broker-operations-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](/develop/docs/dev-dataengine-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 [--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": "", "eventTime": "2026-10-08T14:11:24.228703Z", "eventName": "ObjectCreated:Put", "userIdentity": {"principalId": ""}, "requestParameters": {"sourceIPAddress": ""}, "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": ""}, "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](/develop/docs/dev-dataengine-reference-events#element-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](/develop/docs/dev-event-broker-guides-consuming#consume-in-a-group). - **Filter on `ce_source` or `ce_knativeerrordest`** to 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](/develop/docs/dev-event-broker-guides-topics#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 """ 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](/develop/docs/dev-event-broker-guides-producing#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_id`s 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](/develop/docs/dev-dataengine-concepts-architecture#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](/develop/docs/dev-dataengine-concepts-architecture#retries-and-the-dead-letter-topic): which failures are retried and dead-lettered. - [Pipelines as code: set the retries](/develop/docs/dev-dataengine-guides-pipelines-as-code#set-the-retries) - [Events](/develop/docs/dev-dataengine-reference-events): the attributes in the record's headers. - VAST Event Broker: [consuming](/develop/docs/dev-event-broker-guides-consuming), [producing](/develop/docs/dev-event-broker-guides-producing), [topics](/develop/docs/dev-event-broker-guides-topics) and [security](/develop/docs/dev-event-broker-operations-security). - [Integrations](/develop/docs/dev-dataengine-integrations-index)