ADR-0065: Zerobus sink — substrate seams, the Arrow-IPC version bridge, and a feature-gated SDK adapter

Reference

Implements the Zerobus sink designed in ../blueprints/zerobus-sink-implementation-plan.md, realising the pipelined fire-and-forget ack of ADR-0060, the source-ack invariant of ADR-0038, and the gap-aware coverage of ADR-0053. Builds on the catalog seam of ADR-0062 (twg-table-catalog, already implemented) and the type-mapping authority of ADR-0019 (twg-type-map::zerobus_sink). Honours the single-Arrow-major pin of ADR-0003.

Context

Building twg-sink-zerobus surfaced three decisions the blueprint did not settle, because they only arise at implementation time.

1. The substrate the sink implements did not exist. The blueprint has the sink implement BatchSink (twg-stream-arrow) and Sink (twg-connector-core) and call a coverage store (twg-offset-store). All three were 0-LOC scaffolds. The sink cannot be built without them.
2. The Databricks SDK is on a different Arrow major. databricks-zerobus-ingest-sdk uses Arrow 59; thalweg is pinned to Arrow 58 (ADR-0003, DataFusion’s major). The SDK’s ingest_batch(RecordBatch) therefore takes a different RecordBatch type than the sink holds — they cannot be passed directly.
3. The SDK is a heavy, network-bound dependency. Compiling it (with its Arrow-Flight/tonic stack and a second Arrow major) on every build, and requiring a live workspace to test the sink’s logic, is unacceptable. The correctness core — the ADR-0060 ack lane, coalescing, coverage — must be testable without it.

Decision

Build the minimal real substrate the sink needs, bridge the Arrow-major gap with Arrow IPC bytes, and invert the SDK behind a ZerobusTransport seam with a feature-gated real adapter and an in-memory fake.

Concrete:

1. Substrate, built minimally and for real.

   • twg-connector-core::sink — Sink, Receipt, OffsetSpan (source-position range, half-open [start, end)), SinkId, SinkError.
   • twg-stream-arrow::batch_sink — BatchSink, CommitInfo.
   • twg-offset-store::coverage — CoverageStore, CoveredRange, gap-aware union-merge and gap computation, and an InMemoryCoverage reference backend. Union-merge is a correctness requirement (ADR-0053: two rebalance writers both record what they wrote), not an optimisation.

   These are the seams the sink implements/calls, no more — the broader flow-control, recovery-role, and durable-backing machinery documented in those crates remains design intent for its phase. twg-sink-zerobus implements both traits.

2. ZerobusTransport — the inverted boundary. A single-stream async trait (submit(ipc_bytes) -> offset, wait_for_offset(offset), flush, close, unacked). The whole correctness core (ack lane, pool, coverage) is written and tested against a fake with scriptable ack latency / transient failures / poison. This is what makes the ADR-0060 properties — one wait_for_offset per coalesced burst, delivered monotonicity, confirm-only-after-ack, poison→exit, backoff-retry-on-existing — provable in CI without Databricks.

3. Arrow-IPC version bridge. The sink stays on Arrow 58 throughout. At the transport boundary a RecordBatch is serialised to Arrow IPC stream bytes (twg-sink-zerobus::ipc) and handed to the SDK’s ingest_ipc_batch(Bytes). IPC is forward/backward-compatible across 58↔59, so no Arrow-59 array/batch type ever enters the sink. Only the stream schema is rebuilt into the SDK’s Arrow-59 Schema at open time (a finite type map). This keeps ADR-0003’s single-major pin true for thalweg’s own code; Arrow 59 exists only inside the SDK, behind an optional feature.

4. Feature-gated real adapter. The Databricks ZerobusArrowStream adapter (databricks.rs) lives behind the databricks-sdk feature (off by default), so the SDK and its Arrow 59 are pulled in only when explicitly enabled. Auth uses a catalog-vended bearer token via a HeadersProvider; replay-on-failure is delegated to the SDK’s own recovery (.recovery(true)), so pool-level rebuild is a no-op for the real transport. CI’s --all-features compiles the adapter, so the binding is kept honest against the real API.

Options considered + consequences

Option 1 (chosen) — minimal real substrate + IPC bridge + inverted transport

• Pros: the sink is genuinely built and its hard parts are tested without a live service; Arrow stays single-major for our code; the SDK is opt-in; the substrate crates gain their real, reusable seams.
• Cons: the substrate crates are only as complete as the sink needs (flow control / recovery roles still pending); the real adapter is compiled-but-not executed in CI; IPC encode adds a serialise step per batch (cheap, and it is the version-safe boundary regardless).

Option 2 — bump thalweg to Arrow 59 and use ingest_batch directly

• Cons: ADR-0003 pins Arrow to DataFusion’s major; moving ahead of DataFusion is a breaking change across every published codec crate for one sink’s convenience. Rejected — the IPC bridge is local and reversible; a global version bump is neither.

Option 3 — put the real adapter in a separate out-of-workspace crate

• Cons: more moving parts, and it would not compile under the workspace’s --all-features gate, so the binding could rot undetected. Rejected in favour of a feature that CI exercises.

Option 4 — implement the full Phase 5a/5b substrate now

• Cons: flow control, DLQ, recovery roles, and durable coverage backings are large subsystems in their own right; building them to finish one sink over-reaches and would ship half-designed. Rejected — build the seams the sink needs; leave the rest to its phase.

Invariants pinned by this ADR

• Single Arrow major in our code — the sink and all its logic are Arrow 58; Arrow 59 is confined to the SDK behind a feature, reached only via IPC bytes.
• The correctness core is service-free-testable — the transport is inverted; the ADR-0060 properties are proven against a fake in CI.
• delivered, never submitted — coverage advances only from the ack lane, only after the durable WAL ack (ADR-0038/0053), preserved verbatim.
• Union-merge coverage — never last-write-wins (ADR-0053).
• The SDK is opt-in — default builds and tests do not depend on it.

Deferred work

Each entry carries a trigger.

• D1 — Emit-time auto-project transform. Dropped, not deferred. An earlier plan had the sink optionally auto-to_json a nested column landing in a STRING target. It is not built and not planned: silently JSON-encoding structure at the sink hides a schema mismatch the operator should see. The two explicit resolutions are strictly better and need no sink magic — declare the column with its real nested type (the catalog creates it), or write to_json(col) in the transform (the UDF already ships in twg-udf). The sink therefore keeps only the loud-named refusal, which points the operator at exactly those two fixes.
• D2 — Observability wiring. Metrics are collected in a local atomics struct; twg-observability is still a scaffold. Trigger: the observability surface lands.
• D3 — Full flow-control / recovery-role substrate. The connector-core / offset-store crates carry only the seams the sink needs. Trigger: Phase 5a/5b.
• D4 — Executed integration against a live workspace. The real adapter is compiled, not run, in CI. Trigger: a test Databricks workspace is available to the pipeline.

Interaction with existing ADRs

• ADR-0003 (single Arrow major): honoured via the IPC bridge; Arrow 59 never enters thalweg’s own code.
• ADR-0060 (pipelined ack): realised — the ack lane is the sink’s core, tested.
• ADR-0038/0053 (source-ack / coverage): the delivered watermark and union-merge coverage implement them.
• ADR-0062 (catalog): the sink calls the real TableCatalog; provisioning is the catalog’s, never the sink’s.
• ADR-0019 (type-map): ProtoFieldType describes the target columns; no protobuf descriptor is on the Arrow write path.

Document version history