Error Handling

ZephFlow provides robust error handling with Dead Letter Queue support (S3 and GCS) for capturing and analyzing failed records, ensuring data integrity throughout transformation pipelines.

Introduction

Error handling is a critical aspect of any data processing pipeline. ZephFlow provides robust mechanisms to handle errors that occur during record processing, ensuring that your pipelines remain resilient and your data is not lost when issues arise.

This document covers how ZephFlow handles errors, how to configure error handling options, and best practices for managing errors in your data pipelines.

Error Handling Overview

When an record flows through a ZephFlow pipeline, various errors can occur during processing. For example:

• Invalid data formats
• Failed parsing operations
• Type conversion errors
• Failed assertions or validations
• Errors in function execution

When an error occurs, ZephFlow marks the affected record as an "error record" but continues processing other records. This ensures that a single error doesn't cause the entire pipeline to break or lose other valid records.

Error Handling Modes

ZephFlow provides the following error handling mechanism:

Standalone Process Mode

When running ZephFlow in a standalone process (e.g., with the CLI starter), error records can be sent to a Dead Letter Queue (DLQ). The DLQ serves as a storage location for records that couldn't be processed successfully, allowing you to:

• Investigate the cause of failures
• Potentially reprocess the records after fixing the issue
• Keep a record of problematic data for analysis

To use this mode, you need to configure a Dead Letter Queue in the jobContext section of your DAG definition.

Dead Letter Queue Configuration

Configuring the DLQ

To configure a Dead Letter Queue, add a dlqConfig block to the jobContext section of your DAG definition. ZephFlow supports S3 and GCS (Google Cloud Storage) as destinations for dead letter records.

Specify the type field to select the DLQ backend:

S3 DLQ:

jobContext:
  dlqConfig:
    type: "s3"
    region: "us-west-2"
    bucket: "my-error-records-bucket"
    batchSize: 100
    flushIntervalMillis: 30000

GCS DLQ:

jobContext:
  dlqConfig:
    type: "gcs"
    bucket: "my-error-records-bucket"
    serviceAccountJson: "<paste the service account JSON here>"
    batchSize: 100
    flushIntervalMillis: 30000

DLQ Configuration Parameters

S3 Parameters

