Crate twg_sink_zerobus 

Databricks Zerobus sink: fire-and-forget Arrow ingest with a per-stream background ack lane (ADR-0060), a bandwidth-sized stream pool, catalog-driven table provisioning, and emit-time type reconciliation.

Shape

ZerobusSink implements the [twg_stream_arrow::BatchSink] and [twg_connector_core::Sink] substrate traits. It writes Arrow RecordBatch over the Databricks Arrow-Flight path: no protobuf descriptor is on the write path. The one external boundary — the Zerobus stream — is inverted behind transport::ZerobusTransport; the real Databricks adapter is databricks (behind the databricks-sdk feature), and the whole correctness core is tested against a fake.

Because thalweg is pinned to Arrow 58 (ADR-0003) while the SDK is on Arrow 59, batches cross the boundary as Arrow IPC bytes (ipc::encode_ipc), so no Arrow-59 type reaches the sink. See ADR-0065.

The ack pipeline (ADR-0060)

push/send submit fire-and-forget and hand a typed confirm::Confirm to the stream’s ack lane; the lane coalesces a burst, waits once on its maximum offset, and advances coverage only on the durable ack. The source advances on delivered, never submitted.

Primary-raw eligibility, conditionally

This sink writes Delta, and we carry only an Iceberg reader (ADR-0023). That does not disqualify it from the primary-raw role, because a Delta table can be configured to generate Iceberg metadata asynchronously alongside its own without rewriting data — both formats being Parquet files plus a metadata layer — and the catalog then serves it to Iceberg clients through a read-only Iceberg REST endpoint with vended credentials.

So recovery reads it back through twg-format-iceberg via twg-table-catalog. But this rests on table configuration, not on the sink, so it is a config-validation precondition, not an assumption:

• the table is registered in the catalog (managed or external);
• Iceberg reads / column mapping are enabled on it;
• reader and writer protocol versions meet the feature’s minimum;
• deletion vectors are NOT enabled — they are incompatible with Iceberg reads, and switching them on later silently removes recovery’s read path.

Refuse the primary-raw role if any of these is unmet, and re-check rather than caching the answer: a table property can change underneath a running pipeline.

The read-after-write hazard

Iceberg metadata generation is asynchronous, running on the compute that wrote the Delta commit. A recovery pass reading immediately after a write may therefore see a table state older than the data just committed — which for raw-anchored recovery means silently replaying an incomplete window.

Recovery must therefore either tolerate the lag (bounded windows already trailing the write head make this natural) or trigger metadata generation synchronously before reading. Do not assume read-after-write consistency across the format boundary; it is not offered.

One knock-on

Tables with Iceberg reads enabled use Zstandard rather than Snappy for the underlying Parquet. The read path must therefore support zstd decompression, which twg-wire-compression provides — and note that decode-only is sufficient here, so the pure-Rust decoder covers it without the encode-side purity exception.

Re-exports

pub use config::Role;

pub use config::ZerobusConfig;

pub use confirm::Confirm;

pub use error::ErrorClass;

pub use error::ZerobusSinkError;

pub use metrics::Metrics;

pub use reconcile::Reconciliation;

pub use reconcile::reconcile;

pub use sink::ZerobusSink;

pub use sink::ZerobusSinkDeps;

pub use transport::Offset;

pub use transport::TransportFactory;

pub use transport::ZerobusTransport;

Modules

config

Sink configuration (blueprint §7). Every knob has a conservative default and is logged at startup; the one thing that is deliberately not configurable is whether to wait per-batch — fire-and-forget is structural, not a knob.

confirm

The typed confirmation token (ADR-0060, blueprint §4). Modelling the post-durability action as a typed enum, not a boxed closure, keeps lane separation legible: a stream carries RawAnchor confirmations or DerivedCoverage ones, and applying a confirmation only ever advances coverage after the WAL ack — never on submit.

databricks

The real Databricks Arrow-Flight adapter (behind the databricks-sdk feature). It implements ZerobusTransport/TransportFactory over the databricks-zerobus-ingest-sdk ZerobusArrowStream.

error

Zerobus error taxonomy and its retry / poison / fatal-config classification (blueprint §6). The classification decides the action: transient → honour backoff and retry on the existing stream (never open a new one — that is what caused the predecessor’s stream storm); poison → tear the stream down and rebuild within a bounded budget; fatal-config → fail loud, never degrade.

ipc

Encode an Arrow RecordBatch to Arrow IPC stream bytes.

metrics

Sink metrics (blueprint §9). A plain struct of atomics, shared via Arc, with no coupling to twg-observability (still a scaffold) — a later change wires these into the OTLP/Prometheus surface. All counters are {table,stream}- scoped by the owning stream; there is no shared axis between lanes.

precondition

Primary-raw eligibility (blueprint §8). A Zerobus sink may hold the primary-raw role only if its Delta target is readable back through Iceberg for recovery. These are config-validation preconditions, re-checked not cached (a table property can change under a running pipeline), enforced against the live table properties from the catalog’s describe.

reconcile

Emit-time type reconciliation — the write-boundary guard (blueprint §5.2).

sink

ZerobusSink: the assembly that wires catalog provisioning, the stream pool, and the ADR-0060 ack lane into the [BatchSink] and [Sink] substrate traits.

transport

The Zerobus transport seam: the one boundary the sink inverts so its correctness core (ack lane, pool, coverage) is testable without a live Databricks workspace.