---
title: "End-to-End Exactly-Once Processing in Apache Flink with Apache Kafka"
description: "Explains how Apache Flink extends exactly-once semantics beyond the application to external systems: checkpoint barriers drive a pre-commit phase and a commit phase, so Kafka transactions are committed only when a checkpoint completes. In current Flink versions the KafkaSink provides this with DeliveryGuarantee.EXACTLY_ONCE."
lastUpdated: 2026-10-14T09:41:47.000Z
source_url:
  html: "https://www.ververica.com/developers/flink-exactly-once-kafka"
  md: "https://www.ververica.com/developers/flink-exactly-once-kafka.md"
---

_This post is an adaptation of Piotr Nowojski’s presentation from Flink Forward Berlin 2017._

Apache Flink 1.4.0, released in December 2017, introduced a significant milestone for stream processing with Flink: a new feature called `TwoPhaseCommitSinkFunction` ([relevant Jira here](https://issues.apache.org/jira/browse/FLINK-7210)) that extracts the common logic of the two-phase commit protocol and makes it possible to build end-to-end exactly-once applications with Flink and a selection of data sources and sinks, including Apache Kafka versions 0.11 and beyond.

Throughout the rest of this post, we’ll:

- Describe the role of Flink’s checkpoints for guaranteeing exactly-once results within a Flink application
- Show how Flink interacts with data sources and data sinks via the two-phase commit protocol to deliver _end-to-end_ exactly-once guarantees

## Exactly-once Semantics Within an Apache Flink Application

When we say “exactly-once semantics,” what we mean is that each incoming event affects the final results exactly once. Even in case of a machine or software failure, there’s no duplicate data and no data that goes unprocessed. In [real-time payments processing](https://www.ververica.com/banking/real-time-payments), this is what keeps a payment from being posted twice.

Flink has long provided exactly-once semantics _within_ a Flink application. Over the past few years, we’ve [written in depth about Flink’s checkpointing](https://www.ververica.com/developers/flink-high-throughput-low-latency), which is at the core of Flink’s ability to provide exactly-once semantics. The Flink documentation also [provides a thorough overview of the feature](https://nightlies.apache.org/flink/flink-docs-stable/docs/ops/state/checkpoints/).

Before we continue, here’s a quick summary of the checkpointing algorithm because understanding checkpoints is necessary for understanding this broader topic.

A checkpoint in Flink is a consistent snapshot of:

1. The current state of an application
1. The position in an input stream

Flink generates checkpoints on a regular, configurable interval and then writes the checkpoint to a persistent storage system, such as S3 or HDFS. Writing the checkpoint data to the persistent storage happens asynchronously, which means that a Flink application continues to process data during the checkpointing process.

In the event of a machine or software failure and upon restart, a Flink application resumes processing from the most recent successfully-completed checkpoint; Flink restores application state and rolls back to the correct position in the input stream from a checkpoint before processing starts again. This means that Flink computes results as though the failure never occurred.

Before Flink 1.4.0, exactly-once semantics were limited to the scope of _a Flink application only_ and did not extend to most of the external systems to which Flink sends data after processing. But Flink applications operate in conjunction with a wide range of data sinks, and developers should be able to maintain exactly-once semantics beyond the context of one component.

To provide _end-to-end exactly-once_ semantics--that is, semantics that also apply to the external systems that Flink writes to in addition to the Flink application--these external systems must provide a means to commit or roll back writes that coordinate with Flink's checkpoints.

One common approach for coordinating commits and rollbacks in a distributed system is the [two-phase commit protocol](https://en.wikipedia.org/wiki/Two-phase_commit_protocol). In the next section, we’ll go behind the scenes and discuss how Flink utilizes the two-phase commit protocol to provide end-to-end exactly-once semantics.

## End-to-end Exactly Once Applications with Apache Flink

We’ll walk through the two-phase commit protocol and how it enables end-to-end exactly-once semantics in a sample Flink application that reads from and writes to Kafka. Kafka is a popular messaging system to use along with Flink, and Kafka added support for transactions with its 0.11 release. [This means that Flink now has the necessary mechanism to provide end-to-end exactly-once semantics](https://nightlies.apache.org/flink/flink-docs-stable/docs/connectors/datastream/kafka/#fault-tolerance) in applications when receiving data from and writing data to Kafka.

Flink’s support for end-to-end exactly-once semantics is not limited to Kafka and you can use it with any source / sink that provides the necessary coordination mechanism.

![Diagram: sample application with a Kafka source, a window operator and a Kafka sink, plus the JobManager and the state backend](https://cdn.sanity.io/images/6b38iw1w/production/1e1df756b56e632836b9537932dcc4ed351326de-599x336.webp)

In the sample Flink application that we’ll discuss today, we have:

- A data source that reads from Kafka (in Flink, a [KafkaSource](https://nightlies.apache.org/flink/flink-docs-stable/docs/connectors/datastream/kafka/#kafka-source))
- A windowed aggregation
- A data sink that writes data back to Kafka (in Flink, a [KafkaSink](https://nightlies.apache.org/flink/flink-docs-stable/docs/connectors/datastream/kafka/#kafka-sink))

For the data sink to provide exactly-once guarantees, it must write all data to Kafka within the scope of a transaction. A commit bundles all writes between two checkpoints. This ensures that writes are rolled back in case of a failure.

However, in a distributed system with multiple, concurrently-running sink tasks, a simple commit or rollback is not sufficient, because all of the components must “agree” together on committing or rolling back to ensure a consistent result. Flink uses the two-phase commit protocol and its pre-commit phase to address this challenge.

The starting of a checkpoint represents the “pre-commit” phase of our two-phase commit protocol. When a checkpoint starts, the Flink JobManager injects a checkpoint barrier (which separates the records in the data stream into the set that goes into the current checkpoint vs. the set that goes into the next checkpoint) into the data stream.

The barrier is passed from operator to operator. For every operator, it triggers the operator’s state backend to take a snapshot of its state.

![Diagram: pre-commit phase starts when the JobManager injects a checkpoint barrier at the data source](https://cdn.sanity.io/images/6b38iw1w/production/e76b76144b0882d526e02d61c9aae3ced09b40c2-600x336.webp)

The data source stores its Kafka offsets, and after completing this, it passes the checkpoint barrier to the next operator.

This approach works if an operator has internal state _only_. **Internal state** is everything that can be stored and managed by Flink’s state - for example, the windowed sums in the second operator. When a process has only internal state, there is no need to perform any additional action during pre-commit aside from writing to defined state variables. Flink takes care of correctly committing those writes in case of checkpoint success or aborting them in case of failure.

![Diagram: pre-commit without external state, the data source snapshots its offsets and passes the barrier on](https://cdn.sanity.io/images/6b38iw1w/production/0354cd0f9267f6d7d984ff4a06734439fdc99594-599x337.webp)

However, when a process has _external_ state, this state must be handled a bit differently. External state usually comes in the form of writes to an external system such as Kafka. In that case, to provide exactly-once guarantees, the external system must provide support for transactions that integrates with a two-phase commit protocol.

We know that the data sink in our example has such external state because it’s writing data to Kafka. In this case, in the pre-commit phase, the data sink must pre-commit its external transaction in addition to writing its state to the state backend.

![Diagram: pre-commit with external state, the data sink pre-commits its Kafka transaction while operators snapshot their state](https://cdn.sanity.io/images/6b38iw1w/production/574dda0ee535d83668b5f26fef9a8f8e0c84d2ff-599x337.webp)

The pre-commit phase finishes when the checkpoint barrier passes through all of the operators and the triggered snapshot callbacks complete successfully. All of the triggered state snapshots are considered part of that checkpoint. The checkpoint is a snapshot of the state of the entire application, including pre-committed external state. In case of a failure, we can roll back to the point of time of the last successfully-completed snapshot.

The next step is to notify all operators that the checkpoint has succeeded. This is the commit phase of the two-phase commit protocol and the JobManager issues checkpoint-completed callbacks for every operator in the application.

The data source and window operator have no external state, and so in the commit phase, these operators don’t have to take any action. The data sink does have external state, though, and it’s at this point that the external transaction should be committed:

![Diagram: commit phase, the JobManager notifies all operators that the checkpoint completed and the data sink commits the Kafka transaction](https://cdn.sanity.io/images/6b38iw1w/production/854aa4ea2b0b52c108412054a191618d69f6b086-600x336.webp)

So let’s put all of these different pieces together:

- Once all of the operators complete their pre-commit, they issue a commit.
- If at least one pre-commit fails, all others are aborted, and we roll back to the previous successfully-completed checkpoint.
- After a successful pre-commit, the commit _must_ be guaranteed to eventually succeed -- both our operators and our external system need to make this guarantee. If a commit fails (for example, due to an intermittent network issue), the entire Flink application fails, the application restarts according to the user’s restart strategy, and there is another commit attempt. This process is critical because if the commit does not eventually succeed, data loss occurs.

Therefore, we can be sure that all operators agree on the final outcome of the checkpoint: all operators agree that the data is either committed or that the commit is aborted and rolled back.

In current Flink versions, the `KafkaSink` implements this protocol for you: built with `DeliveryGuarantee.EXACTLY_ONCE` and a transactional ID prefix set via `setTransactionalIdPrefix(...)`, it writes all records in a Kafka transaction that is committed when a checkpoint completes (see the [Kafka connector documentation](https://nightlies.apache.org/flink/flink-docs-stable/docs/connectors/datastream/kafka/#fault-tolerance)).

## Wrapping Up

If you’ve made it this far, thanks for staying with us through a detailed post. Here are some key points that we covered:

- Flink’s checkpointing system serves as Flink’s basis for supporting a two-phase commit protocol and providing end-to-end exactly-once semantics.
- An advantage of this approach is that Flink does not materialize data in transit the way that some other systems do--there’s no need to write every stage of the computation to disk as is the case in most batch processing.
- Starting with [Flink 1.4.0](https://flink.apache.org/2017/12/12/apache-flink-1.4.0-release-announcement/), both the Pravega and Kafka 0.11 producers provide exactly-once semantics; Kafka introduced transactions for the first time in Kafka 0.11, which is what made the Kafka exactly-once producer possible in Flink.

## Frequently Asked Questions

### What does exactly-once semantics mean in Apache Flink?

Exactly-once semantics means each incoming event affects the final results exactly once, with no duplicate data and no unprocessed data even after a machine or software failure. Within a Flink application this is provided by checkpointing, which snapshots application state and input positions.

### How does Flink achieve end-to-end exactly-once with Kafka?

Flink combines its checkpoints with a two-phase commit protocol: when a checkpoint starts, the Kafka sink pre-commits its open transaction, and when the checkpoint completes, the transaction is committed. If any pre-commit fails, everything rolls back to the last successfully completed checkpoint.

### How do I configure the Flink KafkaSink for exactly-once delivery?

Build the KafkaSink with DeliveryGuarantee.EXACTLY_ONCE and set a transactional ID prefix with setTransactionalIdPrefix(...). The sink then writes all records in a Kafka transaction that is committed when a checkpoint completes.

### What requirements must an external system meet for end-to-end exactly-once with Flink?

The external system Flink writes to must be able to commit or roll back writes in coordination with Flink's checkpoints, typically through transactions that integrate with a two-phase commit protocol. Flink's support is not limited to Kafka and works with any source or sink that provides this coordination mechanism.

### What happens if a commit fails in Flink's two-phase commit?

After a successful pre-commit, the commit must eventually succeed. If it fails, for example because of an intermittent network issue, the Flink application fails, restarts according to the restart strategy, and attempts the commit again; otherwise data loss would occur.