Parameter	Description
region	AWS region where the S3 bucket is located
bucket	Name of the S3 bucket to store error records
accessKeyId	AWS access key ID for authentication (optional; uses default credential chain if omitted)
secretAccessKey	AWS secret access key for authentication (optional; uses default credential chain if omitted)
s3EndpointOverride	Custom S3 endpoint URL (optional; useful for S3-compatible storage or local testing)
batchSize	Number of error records to accumulate before writing to S3 (triggers a flush when reached)
flushIntervalMillis	Maximum time in milliseconds to wait before writing accumulated records to S3 (ensures records are written even if batch size isn't reached)

GCS Parameters

Parameter	Description
bucket	Name of the GCS bucket to store error records
serviceAccountJson	GCP service account JSON credentials for authentication
batchSize	Number of error records to accumulate before writing to GCS (triggers a flush when reached)
flushIntervalMillis	Maximum time in milliseconds to wait before writing accumulated records to GCS (ensures records are written even if batch size isn't reached)

Dead Letter Format

When an error record is sent to the DLQ, it's stored using the Avro format with the following schema:

{
  "namespace": "io.fleak.zephflow.lib.deadletter",
  "type": "record",
  "name": "DeadLetter",
  "fields": [
    {
      "name": "processingTimestamp",
      "type": "long",
      "logicalType": "timestamp-millis"
    },
    {
      "name": "key",
      "type": [
        "null",
        "bytes"
      ],
      "default": null
    },
    {
      "name": "value",
      "type": [
        "null",
        "bytes"
      ],
      "default": null
    },
    {
      "name": "metadata",
      "type": [
        "null",
        {
          "type": "map",
          "values": "string"
        }
      ],
      "default": null
    },
    {
      "name": "errorMessage",
      "type": "string"
    },
    {
      "name": "nodeId",
      "type": "string"
    }
  ]
}

The dead letter record includes:

• processingTimestamp: When the error occurred
• key: Original record key (if applicable)
• value: Original record value
• metadata: Additional contextual information about the error
• errorMessage: Description of the error that occurred
• nodeId: The ID of the pipeline node where the error occurred (a top-level field, not part of metadata)

Integration with ZephFlow Pipelines

Here's a complete example of configuring error handling in a ZephFlow pipeline:

# Configure DLQ
jobContext:
  dlqConfig:
    type: "s3"
    region: "us-west-2"
    bucket: "error-events"
    batchSize: 50
    flushIntervalMillis: 60000  # 1 minute

# Create a flow with error-prone operations
dag:
  - id: "kafka_source"
    commandName: "kafkasource"
    config:
      broker: "kafka:9092"
      topic: "input-topic"
      groupId: "consumer-group"
      encodingType: "JSON_OBJECT"
    outputs:
      - "parse_value"

  - id: "parse_value"
    commandName: "eval"
    config:
      # Could fail if string_value is not a number
      expression: dict_merge($, dict(int_value=parse_int($.string_value)))
    outputs:
      - "check_value"

  - id: "check_value"
    commandName: "assertion"
    config:
      # Will mark records as errors if condition is not met
      expression: $.int_value > 0
    outputs:
      - "kafka_sink"

  - id: "kafka_sink"
    commandName: "kafkasink"
    config:
      broker: "kafka:9092"
      topic: "output-topic"
      encodingType: "JSON_OBJECT"
    outputs: [ ]

Dead Letter Enhancements

File Path Prefix

When jobContext.otherProperties.DATA_KEY_PREFIX is set, it is prepended to every dead letter object key (leading and trailing / are stripped), making it easier to organize DLQ files across different jobs or environments:

jobContext:
  otherProperties:
    DATA_KEY_PREFIX: "my-job/prod"
  dlqConfig:
    type: "s3"
    region: "us-west-2"
    bucket: "error-events"
    batchSize: 100
    flushIntervalMillis: 30000

Dead letters are then written under my-job/prod/dead-letters/dt=<yyyy-MM-dd>/hr=<HH>/ instead of dead-letters/dt=<yyyy-MM-dd>/hr=<HH>/.

Node ID in Error Records

Every dead letter record carries the ID of the node that produced it in the top-level nodeId field (not in metadata), helping you quickly identify which node in the pipeline caused the failure.

Raw Data Sampler

The raw data sampler periodically captures raw input data and writes it to the DLQ destination for debugging and inspection purposes. This is useful for verifying what data is entering your pipeline without modifying the pipeline itself.

Configuration

The sampler is configured via the DLQ config (both S3 and GCS support this):

S3 example:

jobContext:
  dlqConfig:
    type: "s3"
    region: "us-west-2"
    bucket: "my-error-records-bucket"
    batchSize: 100
    flushIntervalMillis: 30000
    rawDataSampleIntervalMs: 60000  # Sample raw data every 60 seconds (default)

GCS example:

jobContext:
  dlqConfig:
    type: "gcs"
    bucket: "my-error-records-bucket"
    serviceAccountJson: "<paste the service account JSON here>"
    batchSize: 100
    flushIntervalMillis: 30000
    rawDataSampleIntervalMs: 60000  # Sample raw data every 60 seconds (default)

Parameter	Description	Default
rawDataSampleIntervalMs	Interval in milliseconds between raw data samples	60000

Use Assertions for Data Validation

The assertion node is a powerful tool for data validation in ZephFlow. Unlike the filter node (which simply drops non-matching records), the assertion node marks records that don't meet the specified condition as error records, which can then be routed to your DLQ.

# Validate that required fields exist and have valid values
- id: "validate_record"
  commandName: "assertion"
  config:
    expression: $.user_id != null and $.timestamp != null and $.amount > 0
  outputs:
    - "next_node"