Skip to main content
Last updated on

Acknowledgment callbacks

An acknowledgment callback lets your client react to record acknowledgments and errors asynchronously, without blocking your producer loop. As records become durable, or fail, Zerobus Ingest invokes your callback in the background, so you can track progress and update metrics without slowing your producer, and learn about failures as soon as they happen.

This differs from waiting on an offset or flushing: those are blocking calls where your code waits for durability inline. A callback is not a blocking call. It's a handler the SDK invokes for you when acknowledgments arrive.

Acknowledgment callbacks are supported for JSON and Protocol Buffers (protobuf) SDK streams. Arrow Flight streams do not support callbacks; to confirm durability on an Arrow stream, use wait_for_offset() or flush(). See Use Arrow Flight with Zerobus Ingest.

Method and type names below are from the Python SDK. Other Zerobus SDKs expose acknowledgment callbacks where supported, using equivalent constructs in each language.

How callbacks work​

You define a callback by subclassing AckCallback and implementing two methods:

  • on_ack(offset: int): called when a submission (a record or a batch) is successfully acknowledged as durable by the server. The offset identifies the acknowledged submission.
  • on_error(offset: int, error_message: str): called when a submission encounters an error. on_error is optional. Implement it to handle or log failures.

The callback is invoked once for each submitted record or batch when its logical offset is acknowledged or fails, so it's a running signal of ingestion progress across the stream.

Your callback methods run on the SDK's background threads, so invoking them doesn't block your producer. Keep them fast and non-blocking. What to do about a failure is your client's responsibility: log, alert, retry, or stop. Some errors are terminal, and if on_error reports that the stream has failed permanently, you must recover on a new stream. See Recovery and retry patterns.

Configure a callback​

You attach a callback to a stream by passing an instance of your AckCallback subclass as the ack_callback option in StreamConfigurationOptions when you create the stream. The callback then applies to every record ingested on that stream.

Python
from zerobus.sdk.sync import ZerobusSdk
from zerobus.sdk.shared import AckCallback, StreamConfigurationOptions, TableProperties

class MyAckCallback(AckCallback):
    def on_ack(self, offset: int) -> None:
        print(f"Record acknowledged at offset: {offset}")

    def on_error(self, offset: int, error_message: str) -> None:
        print(f"Error at offset {offset}: {error_message}")

sdk = ZerobusSdk(SERVER_ENDPOINT, DATABRICKS_WORKSPACE_URL)

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

try:
    for row in records:
        stream.ingest_record_offset(row)
finally:
    stream.close()

With the callback registered, you don't wait inline. on_ack fires as each record is confirmed durable, and on_error fires if a record fails.

Callbacks vs. blocking​

Callbacks and the blocking calls solve different problems, and you can use them together:

  • Use an acknowledgment callback to react to durability confirmations and errors as they happen, asynchronously, while sustaining high throughput. Good for progress tracking, metrics, and error logging.
  • Use wait_for_offset() or flush() when your code must block until a specific record, or all pending records, are durable before proceeding.
On this page