Skip to content

rust

Directory actions

More options

Directory actions

More options

Latest commit

 

History

History

rust

README.md

Zerobus Rust SDK

A high-performance Rust client for streaming data ingestion into Databricks Delta tables using the Zerobus service.

Table of Contents

Overview

The Zerobus Rust SDK provides a robust, async-first interface for ingesting large volumes of data into Databricks Delta tables. It abstracts the complexity of the Zerobus service and handles authentication, retries, stream recovery, and acknowledgment tracking automatically.

What is Zerobus? See the project overview for details on the Zerobus service.

Features

  • Async/Await Support - Built on Tokio for efficient concurrent I/O operations
  • Automatic OAuth 2.0 Authentication - Seamless token management with Unity Catalog
  • Built-in Recovery - Automatic retry and reconnection for transient failures
  • High Throughput - Configurable inflight record limits for optimal performance
  • Batch Ingestion - Ingest multiple records at once with all-or-nothing semantics for maximum throughput
  • Flexible Serialization - Support for both JSON (simple) and Protocol Buffers (type-safe) data formats
  • Type Safety - Protocol Buffers ensure schema validation at compile time
  • Schema Generation - CLI tool to generate protobuf schemas from Unity Catalog tables
  • Flexible Configuration - Fine-tune timeouts, retries, and recovery behavior
  • Graceful Stream Management - Proper flushing and acknowledgment tracking
  • Acknowledgment Callbacks - Receive notifications when records are acknowledged or encounter errors
  • Multiplexed Streams (Beta) - Raise aggregate gRPC throughput when one stream is the bottleneck while the SDK manages sub-stream routing and lifecycles
  • Arrow Flight Ingestion (opt-in) — Stream Apache Arrow RecordBatch data directly to Zerobus using Arrow Flight's gRPC transport. Enable with features = ["arrow-flight"]; see examples/arrow/.
  • Avro Ingestion (Beta, opt-in) — Ingest Avro records on ephemeral streams. Enable with features = ["avro"], select via .avro(schema_json). Build records as AvroValue and let the stream encode them against the writer schema (AvroRecord), or ingest pre-encoded datums via AvroBytes. See examples/avro/. Note: the avro feature requires Rust 1.85 (via apache-avro); default builds are unaffected. Feature in development.
  • Zeroparser (opt-in) — Zero-copy, single-pass protobuf parser for runtime-known schemas. Enable with features = ["zeroparser"]; see sdk/src/zeroparser/README.md.

Installation

Add the SDK to your Cargo.toml:

cargo add databricks-zerobus-ingest-sdk
cargo add async-trait tonic
cargo add prost prost-types
cargo add tokio --features macros,rt-multi-thread

Why these dependencies?

  • databricks-zerobus-ingest-sdk - The SDK itself
  • async-trait - Required when implementing the async HeadersProvider trait
  • tonic - Provides transport::Endpoint for custom TlsConfig implementations
  • prost and prost-types - Required for encoding your data to Protocol Buffers and loading schema descriptors
  • tokio - Async runtime required for running async functions (the SDK is fully async)

What's in the crates.io package? The published crate contains only the core Zerobus ingestion SDK. Tools for schema generation (tools/generate_files) and working examples (examples/) are only available in the GitHub repository. You'll need to clone the repo to generate protobuf schemas from your Unity Catalog tables.

For Local Development

Clone the repository and use a path dependency:

git clone https://github.com/databricks/zerobus-sdk.git
cd your_project

Then in your Cargo.toml:

[dependencies]
databricks-zerobus-ingest-sdk = { path = "../zerobus-sdk/rust/sdk" }
prost = "0.14"
prost-types = "0.14"
tokio = { version = "1.52", features = ["macros", "rt-multi-thread"] }

Quick Start

JSON and Protocol Buffers share one stream API, with two ingestion methods. Avro (Beta, opt-in) rides the same stream API. Arrow Flight is a separate columnar API.

Serialization:

  • JSON (Recommended for getting started): Simpler approach using JSON strings, no schema generation required
  • Protocol Buffers (Recommended for production): Type-safe approach with schema validation at compile time
  • Avro (Beta, opt-in): Row-oriented encoding against a writer schema. Enable with features = ["avro"]; feature in development

Ingestion Methods:

  • Single-record (ingest_record_offset): Ingest records one at a time with per-record acknowledgment
  • Batch (ingest_records_offset): Ingest multiple records at once with all-or-nothing semantics for higher throughput

See examples/README.md for detailed setup instructions and examples for all combinations.

Repository Structure

zerobus_rust_sdk/
├── sdk/                                # Core SDK library
│   ├── src/
│   │   ├── lib.rs                      # Crate root and public re-exports
│   │   ├── sdk.rs                      # Main SDK client implementation
│   │   ├── builder/                    # Builder pattern for SDK initialization
│   │   ├── stream/                     # Ingestion streams by transport
│   │   │   ├── mod.rs                  # Stream module entry point
│   │   │   ├── grpc/                   # Proto/JSON gRPC stream internals
│   │   │   │   ├── mod.rs
│   │   │   │   ├── connection.rs       # Connection setup and management
│   │   │   │   ├── ingest.rs           # Record ingestion path
│   │   │   │   ├── sender.rs           # Outbound request sender
│   │   │   │   ├── receiver.rs         # Inbound response receiver
│   │   │   │   ├── acks.rs             # Acknowledgement tracking
│   │   │   │   ├── callback_handler.rs # Ack/error callback dispatch
│   │   │   │   ├── supervisor.rs       # Stream supervision and reconnect
│   │   │   │   ├── close.rs            # Graceful close handling
│   │   │   │   └── types.rs            # Shared gRPC stream types
│   │   │   └── arrow/                  # Arrow Flight stream (feature: arrow-flight)
│   │   │       ├── mod.rs              # Public stream API and lifecycle
│   │   │       ├── connection.rs       # Flight transport and request encoding
│   │   │       ├── acks.rs             # ACK processing and server rotation
│   │   │       ├── supervisor.rs       # Recovery, replay, and finalization
│   │   │       ├── batch.rs            # Pending batches and IPC helpers
│   │   │       ├── metadata.rs         # Flight request/ack metadata
│   │   │       ├── options.rs          # Arrow Flight stream options
│   │   │       ├── close.rs            # Close coordination and terminal finalization
│   │   │       └── README.md           # Arrow Flight internal architecture
│   │   ├── multiplexed_stream.rs       # Multiplexed stream implementation
│   │   ├── default_token_factory.rs    # OAuth 2.0 token handling
│   │   ├── token_cache.rs              # OAuth token caching
│   │   ├── errors.rs                   # Error types and retryable logic
│   │   ├── headers_provider.rs         # Trait for custom authentication headers
│   │   ├── callbacks.rs                # Ack/error callback traits
│   │   ├── client_warnings.rs          # Client-side warning diagnostics
│   │   ├── record_types.rs             # Record encoding types (JSON, proto, raw)
│   │   ├── schema.rs                   # Unity Catalog → proto/Arrow schema generation
│   │   ├── stream_configuration.rs     # Stream options
│   │   ├── stream_options.rs           # Shared stream configuration
│   │   ├── tls_config.rs               # TLS configuration strategies
│   │   ├── landing_zone.rs             # Inflight record buffer
│   │   ├── offset_generator.rs         # Logical offset tracking
│   │   ├── proxy.rs                    # HTTP proxy support
│   │   └── zeroparser/                 # Descriptor-driven protobuf parser (feature: zeroparser)
│   ├── zerobus_service.proto           # gRPC protocol definition
│   ├── build.rs                        # Build script for protobuf compilation
│   └── Cargo.toml
│
├── ffi/                                # C FFI bindings for other languages
│   ├── src/
│   │   ├── lib.rs                      # FFI entry point and exports
│   │   ├── sdk.rs                      # SDK lifecycle FFI
│   │   ├── stream.rs                   # Stream FFI
│   │   ├── builder.rs                  # Builder FFI
│   │   ├── proto_schema.rs             # Proto schema FFI
│   │   ├── arrow.rs                    # Arrow Flight FFI (feature: arrow-flight)
│   │   ├── common.rs                   # Shared FFI helpers
│   │   └── tests.rs                    # FFI unit tests
│   ├── zerobus.h                       # Generated C header
│   ├── cbindgen.toml                   # Header generation config
│   ├── build.rs                        # Build script for header generation
│   └── Cargo.toml
│
├── jni/                                # JNI bindings for Java SDK
│   ├── src/
│   │   ├── lib.rs                      # JNI entry point
│   │   ├── sdk.rs                      # SDK lifecycle JNI
│   │   ├── stream.rs                   # Stream JNI
│   │   ├── arrow_stream.rs             # Arrow Flight stream JNI (feature: arrow-flight)
│   │   ├── async_bridge.rs             # Async runtime bridge
│   │   ├── callbacks.rs                # Ack/error callback bridge
│   │   ├── class_cache.rs              # Cached JNI class/method handles
│   │   ├── errors.rs                   # Error mapping to Java exceptions
│   │   ├── options.rs                  # Stream/SDK option parsing
│   │   ├── runtime.rs                  # Tokio runtime management
│   │   └── test_helper.rs              # JNI test helpers
│   └── Cargo.toml
│
├── tools/
│   └── generate_files/                 # Schema generation CLI tool (package `tools`)
│       ├── src/
│       │   ├── main.rs                 # CLI entry point
│       │   ├── generate.rs             # Unity Catalog -> Proto conversion
│       │   └── token_factory.rs        # OAuth token factory for the CLI
│       ├── README.md                   # Tool documentation
│       └── Cargo.toml
│
├── examples/
│   ├── README.md                       # Examples documentation
│   ├── json/                           # JSON examples (single Cargo package)
│   │   ├── README.md
│   │   ├── Cargo.toml
│   │   ├── single.rs                   # JSON single-record example
│   │   └── batch.rs                    # JSON batch ingestion example
│   ├── proto/                          # Protocol Buffers examples (single Cargo package)
│   │   ├── README.md
│   │   ├── Cargo.toml
│   │   ├── single.rs                   # Protocol Buffers single-record example
│   │   ├── batch.rs                    # Protocol Buffers batch ingestion example
│   │   └── output/                     # Generated schema files (shared)
│   ├── arrow/                          # Arrow Flight example (feature: arrow-flight)
│   │   ├── README.md
│   │   ├── Cargo.toml
│   │   └── src/main.rs                 # Arrow `RecordBatch` ingestion example
│   └── avro/                           # Avro examples (excluded crate, feature: avro; Beta)
│       ├── README.md
│       ├── Cargo.toml
│       ├── single.rs                   # Avro single-record example
│       └── batch.rs                    # Avro batch ingestion example
│
├── tests/                              # Integration tests crate
│   ├── src/
│   │   ├── mock_grpc.rs                # Mock Zerobus gRPC server
│   │   ├── mock_arrow_flight.rs        # Mock Arrow Flight server
│   │   ├── rust_tests.rs               # Core SDK test suite
│   │   ├── proxy_tests.rs              # HTTP proxy tests
│   │   ├── arrow_tests.rs              # Arrow Flight test suite
│   │   ├── multiplexed_stream_tests.rs # Multiplexed stream test suite
│   │   └── utils.rs                    # Shared test utilities
│   ├── build.rs
│   └── Cargo.toml
│
├── third_party/
│   └── arrow-flight/                   # Vendored arrow-flight fork (slice-aware batch-split patch)
│       ├── src/                        # Fork sources (encode/decode/client/sql/…)
│       ├── Cargo.toml
│       ├── regen.sh                    # Re-sync script for the vendored fork
│       └── README.md
│
├── Cargo.toml                          # Workspace configuration
└── README.md                           # This file

