Module coordinator 

Per-sink Iceberg V3 commit coordinator. This is a plain struct (no background task / mpsc): the crate::manager::iceberg_v3_sink::IcebergV3SinkManager holds one per registered sink behind a per-sink async mutex and calls IcebergV3Coordinator::pre_commit / IcebergV3Coordinator::commit directly from the barrier-completion path. The barrier path already serializes pre-commit/commit per epoch, so the coordinator never needs its own request queue.

IcebergV3Coordinator::init is synchronous with respect to registration: it loads the iceberg catalog/table, reads pending_sink_state, and drains any recovered pending epoch via an iceberg overwrite_files transaction BEFORE returning a ready coordinator. Only once init completes does the sink start accepting live pre-commit/commit calls.

The two phases dispatched from complete_barrier:

1. pre_commit — aggregate the reports, generate a snapshot_id, persist the merged file list under pending_sink_state, return. No iceberg I/O.
2. commit — run an iceberg overwrite_files transaction (keyed on the pre-generated snapshot_id for idempotency), then mark the row Committed and prune the prior epoch’s row.

On retry-exhausted commit failure the error propagates to the caller; the barrier then fails and the meta-recovery path drops the coordinator, re-registers it (re-running init), and retries from the persisted pending_sink_state rows.

Structs

EpochCommit 🔒

One epoch’s worth of pre-committed state queued inside the coordinator. Holds the decoded merged file list and the pre-generated snapshot_id. The blob form (PbIcebergV3PreCommitState) is only materialized when persisting to pending_sink_state; in-memory we keep the structured form.

IcebergV3AggResult 🔒

IcebergV3Coordinator

Per-sink Iceberg V3 commit coordinator. Owns the loaded iceberg catalog/table (reused across commits) and the meta SQL connection (used for pending_sink_state exactly-once persistence).

Constants

INIT_TIMEOUT 🔒

Bound the init phase so a register call can’t hang forever if the iceberg endpoint is unreachable; on timeout init returns an error and registration fails (the caller surfaces it / retries).

Functions

aggregate_reports 🔒

align_report_id 🔒

commit_one_epoch 🔒

decode_pre_commit_state 🔒

encode_pre_commit_state 🔒

load_catalog_and_table 🔒

recovery 🔒

Read every persisted row for this sink, recovering prev_committed_epoch and pending commits.