A dead-letter queue (DLQ) holds records a stream processor cannot process. Sending problematic data to a separate MongoDB collection lets the main stream continue while the records remain available for inspection, correction, reprocessing, or deletion. section 15.7 section 15.7.2
In Atlas Stream Processing, set up the DLQ alongside the processor:
- Define a pipeline with a $source and an output stage. Add $validate when incoming documents must meet a schema. section 15.6.4 section 15.7.2
- Pass a dlq configuration naming an existing registered connection, database, and collection to sp.createStreamProcessor; then start the processor. The DLQ is optional and is stored in an Atlas database cluster. section 15.7.2
- For validation failures, set $validate.validationAction to "dlq". Its default is "discard"; the "dlq" action logs nonconforming documents while discarding them, without guarantees. section 15.5.2
Sales-record validation
A processor reads the Kafka topic dailySales, checks that productId, quantity, and price meet the required conditions, and writes valid records to validatedSales. Records that fail validation go to the DLQ for later review or reprocessing. section 15.6.4
Listing 15.8 DLQ configuration
// Set up a dead letter queue (DLQ) configuration ➥using an existing connection let dlqConfig = { dlq: { connectionName: " mongodb-in-action-connection ", ➥// name of the connection in the connection registry db: "ErrorLogs", // target database for the DLQ coll: "TransactionErrors" ➥// collection to store failed records } } // Create a basic stream processing pipeline let inputSource = {$source: {$kafka: {...}}} // input from a Kafka stream let outputSink = {$merge: {...}} // output data to a MongoDB collection let pipeline = [inputSource, outputSink] // Initialize a stream processor with a pipeline and DLQ setup sp.createStreamProcessor("TransactionProc", pipeline, dlqConfig) // Start the processor; any problematic data is sent to the DLQ sp.TransactionProc.start()
The configuration example names ErrorLogs.TransactionErrors as the DLQ collection and passes that configuration when creating a processor with a Kafka input and MongoDB output. section 15.7.2
Query and monitor the DLQ collection to track the rate, size, and types of failures. Atlas UI alerts can flag problems; a capped collection can limit growth, and an Atlas Trigger can support automatic reprocessing. Choose retention and handling that fit the need to investigate failed records. section 15.7.2