Key Components

  • sdk/ - The main library crate containing all SDK functionality
  • ffi/ - C FFI bindings for building language wrappers (Go, C#, C++, etc.)
  • jni/ - JNI bindings for the Java SDK
  • tools/ - CLI tool for generating Protocol Buffer schemas from Unity Catalog tables
  • examples/ - Complete working examples demonstrating SDK usage
  • Workspace - Root Cargo.toml defines a Cargo workspace for unified builds

How It Works

JSON / Protocol Buffers Architecture Overview

The following diagram describes the JSON / Protocol Buffers stream (Avro, when enabled, uses this same stream). Arrow Flight uses a separate Flight request/response exchange and lifecycle state machine.

+-----------------+
|    Your App     |
+-----------------+
        | 1. stream_builder().build()
        v
+-----------------+
|   ZerobusSdk    |
| - Manages TLS   |
| - Creates       |
|   channels      |
+-----------------+
        | 2. Opens bidirectional gRPC stream
        v
+--------------------------------------+
|            ZerobusStream             |
| +----------------------------------+ |
| |           Supervisor             | | Manages lifecycle, recovery
| +----------------------------------+ |
|                  |                   |
|      +-----------+-----------+       |
|      v                       v       |
| +----------+          +----------+   |
| |  Sender  |          | Receiver |   | Parallel tasks
| |  Task    |          |  Task    |   |
| +----------+          +----------+   |
|      ^                       |       |
|      |                       v       |
| +----------------------------------+ |
| |          Landing Zone            | | Inflight buffer
| +----------------------------------+ |
+--------------------------------------+
            | 3. gRPC stream
            v
+-----------------------+
|      Databricks       |
|    Zerobus Service    |
+-----------------------+

JSON / Protocol Buffers Data Flow

  1. Ingestion - Your app calls stream.ingest_record_offset(data) or stream.ingest_records_offset(batch)
  2. Buffering - Records are placed in the landing zone with logical offsets
  3. Sending - Sender task sends records over gRPC with physical offsets
  4. Acknowledgment - Receiver task gets server ack; callers wait via stream.wait_for_offset(offset)
  5. Recovery - If connection fails, supervisor reconnects and resends unacked records

Arrow Flight Lifecycle

Arrow Flight uses a dedicated DoPut request/response exchange:

ingest_batch -> Flight encoder -> DoPut -> ack_up_to_records
  1. Build the stream with .arrow(schema).build_arrow().
  2. Queue application-sized RecordBatches with ingest_batch().
  3. Call flush() once at a durability boundary.
  4. Call close() to half-close and drain the active Flight exchange.

Each input RecordBatch receives one logical SDK offset even if Flight splits it into several wire messages. Durability is tracked through the cumulative ack_up_to_records watermark. Arrow Flight does not support ack_callback.

Recovery reconnects and replays only unacknowledged batch suffixes. After a failed close(), call get_unacked_batches() to inspect the retained work; the returned set can be empty when every record was already durable.

Telemetry (Beta)

Arrow Flight streams do not support ack_callback, but you can register a StatsExporter to receive per-batch telemetry:

use std::num::NonZeroUsize;
use databricks_zerobus_ingest_sdk::{channel_exporter, StreamStat};

let (exporter, mut stats_rx) = channel_exporter(NonZeroUsize::new(1024).unwrap());
let mut stream = sdk
    .stream_builder()
    .table("catalog.schema.table")
    .oauth("client-id", "client-secret")
    .arrow(schema)
    .stats_exporter(exporter)
    .build_arrow()
    .await?;

// Drain concurrently: the bounded channel drops events when full. recv() ends
// after all exporter clones are dropped and the queued events are consumed.
let telemetry = tokio::spawn(async move {
    while let Some(stat) = stats_rx.recv().await {
        if let StreamStat::BatchSent { offset, attempt, stats } = stat {
            // stats.approximate_wire_bytes: FlightData payload after IPC compression
            //   (excludes transport overhead and schema; does not measure socket writes)
            // stats.uncompressed_bytes: emitted IPC buffer bytes before compression
            // stats.records: rows in this transmission; offset: the batch's ingest offset
            // attempt: 0 = first send, >0 = retransmit of the unacked suffix after reconnect
            let _ = (offset, attempt, stats);
        }
    }
});

// Ingest in a loop, then flush() once — never wait per batch.
for batch in batches {
    stream.ingest_batch(batch).await?;
}
stream.flush().await?;
stream.close().await?;

// Release the stream's exporter. Drop any retained exporter clones too.
drop(stream);
telemetry.await?;

StreamStat events:

  • BatchSent { offset, attempt, stats } — a batch was encoded and yielded to the transport. This does not confirm network delivery. stats is a BatchStats with records, approximate_wire_bytes, and uncompressed_bytes. Emitted with the final data frame, including on retransmission, even if the batch is never acknowledged. Later cancellation or drop preserves the event; a transmission canceled before its final frame emits none. attempt is 0 on the first send and > 0 on a retransmit after a reconnect. Batches buffered during recovery still use 0 on their first send; a replay canceled before its first data frame does not advance the count.
  • BatchAcked { offset } — a batch was durably acknowledged. Pure durability signal; byte sizes are on BatchSent.
  • Reconnected { reason } — an automatic reconnect completed setup, queued replay, and published its sender. Replay events may precede it. reason is ReconnectReason::TransientFailure(err) (the failure that triggered the successful attempt, including an authentication rejection recovered by refreshing credentials) or ReconnectReason::ServerRotation (a routine server-requested rotation).

Each reported retry counts every frame it emits: the whole batch if no rows were acknowledged, otherwise only the unacknowledged suffix. Summing BatchSent events gives payload totals for reported completed transmissions. Incomplete transmissions emit no event. If IPC statistics cannot be read, the SDK logs a warning and omits that transmission's event; ingestion and attempt counting continue. The first event can therefore have attempt > 0; filtering for attempt == 0 can miss the batch entirely. These events do not guarantee an original-batch size sample for every offset.

The runnable Arrow example prints telemetry concurrently with ingestion and drains remaining events during shutdown.

Implement StatsExporter directly for custom routing. Its record runs inline on the stream's IO tasks, which may call it concurrently; keep it lightweight. Unwinding panics are caught and logged. channel_exporter counts events dropped when its channel is full or closed; flush() does not wait for the consumer to process queued telemetry.

Offsets and attempts are scoped to one stream instance; events carry no stream identifier. recreate_arrow_stream shares the exporter but starts offsets and attempts over, without emitting Reconnected. Build with separate exporters if you need to distinguish streams.

uncompressed_bytes counts the emitted IPC buffers before compression, including dictionary buffers when sent. It follows the encoder's treatment of slices and shared buffers and excludes IPC metadata, alignment padding, and compression length prefixes. It measures encoded payload, not retained heap memory.

For state ownership, replay, rotation, and concurrency invariants, see the Arrow Flight maintainer architecture guide.

Authentication Flow

The SDK uses OAuth 2.0 client credentials flow:

  1. SDK constructs authorization request with Unity Catalog privileges
  2. Sends request to {uc_endpoint}/oidc/v1/token with client credentials
  3. Token includes scoped permissions for the specific table
  4. Token is attached to gRPC metadata as Bearer token
  5. Tokens are cached per table and reused across connections until they near expiry (see Token Caching below)

Token Caching

OAuth tokens minted via Unity Catalog have a lifetime chosen by Unity Catalog (currently one hour) that the SDK cannot configure, while a single stream lives at most ~15 minutes. By default the SDK caches the token for each table on the ZerobusSdk instance and reuses it across stream creations and recoveries, refreshing it only once it nears the expiry the server reported (so it adapts to whatever lifetime UC returns). This avoids minting a fresh token on every stream and reduces load on the Unity Catalog token endpoint.

Caching applies to the built-in OAuth path (.oauth(...)) and to federated authentication (.federated_auth(...)). Both share one cache on the ZerobusSdk instance, but never share a cached token: each slot is keyed by auth scheme, the token's identity (OAuth client_id, or the federation identity / service principal client_id), and the table, so OAuth and federated tokens (and distinct identities) stay in separate slots. Tokens are shared only across streams created from the same ZerobusSdk, so reuse a single SDK instance rather than constructing a new one per stream. Custom HeadersProvider implementations manage their own caching.

SDK builder options tune connection and token behavior:

use databricks_zerobus_ingest_sdk::ZerobusSdk;
use std::time::Duration;

let sdk = ZerobusSdk::builder()
    .endpoint("https://<your-workspace>.zerobus.<region>.cloud.databricks.com")
    .unity_catalog_url("https://<your-workspace>.cloud.databricks.com")
    // Refresh a cached token once it is within 10 minutes of expiry (default: 5 minutes).
    .token_refresh_buffer(Duration::from_secs(600))
    // Or disable caching entirely to mint a fresh token per stream.
    // .token_cache_enabled(false)
    .build()?;
# Ok::<(), databricks_zerobus_ingest_sdk::ZerobusError>(())

When a proactive refresh fails, the SDK keeps serving the still-valid cached token rather than failing stream creation, including on non-retryable failures such as revoked credentials. The cached token is served until it actually expires, after which the next call mints a fresh one and any failure then surfaces. A proactive refresh is also capped at half the stream's recovery_timeout_ms, but only when the cached token has more life left than that cap, so a stalled endpoint falls back to the cached token in time. When too little of the token's life remains, the refresh runs unbounded instead, like a cold miss.

Custom Authentication

For advanced use cases, you can implement the HeadersProvider trait to supply your own authentication headers. This is useful for integrating with a different OAuth provider, using a centralized token caching service, or implementing alternative authentication mechanisms.

Note: The headers you provide must still conform to the authentication protocol expected by the Zerobus service. The default implementation, OAuthHeadersProvider, serves as the reference for the required headers (authorization and x-databricks-zerobus-table-name). This feature provides flexibility in how you source your credentials, not in changing the authentication protocol itself.

Example:

use databricks_zerobus_ingest_sdk::*;
use std::collections::HashMap;
use std::sync::Arc;
use async_trait::async_trait;

struct MyCustomAuthProvider;

#[async_trait]
impl HeadersProvider for MyCustomAuthProvider {
    async fn get_headers(&self) -> ZerobusResult<HashMap<&'static str, String>> {
        let mut headers = HashMap::new();
        // Custom logic to fetch and cache a token would go here.
        headers.insert("authorization", "Bearer <your-token>".to_string());
        headers.insert("x-databricks-zerobus-table-name", "<your-table-name>".to_string());
        Ok(headers)
    }
}

