Skip to main content
Last updated on

Message blocking and acknowledgment

The Zerobus Ingest SDKs offer several methods for ingesting a record, which trade off throughput against how much durability confirmation you get back. This page explains each method and when to block on durability. To react to acknowledgments asynchronously instead of blocking, see Acknowledgment callbacks.

The examples on this page use the Python SDK. For the exact timeout and configuration options each method accepts (including their defaults and units), see the Zerobus SDK repository. The other language SDKs expose equivalent options.

What is an offset?​

Every record you ingest is assigned an offset: its position in the stream. The offset is how you refer to a specific record when you want to confirm it was durably written. Zerobus Ingest provides at-least-once delivery guarantees, and waiting on an offset is how a client confirms that guarantee for a given record.

Confirming an offset means the record is durable, not that it is queryable in the Delta table yet. Zerobus Ingest materializes durable records into the table shortly afterward. For latency figures, see Latency.

Ingestion methods​

The SDKs provide two ways to ingest a record. (Method names below are from the Python SDK. Other SDKs expose equivalent methods.)

Method

Returns

Use it when

Offset-based, ingest_record_offset()

The record's offset, after the record is queued on the stream.

Recommended default. You want to enqueue records in order and optionally confirm durability later by waiting on an offset.

Future-based, ingest_record()

A RecordAcknowledgment you can wait on.

Deprecated. Prefer offset-based for better performance.

Method

Returns

Use it when

Offset-based, ingest_record_offset()

The record's offset, after the record is queued on the stream.

Recommended default. You want to enqueue records in order and optionally confirm durability later by waiting on an offset.

Future-based, ingest_record()

A RecordAcknowledgment you can wait on.

Deprecated. Prefer offset-based for better performance.

ingest_record_offset() submits the record and returns its offset once the record is queued on the stream. The call runs on your calling thread, so records are enqueued in the order you call the method, and the returned offset lets you confirm durability later with wait_for_offset(). This is the recommended default for most producers, and it's the method used in the Use Zerobus Ingest examples.

Future-based (deprecated)​

ingest_record() returns a RecordAcknowledgment object that you can wait on for durability. It is deprecated in favor of the offset-based method, which performs better. Use it only for existing code that hasn't migrated yet.

Record-by-record vs. batch ingestion​

Each ingestion method has a batch variant (for example, ingest_records_offset()) that submits a list of records in one call. Batching is more efficient than individual calls for bulk ingestion.

For JSON and Protocol Buffers (protobuf), a batch commits atomically: either every record in the batch is accepted and made durable, or the whole batch is rejected. Zerobus Ingest does not perform partial uploads or partial acknowledgment for these formats, so your table never contains a partial batch. A batch that fails validation (for example, a schema mismatch) fails fast, before it touches the table, rather than landing some records and dropping others.

Because a JSON or protobuf batch is sent as a single message, the 10 MB maximum message size applies to both a single record and a whole batch: all records in a batch together must fit within 10 MB. Size your batches to stay under that limit. See Record size.

Arrow Flight batches are the exception​

Apache Arrow Flight ingestion does not follow the all-or-nothing, single-message model above. An Arrow batch can be much larger than a JSON or protobuf batch, and the Arrow Flight path splits a large batch into smaller transport messages that are sent and acknowledged individually rather than as one atomic unit. As a result:

  • The 10 MB per-message limit that applies to JSON and protobuf batches does not apply to an Arrow batch in the same way. A large Arrow batch is divided into transport messages instead of being rejected for size.
  • Durability is confirmed at the transport-message granularity, so a very large logical batch can be partially durable if a failure occurs partway through, rather than committing all-or-nothing.

ingest_batch() still returns a single logical offset for the batch you submitted, and wait_for_offset() on that offset completes only after every transport message that makes up the batch has been acknowledged. For the full Arrow Flight model, batching guidance, and recovery of unacknowledged data, see Use Arrow Flight with Zerobus Ingest.

When should you block on a message?​

Blocking on an offset trades throughput for a stronger per-record durability guarantee in your client code. Choose based on your workload:

  • Don't block: the right default for high-volume streaming, where you care about sustained throughput and can confirm durability in aggregate (for example, at stream close or through an acknowledgment callback). Most producers should start here.
  • Block on an offset: consider this when your application must know that a specific record is durable before it takes another action. For example:
    • You are about to delete or acknowledge the source of the data (a queue message, a file, an upstream cursor) and must not lose it if ingestion fails.
    • You are ingesting in checkpoints or transactional boundaries and need each checkpoint durable before advancing.
    • You are doing low-volume, high-value writes where per-record confirmation matters more than throughput.

Do not block on every record in a high-throughput loop. That serializes your producer on a round-trip to the server for each record and sharply reduces throughput. Instead, Databricks recommends ingesting a large chunk of records and then confirming durability once for the whole chunk. You have two ways to do that: wait on the latest offset, or flush the stream. Blocking per individual record should be reserved for the specific cases above where a single record must be confirmed before the next action.

Wait for an offset​

wait_for_offset() blocks until Zerobus Ingest confirms the record at that offset is durably written, or until it times out. Use it to confirm a specific point in the stream, most commonly the last record of a chunk. Ingest the chunk, keep the final offset the loop returns, and wait on that one offset instead of waiting after every record:

Python
from zerobus.sdk.sync import ZerobusSdk
from zerobus.sdk.shared import TableProperties

sdk = ZerobusSdk(SERVER_ENDPOINT, DATABRICKS_WORKSPACE_URL)

table_properties = TableProperties("main.default.air_quality")
stream = sdk.create_stream(CLIENT_ID, CLIENT_SECRET, table_properties)

try:
    last_offset = 0
    for row in records:
        last_offset = stream.ingest_record_offset(row)

    # Block until everything up to the last record of the chunk is durable
    stream.wait_for_offset(last_offset)
    print("Chunk durably written.")
finally:
    stream.close()

Flush the stream​

flush() blocks until all records you have ingested so far are durably written, then returns. Unlike wait_for_offset(), you don't track an offset: flush waits on everything pending on the stream. It does not close the stream, so you can keep ingesting afterward.

Python
try:
    for row in records:
        stream.ingest_record_offset(row)

    # Block until every pending record is durable
    stream.flush()
    print("All ingested records durably written.")
finally:
    stream.close()

wait_for_offset vs. flush​

Both confirm durability for a chunk. Choose based on what you're confirming:

  • Use wait_for_offset(offset) when you want to confirm up to a specific record, for example a checkpoint boundary, while other records may still be in flight behind it.
  • Use flush() when you want to confirm all pending records are durable before you move on, for example at the end of a batch, before advancing an upstream cursor, or before shutting down. flush() is governed by a configurable flush timeout.

close() flushes and closes the stream, so records are always made durable on a graceful shutdown. Always call it in a finally block.

React to acknowledgments asynchronously​

If instead of blocking you want to react to durability confirmations and errors as they arrive, while your producer keeps pushing at full speed, register an acknowledgment callback on the stream. Callbacks are a separate feature from the blocking calls on this page. See Acknowledgment callbacks.

On this page