Recovery and retry patterns

This deep dive explains how to build a resilient Zerobus Ingest client in Lakeflow Connect. It covers the SDK's built-in recovery, how errors surface, and how to rescue unacknowledged records when a stream fails permanently. Method and option names below are from the Python SDK. Other SDKs expose equivalents.

Built-in recovery

The Zerobus SDKs recover automatically from transient failures. Recovery is triggered when the stream hits a retriable error, typically a timeout or a network breakage, or when the stream receives a graceful shutdown signal. Recovery is on by default, and you tune it through stream configuration options:

Option Description
recovery Enable automatic stream recovery.
recovery_timeout_ms Timeout for a recovery operation.
recovery_backoff_ms Delay between recovery attempts.
recovery_retries Maximum number of recovery attempts.

For most workloads, the defaults are a good starting point, and you don't need to write your own reconnect loop for transient issues. When you use an SDK, OAuth tokens are also refreshed automatically on stream creation and recovery, so there's nothing for your client to manage. The exception is the REST API, where your client fetches and refreshes the OAuth token itself. See Use Zerobus Ingest. For the default values and units of these options, see the Zerobus SDK repository.

Errors and retries

The SDK automatically retries transient errors, such as network issues or temporary server errors, through its built-in recovery. Failures it can't recover from, such as invalid credentials or a missing table, surface as ZerobusException. Catch ZerobusException to handle a failure, then decide whether to fix the underlying cause, recover on a new stream, or stop.

from zerobus.sdk.shared import ZerobusException

try:
    stream.ingest_record_offset(record)
except ZerobusException as e:
    # Handle the failure: log it, fix the cause, recover on a new stream, or stop.
    ...

For the full list of error codes, see Zerobus Ingest error handling.

Recovering unacknowledged records

When a stream fails permanently, after the SDK's automatic recovery is exhausted, records that were submitted but not yet acknowledged by the server are still held by the client in the in-flight buffer described in Asynchronous communication. Retrieve them so you don't lose data:

  • get_unacked_records() returns the unacknowledged records as raw bytes.
  • get_unacked_batches() returns unacknowledged batches (each a list of records) for batch retry logic.

Records come back in their serialized form: decode JSON with json.loads(record.decode('utf-8')), or deserialize Protocol Buffers (protobuf) with your message type. Save them, or replay them on a fresh stream.

Recover a stream after permanent failure

The SDK automatically handles retries for transient errors. Enqueue, flush, and close failures all surface as ZerobusException. get_unacked_records() and recreate_stream() succeed only after the stream has already closed, which a terminal failure does. An enqueue failure leaves the stream active, so those calls fail; in that case, raise the original error and keep the stream. recreate_stream() re-queues the records that were already accepted; it does not retry a payload that failed to enqueue.

from zerobus.sdk.shared import ZerobusException

try:
    for i in range(10000):
        stream.ingest_record_offset(record)
    stream.flush()
except ZerobusException as e:
    print(f"Ingestion failed: {e}")
    try:
        unacked = list(stream.get_unacked_records())
    except ZerobusException:
        raise e
    print(f"{len(unacked)} previously queued records were unacknowledged.")
    try:
        new_stream = sdk.recreate_stream(stream)
        try:
            new_stream.flush()
        finally:
            new_stream.close()
    except ZerobusException:
        raise e
else:
    stream.close()

Use get_unacked_batches() to inspect the original batch grouping after the stream closes:

unacked_batches = list(stream.get_unacked_batches())
print(f"{len(unacked_batches)} batches remain unacknowledged")

Handling duplicates on replay

Zerobus Ingest provides at-least-once delivery, not exactly-once, so replaying rescued records, or any retry, can write a record more than once. If your workload can't tolerate duplicates, deduplicate the data in the lakehouse:

  • Include a stable unique identifier on each record (for example, a source-assigned event ID or a natural key).
  • Deduplicate on read or during downstream processing. For example, use a MERGE INTO that matches on the identifier, or a windowed ROW_NUMBER() in a transformation.

Because records on a stream are committed in order, a monotonically increasing sequence number also works well as the deduplication key.

Flushing and graceful close

  • flush() waits for the server to acknowledge the records you've submitted as durable, without closing the stream. Call it when you need a durability checkpoint mid-stream.
  • close() flushes and closes the stream gracefully, waiting for pending records to be acknowledged as durable before it returns. Use it for a graceful shutdown, not to recover from a stream failure. When a stream has failed, rescue unacknowledged records instead, as shown in Recover a stream after permanent failure.

For confirming durability of a specific record rather than the whole stream, see Message blocking and acknowledgment.

Recovering data from the durable fallback location

If a breaking change is made to your target table after Zerobus Ingest has made your data durable but before it can publish, Zerobus Ingest writes that data as Parquet files to a fallback directory under your table's storage root, rather than dropping it. See Durable fallback location.

How to tell data has been written there: the fallback directory is _zerobus/table_rejected_parquets/, relative to the table's physical root storage location. If ingestion continued but rows are missing from the table after a table change, check that directory for Parquet files.

Reprocess fallback data into the table

Once you have fixed the cause (typically by aligning the table schema with what your producers send), reprocess the fallback Parquet files into the target table. The files are standard Parquet in the table's storage location, so you can load them with COPY INTO:

  1. Resolve the schema mismatch. Evolve the target table (or your producer's schema) so the fallback records fit. See Schema management.

  2. Inspect the fallback data before loading. Point a query at the fallback path to confirm what's there and that it now matches the table:

    SELECT * FROM parquet.`<table-storage-root>/_zerobus/table_rejected_parquets/` LIMIT 10;

  3. Load the files with COPY INTO. COPY INTO is idempotent: it tracks the files it has already loaded, so re-running it won't double-load the same fallback files:

    COPY INTO <catalog>.<schema>.<table>
FROM '<table-storage-root>/_zerobus/table_rejected_parquets/'
FILEFORMAT = PARQUET
COPY_OPTIONS ('mergeSchema' = 'false');

  4. Verify the expected row counts landed, then, once you've confirmed the data is in the table, clean up the fallback directory if you no longer need it.

Because Zerobus Ingest is at-least-once, records that were both published to the table and written to the fallback location could be loaded twice. If duplicates matter, deduplicate as described in Handling duplicates on replay. For continuous or automated reprocessing, you can point Auto Loader at the fallback path instead of running COPY INTO manually.

Note

This procedure is a general first-pass approach using standard Delta tooling on the fallback Parquet files. Validate it against your table and storage configuration before relying on it in production.

  • Last updated on