async fn example(sdk: ZerobusSdk) -> ZerobusResult<()> {
    let custom_provider = Arc::new(MyCustomAuthProvider {});
    let stream = sdk
        .stream_builder().table("catalog.schema.table")
        .headers_provider(custom_provider)
        .json()
        .build()
        .await?;
    Ok(())
}

Usage Guide

The SDK supports these approaches for data serialization:

  1. JSON - Simpler approach that uses JSON strings. No schema generation required, making it ideal for quick prototyping. See examples/README.md for a complete example.
  2. Protocol Buffers - Type-safe approach with schema validation at compile time. Recommended for production use cases. This guide focuses on the Protocol Buffers approach.
  3. Avro (Beta, opt-in) - Encode against a writer schema declared with .avro(schema_json). Enable with features = ["avro"]; feature in development. See Avro Stream.

For JSON-based ingestion, you can skip the schema generation step and directly pass JSON strings to ingest_record_offset().

1. Generate Protocol Buffer Schema (Protocol Buffers approach only)

Important Note: The schema generation tool and examples are only available in the GitHub repository. The crate published on crates.io contains only the core Zerobus ingestion SDK logic. To generate protobuf schemas or see working examples, clone the repository:

git clone https://github.com/databricks/zerobus-sdk.git
cd zerobus-sdk/rust

Use the included tool to generate schema files from your Unity Catalog table:

cd tools/generate_files

# For AWS
cargo run -- \
  --uc-endpoint "https://<your-workspace>.cloud.databricks.com" \
  --client-id "your-client-id" \
  --client-secret "your-client-secret" \
  --table "catalog.schema.table" \
  --output-dir "../../output"

# For Azure
cargo run -- \
  --uc-endpoint "https://<your-workspace>.azuredatabricks.net" \
  --client-id "your-client-id" \
  --client-secret "your-client-secret" \
  --table "catalog.schema.table" \
  --output-dir "../../output"

This generates three files:

  • {table}.proto - Protocol Buffer schema definition
  • {table}.rs - Rust structs with serialization code
  • {table}.descriptor - Binary descriptor for runtime validation

See tools/generate_files/README.md for supported data types and limitations.

See examples/README.md for more information on how to get OAuth credentials.

2. Initialize the SDK

Create an SDK instance using the builder pattern:

// For AWS
let sdk = ZerobusSdk::builder()
    .endpoint("https://<your-shard-id>.zerobus.<region>.cloud.databricks.com")
    .unity_catalog_url("https://<your-workspace>.cloud.databricks.com")
    .build()?;

// For Azure
let sdk = ZerobusSdk::builder()
    .endpoint("https://<your-shard-id>.zerobus.<region>.azuredatabricks.net")
    .unity_catalog_url("https://<your-workspace>.azuredatabricks.net")
    .build()?;

