Exactly-once stream processing is the promise that a read-process-write computation produces the same durable result it would have produced if every input record affected the computation one time, even when workers crash, network requests time out, producers retry, or partitions rebalance. The phrase is easy to oversell. In a distributed system, messages can be retried, consumers can be restarted from older offsets, and external databases can observe side effects outside the streaming engine. Real exactly-once behavior is therefore a cooperation protocol across input offsets, operator state, output writes, and recovery.
DDIA frames the problem as the path from at-least-once delivery to correct derived data. At-least-once systems avoid data loss by retrying uncertain work, but retries create duplicate side effects. At-most-once systems avoid duplicates by dropping uncertain work, but they lose data. Exactly-once stream processors try to keep the safety of retries while making duplicate attempts invisible to the final result.
Kafka’s 0.11 release added idempotent producers and transactions so a stream task can atomically publish output records and commit the input offsets that led to those records. Flink’s checkpointing model approaches the same end-to-end problem from the engine side: it takes consistent snapshots of source offsets and operator state, then coordinates transactional sinks with a two-phase commit protocol. Both designs are useful because they expose the same invariant: offsets, state, and output must move forward together or not at all.
The practical lesson is conservative. Exactly-once inside Kafka Streams, Flink, or another managed runtime does not automatically make arbitrary HTTP calls, emails, payment captures, database writes, or machine-learning feature updates happen exactly once. When work escapes the transactional boundary, the sink must provide idempotence, transactions, deduplication keys, or reconciliation. The label “exactly once” is not a property of a broker alone; it is a property of a whole pipeline and its side effects.
Message delivery semantics describe what a broker and client do when acknowledgement is uncertain. At-least-once delivery retries after uncertainty, so consumers may see duplicates. At-most-once delivery does not retry, so consumers may miss records. Exactly-once processing is stricter and narrower: after failures, a deterministic stream computation should materialize the same output and state as a failure-free run.
A read-process-write task consumes input offsets, updates local or remote state, and writes output. If it commits input offsets before output is durable, a crash can skip records. If it writes output before offsets are committed, a crash can write the output again. Exactly-once runtimes solve this by placing offsets and output in one transaction or one checkpoint outcome.
Kafka’s idempotent producer assigns producer identities, epochs, and per-partition sequence numbers so broker retries do not append duplicate records for the same producer session. Transactions add a coordinator and a transactional identifier so a replacement producer can fence off a previous “zombie” producer that might still be alive after a partition or long pause.
A stream engine may execute the same user code more than once after a crash. The guarantee is that the committed state and visible output correspond to one logical execution for each input record within the transactional boundary.
Flink can checkpoint state and Kafka can transact topic writes, but a JDBC database, email service, cache, object store, or payment API needs its own commit, idempotency key, or deduplication table. Otherwise the pipeline is only exactly-once up to the point where the unmanaged side effect begins.
A Kafka Streams-style task reads records, updates state, sends output records with a transactional producer, sends consumed offsets to the transaction, and commits the transaction. Consumers using read-committed isolation do not observe aborted output. If the task crashes before commit, both output and offsets are aborted; after restart, the input is read again and the same logical transaction can be rebuilt.
func processBatch(ctx context.Context, consumer Consumer, producer TxProducer, store StateStore) error {
records := consumer.Poll(ctx)
if len(records) == 0 {
return nil
}
if err := producer.BeginTransaction(ctx); err != nil {
return err
}
for _, r := range records {
state := store.Get(r.Key)
nextState, out := applyBusinessLogic(state, r)
if err := store.Put(r.Key, nextState); err != nil {
producer.AbortTransaction(ctx)
return err
}
if err := producer.Send(ctx, outputTopic, out); err != nil {
producer.AbortTransaction(ctx)
return err
}
}
offsets := highestConsumedOffsets(records)
if err := producer.SendOffsetsToTransaction(ctx, offsets, consumer.GroupID()); err != nil {
producer.AbortTransaction(ctx)
return err
}
return producer.CommitTransaction(ctx)
}Flink’s model is checkpoint-centered. The engine injects checkpoint barriers into streams, snapshots source positions and operator state, and lets sinks pre-commit external transactions for the records before the barrier. Once all operators complete the checkpoint, the coordinator notifies sinks to commit. If a checkpoint fails, the runtime aborts or rolls back to the previous completed checkpoint.
type ExactlyOnceSink interface {
BeginTransaction(checkpointID int64) (Txn, error)
Write(txn Txn, event Event) error
PreCommit(txn Txn) error
Commit(txn Txn) error
Abort(txn Txn) error
}
func onCheckpointBarrier(sink ExactlyOnceSink, txn Txn, checkpointID int64) (Txn, error) {
if err := sink.PreCommit(txn); err != nil {
sink.Abort(txn)
return Txn{}, err
}
rememberPendingTransaction(checkpointID, txn)
return sink.BeginTransaction(checkpointID + 1)
}
func onCheckpointComplete(sink ExactlyOnceSink, checkpointID int64) error {
for _, txn := range pendingTransactionsUpTo(checkpointID) {
if err := sink.Commit(txn); err != nil {
return err
}
forgetPendingTransaction(txn)
}
return nil
}For non-transactional sinks, exactly-once is normally implemented with idempotent writes. The event carries a stable operation identifier derived from the input topic, partition, offset, or business event ID. The sink stores the identifier with the effect and refuses to apply it twice. This pattern is weaker than a full distributed transaction but is often the most reliable design for databases, object stores, and APIs.
Counts, balances, inventory reservations, ledger projections, fraud features, and entitlement tables cannot tolerate duplicate increments. Exactly-once stream processing is useful when a duplicate record would inflate a counter, double-apply a credit, or mark an item unavailable twice. The stream engine should make state updates and emitted events transactional, while final sinks should still store event IDs or ledger entry IDs.
A service can write business rows and outbox rows in one database transaction, then use CDC to publish the outbox to Kafka. Downstream processors can use Kafka transactions to transform those events into derived topics. This composes two important guarantees: the source application avoids dual writes, and the stream processor atomically commits outputs with consumed offsets.
Joins and aggregations maintain local state. A failure during recovery can otherwise replay one input side and not the other, producing inconsistent results. Checkpointed state plus transactional output lets the job restore source offsets and join state together, then continue from a consistent boundary.
Exactly-once techniques also support controlled reprocessing. A team can deploy corrected logic, replay retained input into a new output topic or table, and compare outputs without corrupting the live projection. Stable operation IDs and deterministic transformations make replay a normal operational tool instead of a one-off migration.
When the sink database cannot participate in Kafka or Flink transactions, store an operation key in the same database transaction as the side effect. The operation key should be stable across retries. Topic-partition-offset works for replaying the same source log; a business event ID is better when the same event may be copied across topics or regenerated during backfills.
CREATE TABLE applied_events (
event_id text PRIMARY KEY,
applied_at timestamptz NOT NULL DEFAULT now()
);
CREATE TABLE account_balances (
account_id text PRIMARY KEY,
cents bigint NOT NULL
);
-- Apply the effect and remember the event atomically.
BEGIN;
INSERT INTO applied_events(event_id)
VALUES ($1)
ON CONFLICT DO NOTHING;
UPDATE account_balances
SET cents = cents + $2
WHERE account_id = $3
AND EXISTS (SELECT 1 FROM applied_events WHERE event_id = $1);
COMMIT;For a custom consumer writing to one database, keep the last applied source offset in the sink transaction. On restart, resume from the stored offset rather than from an independently committed broker offset. This is a single-system transaction, so it is much simpler than trying to coordinate a database commit and a broker commit after the fact.
func applyPartitionBatch(tx *sql.Tx, partition int, records []Record) error {
last := loadSinkOffset(tx, partition)
for _, r := range records {
if r.Offset <= last {
continue
}
if err := applyRecord(tx, r); err != nil {
return err
}
last = r.Offset
}
return storeSinkOffset(tx, partition, last)
}Kafka producers need a stable transactional ID per task, idempotence enabled, acknowledgements from all in-sync replicas, and careful timeout configuration. Consumers of transactional output should use read-committed isolation when they must hide aborted records. Kafka Streams packages most of this behind processing guarantees, but custom clients need explicit configuration and failure tests.
TwoPhaseCommitSinkFunction abstraction for end-to-end exactly-once sinks. Kafka 0.11 provided the broker-side transactions that made Kafka sinks participate in that protocol.