Exactly-once delivery

ArkFlow delivers at-least-once by default. On recovery, in-flight messages are replayed and MAY be redelivered to outputs. For duplicate-intolerant sinks, ArkFlow can deliver exactly-once (more precisely, effectively-once) opt-in, via a transactional Kafka output. This page describes what that guarantees, how to configure it, and where its boundary lies.

How it works

A Kafka output configured for exactly-once uses a transactional producer:

• init_transactions is called once at connect();
• before each acknowledged batch, the output begins a transaction, sends every message in the batch, then commits;
• the WAL cursor advances (and the source is committed) only after the transaction commits successfully.

The unit of work is one ack range = one write_batch call = one Kafka transaction. If a buffer (memory, a tumbling/sliding/session window, or a join) aggregates several input messages into one output batch, that whole batch is one atomic transaction unit. Downstream consumers reading with isolation.level=read_committed observe each batch atomically — all messages or none.

On any failure (a commit_transaction error that requires abort, or a crash), the batch is not acknowledged, the WAL cursor does not advance, and the range is replayed on recovery — which begins a fresh transaction.

Configuration

Enable exactly-once on the Kafka output with two keys:

yaml

output:
  type: kafka
  brokers:
    - localhost:9092
  topic:
    type: value
    value: orders-copy
  exactly_once: true
  transactional_id: arkflow-orders-copy-0   # stable across restarts; unique per producer

• exactly_once: true turns on transactional production.
• transactional_id is required when exactly_once is enabled. It MUST be stable across restarts (you own this) and unique per stream producer. On restart the broker uses the same id to fence the prior producer epoch and abort its in-flight (zombie) transaction, so zombie writes are never visible to read_committed consumers.

Exactly-once is layered on top of at-least-once ingestion, so the input side needs durability enabled so that a crash between read and output does not lose data:

yaml

durability:
  enabled: true
  path: "./data/wal-eos"
  sync: group_commit

A complete runnable example is in examples/eos-kafka.yaml (Kafka → Kafka, consume-transform-produce).

The honest boundary (read this)

The Kafka transactional output eliminates two specific sources of duplicates:

• in-transaction partial writes — the transaction commits atomically, so a read_committed consumer never sees a partial batch;
• zombie-producer duplicates — the stable transactional_id fences stale producer epochs across restarts.

It does not guarantee the absence of duplicates when a crash occurs after the producer transaction is committed and before the source offset is committed. The source offset is committed asynchronously (bounded by the source's auto-commit interval, e.g. Kafka's default 5s). If the process crashes in that window, on recovery the source redelivers the range, a new producer writes it again, and a read_committed downstream consumer observes duplicate rows.

Such residual duplicates MUST be absorbed downstream — by a dedup key, business-level idempotency, or an idempotent sink (e.g. UPSERT). Design your downstream consumers accordingly.

That residual window — a crash between producer commit and source offset commit — is what L3 closes. Configure the paired Kafka input and output to commit the source offset inside the producer transaction via send_offsets_to_transaction:

yaml

input:
  type: kafka
  brokers:
    - localhost:9092
  topics:
    - orders
  consumer_group: orders-copy-group
  start_from_latest: false
  transactional_offsets: true

yaml

output:
  type: kafka
  brokers:
    - localhost:9092
  topic:
    type: value
    value: orders-copy
  exactly_once: true
  transactional_id: arkflow-orders-copy-0
  offset_commit_group: orders-copy-group   # must name the input's consumer_group

• Input transactional_offsets: true registers the consumer group's metadata in a process-internal registry and changes ack() to advance only the in-memory frontier — the input no longer commits offsets itself (its undo() compensation rewinds only that frontier); broker group offsets advance only with output transactions.
• Output offset_commit_group (requires exactly_once: true) derives the covered offsets from each batch's __meta_partition/__meta_offset columns and folds them into the same transaction as the writes. Each partition's commit is clamped to the paired input's contiguous acknowledged frontier — min(batch max offset + 1, frontier next) per partition, never regressing below what this producer already sent — so offsets still settling elsewhere in the graph (content routing, fan-out, filtered branches) can never be transactionally skipped: any crash replays at least the unsettled part. The clamp makes the commit lag one batch behind the writes (a transaction may only commit what previous acknowledgements settled), so the bound is at-least-once with at most one batch of tail duplication: the last written batch before a shutdown may replay. A read_committed consumer of the source group still observes writes and offset advances atomically, and a rolled-back transaction leaves the group offsets unchanged.

L3 boundaries, enforced by explicit errors rather than silent degradation:

• Pairing is validated at startup (fail-closed) — a Kafka input declaring transactional_offsets: true whose consumer group is never named by any output's offset_commit_group would silently never commit its group offsets (a crash then restarts the group purely per auto.offset.reset, and latest skips everything produced meanwhile). Once all components of a stream/job are built, startup validation rejects such a configuration with a configuration error naming the group and both keys.
• Process-internal pairing — the output resolves the named group through the in-process registry; the input and output must run in the same ArkFlow process. Distributed jobs that split input and output across nodes are not covered (independent change). An offset_commit_group naming a group with no live transactional_offsets input fails the write.
• Single input topic — batch metadata carries partitions without topics, so the registry routes by the input's topic list; a multi-topic subscription is rejected.
• Kafka → Kafka only — a batch without Kafka source metadata columns contributes no offsets (the writes still proceed); non-Kafka sources cannot pair with the transactional output. Such batches sit outside the L3 guarantee: with transactional_offsets: true the input no longer commits offsets itself, so if no later metadata-carrying transaction advances the group offset, a restart replays those records and duplicates the output — keep the pipeline free of metadata-dropping processors (batch-rebuilding transformations such as json_to_arrow) between the Kafka input and the output, and absorb duplicates downstream otherwise.

A complete runnable L3 example is in examples/eos-kafka-l3.yaml (examples/eos-kafka.yaml shows the L2-only variant).

Requirements summary

• exactly_once: true requires a non-empty transactional_id; validation fails with a clear error otherwise.
• offset_commit_group requires exactly_once: true; the builder rejects the configuration otherwise. The named group must have a same-process Kafka input with transactional_offsets: true: a configuration whose transactional input is never claimed by any output fails startup with a configuration error, and a write against a group with no live input fails the write.
• The WAL's object-store node_id and the Kafka transactional_id are independent configuration values — neither is derived from the other.
• Outputs other than the transactional Kafka output keep today's default at-least-once behavior. Idempotent adapters for other sinks (SQL UPSERT, etc.) can be added in later changes.