Note: The workspace ID is automatically extracted from the Zerobus endpoint. The https:// scheme is optional — if omitted, the SDK automatically prepends https://.

TLS Configuration

By default, the SDK uses SecureTlsConfig which enables TLS with the operating system's trusted CA certificates. For testing against a local http:// server, use NoTlsConfig (requires the testing feature):

cargo add databricks-zerobus-ingest-sdk --features testing
use databricks_zerobus_ingest_sdk::{ZerobusSdk, NoTlsConfig};
use std::sync::Arc;

let sdk = ZerobusSdk::builder()
    .endpoint("http://localhost:50051")
    .tls_config(Arc::new(NoTlsConfig))
    .build()?;

A .no_tls() shorthand on ZerobusSdkBuilder is also available behind the testing feature and is equivalent to the snippet above. Paired with StreamBuilder::no_auth() (also testing-gated, uses NoAuthHeadersProvider), it covers local testing against an unauthenticated plaintext server.

For custom certificate handling, implement the TlsConfig trait:

use databricks_zerobus_ingest_sdk::{TlsConfig, ZerobusResult};
use tonic::transport::Endpoint;

struct MyTlsConfig { /* ... */ }

impl TlsConfig for MyTlsConfig {
    fn configure_endpoint(&self, endpoint: Endpoint) -> ZerobusResult<Endpoint> {
        // Custom TLS configuration logic
        Ok(endpoint)
    }
}

3. Configure Authentication

The SDK handles authentication automatically. You just need to provide:

  • Client ID - Your OAuth client ID
  • Client Secret - Your OAuth client secret
  • Unity Catalog Endpoint - Passed to SDK constructor
  • Table Name - Included in table properties
let client_id = "your-client-id".to_string();
let client_secret = "your-client-secret".to_string();

See examples/README.md for more information on how to get these credentials.

External-IdP federation (e.g. Entra ID)

To authenticate with an external identity provider instead of a Databricks OAuth secret, use the federated_auth builder method. You provide an [IdpTokenSupplier] callback that returns the current external IdP token; the SDK exchanges it for a Zerobus-scoped Databricks token (RFC 8693 token exchange) and caches and refreshes it, exactly like the OAuth path.

use std::sync::Arc;

// Account-level federation: no Databricks service principal. The identity is
// synced into Databricks via Automatic Identity Management (SCIM). Pass `None`
// for the client_id. The shared token cache is partitioned by the supplier's
// identity, so if you drive multiple account-level identities from one SDK
// instance, give each its own supplier (clone one supplier to share its token).
let supplier =
    IdpTokenSupplier::new(Arc::new(|| Box::pin(async { get_idp_token().await })));
let stream = sdk
    .stream_builder()
    .table("catalog.schema.table")
    .federated_auth(supplier.clone(), None)
    .json()
    .build()
    .await?;

// Workload identity federation: a Databricks service principal with a client_id
// and no secret, with a federation policy attached. The cache keys by the
// service principal client_id.
let stream = sdk
    .stream_builder()
    .table("catalog.schema.table")
    .federated_auth(
        IdpTokenSupplier::new(Arc::new(|| Box::pin(async { get_idp_token().await }))),
        Some("<sp-client-id>".to_string()),
    )
    .json()
    .build()
    .await?;

The existing .oauth(...) and .headers_provider(...) paths are unchanged.

4. Create a Stream

Use the stream_builder() API to create a stream:

JSON Stream

let mut stream = sdk
    .stream_builder().table("catalog.schema.orders")
    .oauth(client_id, client_secret)
    .json()
    .max_inflight_requests(10_000)
    .recovery_timeout_ms(15_000)
    .recovery_backoff_ms(2_000)
    .recovery_retries(4)
    .build()
    .await?;

Protocol Buffers Stream

use std::fs;
use prost::Message;
use prost_types::{FileDescriptorSet, DescriptorProto};

// Load descriptor from generated files
fn load_descriptor(path: &str, file: &str, msg: &str) -> DescriptorProto {
    let bytes = fs::read(path).expect("Failed to read descriptor");
    let file_set = FileDescriptorSet::decode(bytes.as_ref()).unwrap();

    let file_desc = file_set.file.into_iter()
        .find(|f| f.name.as_deref() == Some(file))
        .unwrap();

    file_desc.message_type.into_iter()
        .find(|m| m.name.as_deref() == Some(msg))
        .unwrap()
}

let descriptor_proto = load_descriptor(
    "output/orders.descriptor",
    "orders.proto",
    "table_Orders",
);

let mut stream = sdk
    .stream_builder().table("catalog.schema.orders")
    .oauth(client_id, client_secret)
    .compiled_proto(descriptor_proto)
    .max_inflight_requests(10_000)
    .recovery_timeout_ms(15_000)
    .recovery_backoff_ms(2_000)
    .recovery_retries(4)
    .build()
    .await?;

Dynamic Protobuf Stream

When the table's schema is known only at runtime — for example a descriptor fetched from Unity Catalog or built in code with schema::descriptor_from_uc_columns — there is no compiled prost::Message type. Resolve the descriptor with message_descriptor, pass it to .dynamic_proto(descriptor), and fill records field-by-field with DynamicRecord:

use databricks_zerobus_ingest_sdk::{message_descriptor, ProtoBytes};
use databricks_zerobus_ingest_sdk::schema::{descriptor_from_uc_columns, UcColumn};

// Build the descriptor at runtime (a column's proto field number is `position + 1`).
let descriptor_proto = descriptor_from_uc_columns(&columns, "table_Orders")?;
let descriptor = message_descriptor(&descriptor_proto)?;

let mut stream = sdk
    .stream_builder().table("catalog.schema.orders")
    .oauth(client_id, client_secret)
    .dynamic_proto(descriptor)
    .build()
    .await?;

// Fill records field-by-field; `set()` validates the field name and type (the
// value must match the field's proto type, e.g. a BIGINT column takes an i64).
// `encode()` then checks proto2 required fields before producing the bytes.
for i in 0..100_000i64 {
    let mut record = stream.new_record()?; // bound to the stream's schema
    record.set("id", i)?.set("customer_name", "Alice Smith")?;
    let _offset = stream.ingest_record_offset(ProtoBytes(record.encode()?)).await?; // queue only — do NOT wait here
}
stream.flush().await?; // wait once for all pending acknowledgments

.dynamic_proto() takes a resolved MessageDescriptor. Get one from message_descriptor(&proto) or build it against your own prost_reflect::DescriptorPool.

On the wire this is identical to .compiled_proto(...); the difference is that records are built dynamically rather than from a generated struct. See the dynamic_proto module and the proto_dynamic_single example for details.

Multiplexed gRPC Stream

When you need more throughput than one gRPC stream can provide and do not require global ordering, select multiplexed mode after configuring an ordinary builder. The SDK opens all requested sub-streams, routes records round-robin, and manages recovery, flush, and close:

JSON streams: First migrate to compiled Protocol Buffers and measure throughput again before considering multiplexing.

let mut stream = sdk
    .stream_builder()
    .table("catalog.schema.orders")
    .oauth(client_id, client_secret)
    .compiled_proto(descriptor_proto)
    .max_inflight_requests(10_000)
    .multiplexed(4)
    .build()
    .await?;

for record in records {
    let _message_id = stream.ingest_record(record).await?;
}
stream.flush().await?;
stream.close().await?;

The stream count must be between 1 and 64. JSON, compiled protobuf, dynamic protobuf, and Avro (with the avro feature) are supported. Arrow Flight is not multiplexed. For maximum protobuf encoding throughput, prefer compiled protobuf—multiplexing addresses stream/network bottlenecks, not the reflection cost of constructing dynamic records.

max_inflight_requests is a mux-wide memory and backpressure budget, not a per-sub-stream limit. Each sub-stream receives max_inflight_requests / stream_count capacity using integer division, and any remainder is unused. The budget must be at least the stream count.

Each ingest returns a MessageId containing the selected sub-stream index and its local offset. Per-sub-stream ordering is preserved, but records, MessageIds, acknowledgments, and recovered unacknowledged records have no global order. A MessageId must only be used with the mux that created it.

Retryable failures follow each sub-stream's configured recovery policy. If one sub-stream reaches a terminal failure observed by a mux operation, the mux is poisoned and stops accepting records. Healthy lanes remain active until explicit close or drop. flush() waits for every lane's flush attempt; close() flushes once and then closes all lanes concurrently. get_unacked_records() and get_unacked_batches() aggregate recoverable records across streams but do not reconstruct submission order. There is no multiplexed recreation API.

Use multiplexed_ack_callback with AckCallback<MessageId> for mux acknowledgments; ordinary ack_callback is rejected. One callback object is shared across the sub-stream callback workers, which may invoke it concurrently and without global ordering.

Setters can be called in any order. The builder validates at build() time that both authentication and format have been configured.

Avro Stream (Beta, opt-in)

(Requires features = ["avro"]; ephemeral streams only, feature in development.)

Declare the Avro writer schema (JSON) with .avro(schema), then ingest either an AvroRecord the stream encodes against that schema, or a pre-encoded AvroBytes:

use databricks_zerobus_ingest_sdk::{AvroRecord, AvroValue};

let mut stream = sdk
    .stream_builder().table("catalog.schema.orders")
    .oauth(client_id, client_secret)
    .avro(schema_json)
    .build()
    .await?;

let record = AvroValue::Record(vec![
    ("id".to_string(), AvroValue::Long(1)),
    ("customer_name".to_string(), AvroValue::String("Alice".to_string())),
]);
stream.ingest_record_offset(AvroRecord(record)).await?; // queue only — do NOT wait here
stream.flush().await?; // wait once for all pending acknowledgments

AvroValue is apache_avro's value type (re-exported), so it can represent any Avro type — unions, fixed, decimal, and logical types included. See examples/avro/. Avro can also be multiplexed by calling .multiplexed(n) after .avro(schema); all managed sub-streams use the same writer schema. See the avro_multiplexed example.

5. Ingest Data

The SDK provides flexible ways to ingest data with different levels of abstraction:

Wrapper Format Description
ProtoMessage<T> Proto Auto-encoding: pass structs, SDK handles encoding
ProtoBytes Proto Pre-encoded: pass bytes with explicit wrapper
Vec<u8> Proto Backward-compatible: raw bytes without wrapper
ProtoBytes(record.encode()?) Proto Dynamic: build fields at runtime against a descriptor, then encode (dynamic_proto)
JsonValue<T> JSON Auto-serializing: pass structs, SDK handles JSON conversion
JsonString JSON Pre-serialized: pass JSON strings with explicit wrapper
String JSON Backward-compatible: raw strings without wrapper
AvroRecord Avro Object: pass an AvroValue, SDK encodes it against the writer schema (feature: avro, Beta)
AvroBytes Avro Pre-encoded: pass a raw Avro datum with explicit wrapper (feature: avro, Beta)

How acknowledgment works: ingest_record_offset() returns as soon as the record is queued; the SDK sends it and tracks its acknowledgment in the background. To confirm records are durably committed, call flush() — it returns once everything queued so far is acknowledged. The returned OffsetId is a handle you can also wait on individually with wait_for_offset() when a specific record must be confirmed before continuing. Avoid calling wait_for_offset() after every record in a loop, though: that waits out a full round-trip before sending the next record and limits throughput to one record per round-trip.

Recommended: high-throughput ingestion

Ingest in a loop, then flush() to confirm all pending acknowledgments. For long-running streams, flush() periodically to bound memory:

use databricks_zerobus_ingest_sdk::ProtoMessage;

for i in 0..100_000 {
    let record = YourMessage { id: Some(i), /* ... */ };
    // Returns immediately once queued; flush() below confirms the batch.
    let _offset = stream.ingest_record_offset(ProtoMessage(record)).await?;

    // Periodically flush to bound memory on long-running streams.
    if (i + 1) % 10_000 == 0 {
        stream.flush().await?;
    }
}
stream.flush().await?; // Wait once for every pending acknowledgment.

Batch Ingestion

For records you already have grouped, send them as one batch with all-or-nothing semantics. This is the most efficient option in hot paths — it amortizes per-call overhead:

use databricks_zerobus_ingest_sdk::ProtoMessage;

let records: Vec<ProtoMessage<YourMessage>> = vec![
    ProtoMessage(YourMessage { id: Some(1), /* ... */ }),
    ProtoMessage(YourMessage { id: Some(2), /* ... */ }),
    ProtoMessage(YourMessage { id: Some(3), /* ... */ }),
];

// Returns Some(offset) for non-empty batches, None for empty batches.
// Queue many batches this way; flush() once when done.
let _offset = stream.ingest_records_offset(records).await?;
stream.flush().await?;

Per-record confirmation

When a specific record must be confirmed as durably committed before you continue — e.g. low-volume control messages — wait on its returned offset. This is the right tool for that case; for high-volume ingestion, prefer the loop-then-flush() pattern above, since waiting on each record limits throughput to one round-trip per record.

use databricks_zerobus_ingest_sdk::ProtoMessage;

let record = YourMessage { id: Some(1), name: Some("Alice".to_string()) };
let offset = stream.ingest_record_offset(ProtoMessage(record)).await?;
stream.wait_for_offset(offset).await?; // OK for one-off / low-volume records.

See examples/ for complete working examples with all wrapper types, serialization formats, and ingestion patterns.

6. Handle Acknowledgments

The recommended ingest_record_offset() and ingest_records_offset() methods return offsets directly (after queuing):

  • ingest_record_offset() returns OffsetId (the logical offset)
  • ingest_records_offset() returns Option<OffsetId> (None if the batch is empty)
// Preferred: ingest without waiting, then confirm everything at once.
for i in 0..1000 {
    // Returns the offset as soon as the record is queued; does NOT block on the ack.
    let _offset = stream.ingest_record_offset(record).await?;
}
stream.flush().await?; // Waits for all pending acknowledgments in one shot.

// Batches return Option<OffsetId> (None if the batch is empty). Queue them the
// same way — keep ingesting, and flush() once when you're done.
let batch = vec![data1, data2, data3];
let _offset = stream.ingest_records_offset(batch).await?;
stream.flush().await?;

// Low volume only: wait on a single returned offset when you must confirm that
// specific record is committed before continuing. Never do this inside a hot loop.
let offset_id = stream.ingest_record_offset(data).await?;
stream.wait_for_offset(offset_id).await?;
println!("Record committed at offset: {}", offset_id);

Using Acknowledgment Callbacks

For scenarios where you need to track acknowledgments without explicitly waiting (e.g., for metrics or logging), you can use callbacks:

use databricks_zerobus_ingest_sdk::{AckCallback, OffsetId};
use std::sync::Arc;

// Define a callback that implements the AckCallback trait
struct MyCallback;

impl AckCallback for MyCallback {
    fn on_ack(&self, offset_id: OffsetId) {
        // Called when a record is acknowledged
        println!("✓ Acknowledged offset: {}", offset_id);
    }

    fn on_error(&self, offset_id: OffsetId, error_message: &str) {
        // Called when a record encounters an error
        eprintln!("✗ Error for offset {}: {}", offset_id, error_message);
    }
}

// Configure stream with callback
let mut stream = sdk
    .stream_builder().table("catalog.schema.orders")
    .oauth(client_id, client_secret)
    .compiled_proto(descriptor_proto)
    .max_inflight_requests(10_000)
    .ack_callback(Arc::new(MyCallback))
    .build()
    .await?;

for i in 0..1000 {
    let record = YourMessage { id: Some(i), /* ... */ };
    stream.ingest_record_offset(ProtoMessage(record)).await?;
    // Callback fires when this record is acknowledged
}

stream.flush().await?;

Important: Callbacks run synchronously in a dedicated callback handler task. Keep them lightweight (simple logging, metrics increment) to avoid callback backlog. For heavy work like database writes or network calls, send data to a channel for processing in a separate task:

use tokio::sync::mpsc;

struct ChannelCallback {
    tx: mpsc::UnboundedSender<OffsetId>,
}

impl AckCallback for ChannelCallback {
    fn on_ack(&self, offset_id: OffsetId) {
        // Lightweight: just send to channel
        let _ = self.tx.send(offset_id);
    }

    fn on_error(&self, offset_id: OffsetId, error_message: &str) {
        eprintln!("Error: {}", error_message);
    }
}

let (tx, mut rx) = mpsc::unbounded_channel();
let callback = Arc::new(ChannelCallback { tx });

// Heavy processing in separate task
tokio::spawn(async move {
    while let Some(offset) = rx.recv().await {
        // Heavy work here (database writes, API calls, etc.)
        write_to_database(offset).await;
    }
});

7. Close the Stream

Always close streams to ensure data is flushed:

// Close gracefully (flushes automatically)
stream.close().await?;

If the stream fails, retrieve unacknowledged records:

match stream.close().await {
    Err(_) => {
        // Option 1: Get individual records (flattened)
        let unacked = stream.get_unacked_records().await?;
        let total_records = unacked.count();
        println!("Failed to ack {} records", total_records);

        // Option 2: Get records grouped by batch (preserves batch structure)
        let unacked_batches = stream.get_unacked_batches().await?;
        let total_records: usize = unacked_batches.iter().map(|batch| batch.get_record_count()).sum();
        println!("Failed to ack {} records in {} batches", total_records, unacked_batches.len());

        // Retry with a new stream
    }
    Ok(_) => println!("Stream closed successfully"),
}

Multiplexed streams use multiplexed_ack_callback with AckCallback<MessageId>. Different sub-stream workers may call it concurrently, so implementations must be thread-safe and must not rely on global ordering. See the complete proto_compiled_multiplexed example.

Client-side warnings

The SDK logs a WARN-level message via tracing when 100 or more streams for the same table are opened within a 60-second sliding window. This usually indicates a "one stream per record" misuse pattern. The warning fires again if the rate drops below the threshold and later surges again.

To suppress, set the environment variable before starting the process:

ZEROBUS_SDK_WARNINGS_ENABLED=false

Also accepts 0 or no.

Configuration Options

SDK builder options

Method Default Description
endpoint(...) Required Set the Zerobus API endpoint.
unity_catalog_url(...) Unset Set the Unity Catalog endpoint. Required for built-in OAuth or federated authentication (both exchange tokens at this URL).
tls_config(...) System CA certificates Provide custom TLS configuration.
application_name(...) Unset Append an application identifier to the SDK's HTTP user-agent header.
connection_per_stream(bool) true Give each JSON/protobuf stream a dedicated gRPC connection. Pass false to multiplex streams over one shared HTTP/2 connection. Arrow Flight streams are unaffected because they already use dedicated connections.
token_cache_enabled(bool) true Enable or disable token caching for the built-in OAuth and federated paths.
token_refresh_buffer(Duration) 5 minutes Set how long before expiry a cached OAuth or federated token is refreshed.

HTTP/2 multiplexes logical streams over one TCP connection. On high-throughput workloads over the public internet, packet loss and TCP retransmissions can cause head-of-line blocking that stalls every stream on that connection and reduces aggregate throughput. Dedicated connections isolate that loss. If you run many smaller, low-throughput streams, set connection_per_stream(false); shared multiplexing is recommended there to reduce connection overhead.

StreamConfigurationOptions

Field Type Default Description
max_inflight_requests usize 1,000,000 Maximum unacknowledged requests in flight. For a mux, this is divided evenly across sub-streams using integer division.
recovery bool true Enable automatic stream recovery on failure
recovery_timeout_ms u64 15,000 Timeout for recovery operations (ms)
recovery_backoff_ms u64 2,000 Delay between recovery retry attempts (ms)
recovery_retries u32 4 Maximum number of recovery attempts
server_lack_of_ack_timeout_ms u64 60,000 Timeout waiting for server acks (ms)
flush_timeout_ms u64 300,000 Timeout for flush operations (ms)
record_type RecordType RecordType::Proto Record serialization format (Proto or Json)
stream_paused_max_wait_time_ms Option<u64> None Max time to wait for outstanding acknowledgments during graceful close (None = server grace remaining after reserving transport cleanup time, Some(0) = skip the ACK wait, Some(x) = the smaller of x and that remaining grace). A bounded request/response drain still runs after the ACK wait.
ack_callback Option<Arc<dyn AckCallback>> None Ordinary JSON and Protocol Buffer gRPC streams only. Optional callback for acknowledgment notifications. Multiplexed streams use the builder's multiplexed_ack_callback method; Arrow Flight does not support acknowledgment callbacks.
callback_max_wait_time_ms Option<u64> Some(5_000) JSON and Protocol Buffer gRPC streams only. Maximum time to wait for callback processing to complete after closing the stream (None = wait indefinitely, Some(x) = wait up to x ms)

ArrowStreamConfigurationOptions

Field Type Default Description
max_inflight_batches usize 1,000 Maximum accepted batches awaiting acknowledgment; bounds memory and applies backpressure
recovery bool true Reconnect and replay unacknowledged batch suffixes after retryable failures
recovery_timeout_ms u64 15,000 Timeout for one recovery attempt (ms)
recovery_backoff_ms u64 2,000 Delay between recovery attempts (ms)
recovery_retries u32 4 Maximum recovery attempts
server_lack_of_ack_timeout_ms u64 60,000 Maximum pending lifetime for the oldest submitted batch on an active connection
flush_timeout_ms u64 300,000 Timeout for flush(), wait_for_offset(), and the explicit-close acknowledgment wait
connection_timeout_ms u64 30,000 Timeout for Flight connection establishment
ipc_compression Option<CompressionType> None Optional LZ4_FRAME or ZSTD Arrow IPC compression
stream_paused_max_wait_time_ms Option<u64> None ACK-wait cap during server rotation; None uses the available server grace, while Some(0) skips ACK waiting but not bounded transport cleanup 
let stream = sdk
    .stream_builder()
    .table("catalog.schema.orders")
    .oauth(client_id, client_secret)
    .arrow(schema)
    .max_inflight_batches(1_000)
    .connection_timeout_ms(30_000)
    .ipc_compression(Some(arrow_ipc::CompressionType::ZSTD))
    .build_arrow()
    .await?;

For Arrow Flight streams, server_lack_of_ack_timeout_ms is an absolute limit on how long the oldest batch may remain pending during normal stream operation, not an inactivity timeout. No timer runs while the stream is idle or while a batch is buffered but not submitted. For the oldest submitted pending batch, the deadline is based on when it became pending; responses and partial acknowledgments do not pause or extend it. Replayed batches receive a fresh deadline after the full replay completes and acknowledgment processing can resume on the recovered connection. Configure this timeout together with max_inflight_batches so the server can acknowledge a full allowed backlog within the timeout.

Arrow stream construction rejects recovery_timeout_ms and server_lack_of_ack_timeout_ms values whose deadlines cannot be represented by the platform monotonic clock. Server-advertised graceful-rotation periods are capped at one year. If ack_callback or multiplexed_ack_callback was configured on the builder, build_arrow() returns ZerobusError::InvalidArgument instead of silently ignoring the callback.

For internal lifecycle, offset, recovery, and concurrency details, see the Arrow Flight maintainer architecture guide.

JSON / protobuf example:

let stream = sdk
    .stream_builder()
    .table("catalog.schema.orders")
    .oauth(client_id, client_secret)
    .json()
    .max_inflight_requests(50_000)
    .recovery(true)
    .recovery_timeout_ms(20_000)
    .recovery_retries(5)
    .flush_timeout_ms(600_000)
    .build()
    .await?;

Proxy Configuration

JSON / Protocol Buffer and Arrow Flight streams use the same proxy policy for initial connections, automatic recovery, and stream recreation. By default, the SDK uses the first non-empty proxy variable in this order:

  1. grpc_proxy, then GRPC_PROXY
  2. https_proxy, then HTTPS_PROXY
  3. http_proxy, then HTTP_PROXY

Set no_grpc_proxy/NO_GRPC_PROXY, falling back to no_proxy/NO_PROXY, to a comma-separated list of host suffixes that should connect directly. * bypasses the proxy for every host. Both http:// and https:// CONNECT proxies are supported; HTTPS proxy certificates are validated with the system trust store.

For application-specific routing, install a ConnectorFactory with ZerobusSdkBuilder::connector_factory. A caller-supplied factory completely replaces environment discovery and must implement any desired no-proxy rules itself. Returning None connects directly; returning a ProxyConnector created with ProxyConnector::new uses that proxy. The selected factory is retained for every Arrow replacement channel.

Error Handling

The SDK categorizes errors as retryable or non-retryable:

Retryable Errors

Auto-recovered if recovery is enabled:

  • Network failures
  • Connection timeouts
  • Temporary server errors
  • Stream closed by server

Non-Retryable Errors

Require manual intervention:

  • InvalidUCTokenError - Invalid OAuth credentials
  • InvalidTableName - Table doesn't exist or invalid format
  • InvalidArgument - Invalid parameters, schema mismatch, or payload too large (see Payload Size Limit)
  • InvalidSchema (Arrow Flight) - The client's Arrow schema does not match the target Delta table (see Schema Mismatch)
  • Code::Unauthenticated - Authentication failure
  • Code::PermissionDenied - Insufficient table permissions
  • ChannelCreationError - Failed to establish TLS connection

Code::Unauthenticated and Code::PermissionDenied remain globally non-retryable and require manual intervention. During initial stream setup only, if recovery is enabled and at least one configured retry remains, the SDK may invalidate the rejected credentials and spend at most one recovery retry on another setup attempt. A second authentication rejection is terminal. This one-shot setup exception does not change reconnect behavior.

Schema Mismatch (Arrow Flight)

(Requires features = ["arrow-flight"].)

When the server rejects an Arrow Flight stream because the client's schema no longer matches the target Delta table — for example, a column was added to or dropped from the table — the SDK surfaces a structured ZerobusError::InvalidSchema rather than an opaque CreateStreamError. This applies both to initial stream setup (build_arrow()) and to mid-stream reconnects: a schema change detected during recovery is surfaced immediately (via wait_for_offset / flush) instead of being retried until the recovery budget drains.

InvalidSchema carries the raw, machine-readable facts the server reported so you can branch programmatically instead of parsing the message string. The SDK deliberately does not decide whether a mismatch is recoverable — that policy is yours (for example, re-resolving the table schema from Unity Catalog and rebuilding the stream):

use databricks_zerobus_ingest_sdk::{SchemaValidationCause, ZerobusError};

match stream.wait_for_offset(offset).await {
    Ok(()) => { /* durable */ }
    // `InvalidSchema` is #[non_exhaustive], so match with `..`.
    Err(ZerobusError::InvalidSchema { causes, error_code, message, .. }) => {
        // `causes` are typed tokens (e.g. FieldNotInTable, MissingRequiredColumn);
        // `error_code` is the server's numeric code (e.g. "8001") for telemetry;
        // `message` carries per-field detail for diagnostics.
        let drift_only = !causes.is_empty()
            && causes.iter().all(|c| matches!(
                c,
                SchemaValidationCause::FieldNotInTable
                    | SchemaValidationCause::MissingRequiredColumn
            ));
        if drift_only {
            // Re-resolve the table schema and rebuild the stream, then replay.
        } else {
            // e.g. TypeIncompatible — a re-resolve won't help; surface to the operator.
        }
        let _ = (error_code, message);
    }
    Err(e) => { /* handle other errors */ }
}

SchemaValidationCause is #[non_exhaustive] and includes an Unknown(String) variant, so a newer server cause the SDK doesn't recognize is still surfaced rather than dropped.

Variant Extension Annotation (Arrow Flight)

(Requires features = ["arrow-flight"].)

arrow_schema_from_uc_columns / arrow_schema_from_uc_schema build an Arrow schema from Unity Catalog columns. By default VARIANT columns become a plain Struct<metadata, value> with no marker. Pass ArrowSchemaOptions { annotate_variant_extension: true } to also tag every variant field (top-level and nested) with the canonical arrow.parquet.variant Arrow extension, so downstream consumers can tell which fields are variants:

use databricks_zerobus_ingest_sdk::schema::{
    arrow_schema_from_uc_columns_with_options, ArrowSchemaOptions, UcColumn,
};

let mut options = ArrowSchemaOptions::default();
options.annotate_variant_extension = true;
let schema = arrow_schema_from_uc_columns_with_options(&uc_columns, &options)?;
# Ok::<(), databricks_zerobus_ingest_sdk::schema::SchemaError>(())

The physical Struct<metadata, value> shape is unchanged. Leave it off (the default) unless a consumer needs the marker: the Arrow Flight server's target schema is unmarked, so a marked schema forces a per-batch server-side cast. If you enable it, build your RecordBatch from the same marked schema — ingest_batch compares schemas including field metadata and rejects a marked-stream/unmarked-batch mismatch locally.

Payload Size Limit

The Zerobus server applies a 10 MiB limit to the raw record bytes of an ingest call, and additionally enforces a transport-layer limit on the full serialized request (record bytes plus protobuf framing and stream metadata). The SDK enforces a limit client-side so you get an immediate InvalidArgument error rather than a server rejection:

// This will immediately return Err(ZerobusError::InvalidArgument(...))
let oversized = vec![0u8; 11 * 1024 * 1024];
let result = stream.ingest_record_offset(oversized).await;
// Err: Ingest payload too large: 11534336 bytes exceeds the configured limit of 10420224 bytes

The limit applies to the total encoded size of the call — the sum of all record bytes passed to ingest_record_offset or ingest_records_offset. Split large payloads across multiple calls to stay within the limit.

The default limit is set slightly below 10 MiB to leave headroom for the request envelope, so that a payload accepted client-side isn't later rejected by the server's transport layer. It is configurable per stream via the builder (JSON and Protocol Buffer streams only — Arrow Flight streams do not enforce this limit, and setting it before build_arrow() logs a warning):

let stream = sdk
    .stream_builder()
    .table("catalog.schema.table")
    .oauth("client-id", "client-secret")
    .json()
    .max_ingest_payload_bytes(5 * 1024 * 1024) // 5 MiB
    .build()
    .await?;

Check if an error is retryable:

match stream.ingest_record_offset(payload).await {
    Ok(offset) => {
        stream.wait_for_offset(offset).await?;
    }
    Err(e) if e.is_retryable() => {
        eprintln!("Retryable error, SDK will auto-recover: {}", e);
    }
    Err(e) => {
        eprintln!("Fatal error, manual intervention needed: {}", e);
        return Err(e.into());
    }
}

Examples

Complete Working Examples

The examples/ directory contains working examples covering different serialization formats and ingestion patterns:

Example Serialization Ingestion Run with
json/single.rs JSON Single-record cargo run -p rust-examples-json --example json_single
json/batch.rs JSON Batch cargo run -p rust-examples-json --example json_batch
proto/compiled/single.rs Protocol Buffers Single-record cargo run -p rust-examples-proto --example proto_compiled_single
proto/compiled/batch.rs Protocol Buffers Batch cargo run -p rust-examples-proto --example proto_compiled_batch
proto/dynamic/single.rs Protocol Buffers (runtime schema) Single-record cargo run -p rust-examples-proto --example proto_dynamic_single
proto/dynamic/batch.rs Protocol Buffers (runtime schema) Batch cargo run -p rust-examples-proto --example proto_dynamic_batch
avro/single.rs Avro (Beta) Single-record cd examples/avro && cargo run --example avro_single
avro/batch.rs Avro (Beta) Batch cd examples/avro && cargo run --example avro_batch
avro/multiplexed.rs Avro (Beta) Multiplexed cd examples/avro && cargo run --example avro_multiplexed

The Avro examples live in an excluded crate (examples/avro), so they run with cd examples/avro rather than -p.

Check examples/README.md for setup instructions and detailed comparisons.

Stream Recovery

let sdk = ZerobusSdk::builder()
    .endpoint(endpoint)
    .unity_catalog_url(uc_endpoint)
    .build()?;

let mut stream = sdk
    .stream_builder().table("catalog.schema.table")
    .oauth(client_id, client_secret)
    .json()
    .build()
    .await?;

// Ingest data...
match stream.close().await {
    Err(_) => {
        // Stream failed, recreate with unacked records
        stream = sdk.recreate_stream(&stream).await?;
    }
    Ok(_) => println!("Closed successfully"),
}

Tests

Integration tests live in the tests/ crate and run against a lightweight mock Zerobus gRPC server.

  • Mock server: tests/src/mock_grpc.rs
  • Test suite: tests/src/rust_tests.rs

Run tests with logs:

cargo test -p tests -- --nocapture

Best Practices

  1. Reuse SDK Instances - Create one ZerobusSdk per application and reuse for multiple streams
  2. Always Close Streams - Use stream.close().await? to ensure all data is flushed
  3. Confirm with flush(), not per-record waits - The idiomatic flow is to ingest in a loop and flush() once (or periodically). Use wait_for_offset() only when a specific record must be confirmed individually — calling it after every record serializes the pipeline into one round-trip per record.
  4. Choose the Right Ingestion Method:
    • Use ingest_records_offset() for high throughput batch ingestion
    • Use ingest_record_offset() when processing records individually
    • Both return offsets directly; confirm with flush() at the end, or wait_for_offset() only when you need per-record confirmation
  5. Tune Inflight Limits - Adjust max_inflight_requests based on memory and throughput needs
  6. Enable Recovery - Always set recovery: true in production environments
  7. Use Ack Callbacks for async tracking - Register an ack_callback for metrics/logging instead of blocking on wait_for_offset()
  8. Monitor Errors - Log and alert on non-retryable errors
  9. Validate Schemas - Use the schema generation tool to ensure type safety (for Protocol Buffers)
  10. Secure Credentials - Never hardcode secrets; use environment variables or secret managers
  11. Test Recovery - Simulate failures to verify your error handling logic
  12. Use Multiplexing for Stream Bottlenecks - When you need more throughput than one gRPC stream can provide and do not require global ordering, select .multiplexed(n)

API Reference

ZerobusSdk

Main entry point for the SDK.

Builder:

let sdk = ZerobusSdk::builder()
    .endpoint("https://workspace.zerobus.databricks.com")  // Required
    .unity_catalog_url("https://workspace.cloud.databricks.com")  // Optional with custom headers
    .tls_config(Arc::new(SecureTlsConfig::new()))  // Optional, defaults to SecureTlsConfig
    .build()?;

Methods:

pub fn stream_builder(&self) -> StreamBuilder<'_>

Returns a fluent builder for creating ingestion streams. All setters can be called in any order. Call validate() to check the configuration without opening a stream, or build() / build_arrow() to open the stream. See Create a Stream for usage.

pub async fn recreate_stream(
    &self,
    stream: &ZerobusStream
) -> ZerobusResult<ZerobusStream>

Recreates a failed stream, preserving and re-ingesting unacknowledged records.

ZerobusStream

Represents an active ingestion stream.

Methods:

pub async fn ingest_record_offset(
    &self,
    payload: impl Into<PreparedInput>
) -> ZerobusResult<OffsetId>

Ingests a single record — any record wrapper or raw payload matching the stream's format (see Ingest Data). The await queues the record for sending and returns the logical offset ID directly. Use wait_for_offset() to explicitly wait for server acknowledgment of this offset.

pub async fn ingest_records_offset(
    &self,
    payloads: Vec<impl Into<PreparedInput>>
) -> ZerobusResult<Option<OffsetId>>

Ingests multiple records as a batch with all-or-nothing semantics. The entire batch either succeeds or fails as a unit. The await queues the batch for sending and returns the logical offset ID directly (or None for empty batches). Use wait_for_offset() to explicitly wait for server acknowledgment.

pub async fn wait_for_offset(&self, offset_id: OffsetId) -> ZerobusResult<()>

Waits for acknowledgment of a specific logical offset. Use this method with offsets returned from ingest_record_offset() or ingest_records_offset() to explicitly wait for server acknowledgment.

pub async fn flush(&self) -> ZerobusResult<()>

Flushes all pending records and waits for acknowledgment.

pub async fn close(&mut self) -> ZerobusResult<()>

Flushes and closes the stream gracefully.

pub async fn get_unacked_records(&self) -> ZerobusResult<impl Iterator<Item = EncodedRecord>>

Returns an iterator over all unacknowledged records as individual EncodedRecord items. This flattens batches into individual records. Only call after stream failure.

pub async fn get_unacked_batches(&self) -> ZerobusResult<Vec<EncodedBatch>>

Returns unacknowledged records grouped by batch, preserving the original batch structure. Records ingested together remain grouped:

  • Each ingest_record_offset() call creates a batch containing one record
  • Each ingest_records_offset() call creates a batch containing multiple records

Only call after stream failure.

StreamBuilder

Configure stream parameters via fluent setters; all configuration goes through the builder. See Create a Stream for usage and Configuration Options for the full list of available setters and their defaults.

Calling multiplexed(stream_count) consumes the ordinary builder and returns a MultiplexedStreamBuilder, which exposes only validate() and build(). Configure ordinary streams with ack_callback; configure mux streams with multiplexed_ack_callback. Each terminal build rejects the other callback type.

MultiplexedStream

A managed group of 1–64 homogeneous gRPC sub-streams. Its ingest_record() and ingest_records() methods return MessageIds, while flush(), close(), and unacknowledged-record retrieval operate across all sub-streams.

AckCallback

Trait for receiving acknowledgment notifications.

Methods:

pub trait AckCallback<Id = OffsetId>: Send + Sync {
    fn on_ack(&self, id: Id);
    fn on_error(&self, id: Id, error_message: &str);
}
  • on_ack() - Called when a record/batch is successfully acknowledged
  • on_error() - Called when a record/batch encounters an error

HeadersProvider

Trait for custom authentication header providers.

Methods:

#[async_trait]
pub trait HeadersProvider: Send + Sync {
    async fn get_headers(&self) -> ZerobusResult<HashMap<&'static str, String>>;
}

Implement this trait to provide custom authentication headers. The default implementation (OAuthHeadersProvider) handles OAuth 2.0 token management. Use this for:

  • Custom token caching strategies
  • Alternative authentication mechanisms
  • Integration with centralized credential services

See Custom Authentication section for usage examples.

TlsConfig

Trait for TLS configuration strategies.

Implementations:

  • SecureTlsConfig (default) - Production TLS with system CA certificates
  • NoTlsConfig - No-op TLS for testing with http:// endpoints (requires testing feature)
pub trait TlsConfig: Send + Sync {
    fn configure_endpoint(&self, endpoint: Endpoint) -> ZerobusResult<Endpoint>;
}

ZerobusError

Error type for all SDK operations.

Methods:

pub fn is_retryable(&self) -> bool

Returns true if the error can be automatically recovered by the SDK.

Language Bindings

This repository includes bindings for building SDKs in other languages.

C FFI (ffi/)

C Foreign Function Interface for languages that can call C functions (Go, C#, C++, etc.).

# Build static and dynamic libraries
cargo build -p zerobus-ffi --release

# Output:
# target/release/libzerobus_ffi.a      (static library for Go, C++)
# target/release/libzerobus_ffi.so     (Linux dynamic library for C#)
# target/release/libzerobus_ffi.dylib  (macOS dynamic library for C#)
# target/release/zerobus_ffi.dll       (Windows dynamic library for C#)
# ffi/zerobus.h                        (C header file)

Pre-built binaries for all platforms are available in GitHub Releases with tags ffi/vX.X.X.

JNI (jni/)

Java Native Interface bindings for the Zerobus Java SDK.

# Build JNI library
cargo build -p zerobus-jni --release

# Output:
# target/release/libzerobus_jni.so     (Linux)
# target/release/libzerobus_jni.dylib  (macOS)
# target/release/zerobus_jni.dll       (Windows)

Pre-built binaries are available in GitHub Releases with tags jni/vX.X.X.

Building from Source

For contributors or those who want to build and test the SDK:

git clone https://github.com/databricks/zerobus-sdk.git
cd zerobus-sdk/rust
cargo build --workspace

Build specific components:

# Build only SDK
cargo build -p databricks-zerobus-ingest-sdk

# Build only schema tool
cargo build -p tools

# Build and run JSON single-record example
cargo run -p rust-examples-json --example json_single

# Build and run JSON batch example
cargo run -p rust-examples-json --example json_batch

# Build and run Protocol Buffers single-record example (compiled schema)
cargo run -p rust-examples-proto --example proto_compiled_single

# Build and run Protocol Buffers batch example (compiled schema)
cargo run -p rust-examples-proto --example proto_compiled_batch

# Build and run Protocol Buffers dynamic-schema examples
cargo run -p rust-examples-proto --example proto_dynamic_single
cargo run -p rust-examples-proto --example proto_dynamic_batch

# Build and run Avro examples (Beta; excluded crate, enables the avro feature)
(cd examples/avro && cargo run --example avro_single)
(cd examples/avro && cargo run --example avro_batch)

Community and Contributing

This is an open source project. We welcome contributions, feedback, and bug reports.

License

This SDK is licensed under the Apache License 2.0. See LICENSE for the full text.

Requirements

  • Rust 1.70 or higher (2021 edition)
  • TLS - Uses native OS certificate store by default (configurable via TlsConfig trait)
  • See prerequisites for Databricks workspace and credential requirements

For issues, questions, or contributions, please visit the GitHub repository. See the monorepo README for an overview of all available SDKs.