Implementing Exactly-Once Delivery Guarantees Across Kafka, Flink, and Sink Systems: Transactions, Offsets, and Idempotent Writes in Practice
Introduction
Picture this: your e-commerce platform processes payment events from a Kafka topic, enriches them in Flink, and writes the results to a PostgreSQL database and an S3 data lake simultaneously. The Flink job crashes mid-checkpoint. When it restarts, does it double-count revenue? Write partial records? Silently skip transactions? If you can't answer those questions with certainty, you don't have exactly-once delivery — you have a polite fiction that holds up until the worst possible moment.
Exactly-once semantics is one of the most misunderstood concepts in distributed systems, and the gap between theory and production reality is enormous. Most teams believe they have it because they've enabled a config flag or two. What they actually have is a system that mostly works, with subtle failure modes lurking in edge cases: broker leader elections, Flink task manager crashes, slow network flushes, and sink connector misconfiguration. When those failure modes surface, they do so in production, usually during a high-traffic event, and usually in ways that are painful to detect and fix.
By the end of this lesson, you'll understand how exactly-once delivery actually works across the full pipeline stack — from Kafka producers and consumer groups through Flink's checkpointing machinery to idempotent and transactional sinks. You'll be able to configure each layer correctly, diagnose the gaps between them, and make deliberate architectural decisions about where to accept trade-offs.
What you'll learn:
- How Kafka's idempotent producers and transactional API work at the protocol level, and where they fall short
- How Flink's two-phase commit checkpointing integrates with Kafka to achieve end-to-end exactly-once
- How to implement idempotent and transactional writes to diverse sink systems: PostgreSQL, S3, Elasticsearch, and external HTTP APIs
- Common failure scenarios that break exactly-once guarantees silently, and how to detect them
- The real performance cost of exactly-once and when to consciously trade it away
Prerequisites
You should be comfortable with:
Why "At-Least-Once" Is the Invisible Default
Before we can appreciate exactly-once, we need to be precise about what the alternatives actually mean in practice.
At-most-once means records can be lost but never duplicated. You commit offsets before processing. If the process dies after committing but before completing the work, those records are gone. This is the default behavior if you naively set enable.auto.commit=true in your Kafka consumer with a short commit interval and then do expensive downstream work.
At-least-once means records are never lost but can be duplicated. You commit offsets only after successfully processing. If the process dies after processing but before committing, the records will be reprocessed on restart. This is what most "production-ready" pipelines actually implement, because it's relatively easy to achieve and data loss is usually more dangerous than duplication.
Exactly-once means each record is processed and reflected in the output exactly one time, even across failures and restarts. This is hard because it requires coordination across multiple systems that don't share a transaction boundary.
The insidious thing about at-least-once is that it looks like exactly-once during normal operation. Duplicates only appear during failure recovery, which is exactly when your observability is strained and your team is under pressure. If you're not measuring duplicate rates on your sinks, you probably don't know what semantics you actually have.
Warning: Enabling Kafka's enable.auto.commit=true with default settings gives you at-most-once delivery if your consumer crashes during a long processing operation. Many tutorials recommend this configuration for simplicity. Don't use it in production pipelines where data loss is unacceptable.
Kafka's Exactly-Once Building Blocks
Kafka's exactly-once semantics (EOS) is built on two features that work together: the idempotent producer and the transactional API. Understanding each independently before combining them is essential.
Idempotent Producers
Without idempotency, Kafka producers can create duplicates in a specific failure scenario: the producer sends a message batch, the broker writes it and sends an acknowledgment, but the network drops the ack before it reaches the producer. The producer, seeing no ack and facing a timeout, retries — and the broker now has two copies of the same batch.
The idempotent producer solves this with two identifiers:
- Producer ID (PID): A unique 64-bit integer assigned by the broker when the producer first connects
- Sequence number: A monotonically increasing counter maintained per (PID, topic-partition) pair
When the broker receives a batch, it checks the sequence number. If the sequence number is equal to the last accepted sequence number plus one, it's a new message — accept it. If the sequence number is less than or equal to the last accepted, it's a duplicate — silently drop it. If it's greater by more than one, something went out of order — raise an error.
Enable idempotency in your producer like this:
from confluent_kafka import Producer
producer = Producer({
'bootstrap.servers': 'kafka-broker:9092',
'enable.idempotence': True, # Enables PID assignment and sequence tracking
'acks': 'all', # Required for idempotence - auto-set when enabled
'max.in.flight.requests.per.connection': 5, # Max 5 per Kafka spec; auto-set
'retries': 2147483647, # Effectively infinite retries; auto-set
})
Note: Setting enable.idempotence=True automatically configures acks=all, retries=Integer.MAX_VALUE, and max.in.flight.requests.per.connection=5. You don't need to set those separately, but knowing their values is important for reasoning about behavior.
The critical limitation: idempotency is scoped to a single producer session. If the producer process restarts, it gets a new PID, and the broker has no memory of previous sequence numbers. A restart followed by a retry will produce duplicates from the broker's perspective. This is why idempotency alone is insufficient for exactly-once across failures.
The Transactional API
Kafka's transactional API extends idempotency with two additional capabilities:
- Atomic multi-partition writes: A set of messages to multiple topic-partitions either all commit or all roll back, atomically
- Consumer offset commits within a transaction: You can atomically write output messages and commit the input consumer offsets in a single transaction, which is the foundation of the "read-process-write" exactly-once pattern
The transactional workflow:
from confluent_kafka import Producer, Consumer, KafkaException
# Producer with transactional ID - this is persistent across restarts
producer = Producer({
'bootstrap.servers': 'kafka-broker:9092',
'enable.idempotence': True,
'transactional.id': 'payment-enricher-v1', # MUST be stable across restarts
'transaction.timeout.ms': 60000,
})
consumer = Consumer({
'bootstrap.servers': 'kafka-broker:9092',
'group.id': 'payment-enricher-group',
'auto.offset.reset': 'earliest',
'enable.auto.commit': False, # We'll commit offsets inside the transaction
'isolation.level': 'read_committed', # Only read committed messages
})
producer.init_transactions()
consumer.subscribe(['payments.raw'])
while True:
messages = consumer.consume(num_messages=500, timeout=1.0)
if not messages:
continue
try:
producer.begin_transaction()
offsets_to_commit = {}
for msg in messages:
if msg.error():
raise KafkaException(msg.error())
# Process the message
enriched = enrich_payment(msg.value())
# Write to output topic within the transaction
producer.produce('payments.enriched', value=enriched)
# Track which offsets we're consuming
tp = TopicPartition(msg.topic(), msg.partition(), msg.offset() + 1)
offsets_to_commit[f"{msg.topic()}-{msg.partition()}"] = tp
# Commit offsets as part of the transaction - this is the magic
producer.send_offsets_to_transaction(
list(offsets_to_commit.values()),
consumer.consumer_group_metadata()
)
producer.commit_transaction()
except Exception as e:
producer.abort_transaction()
# Don't commit consumer offsets - messages will be reprocessed
raise
The transactional.id is the linchpin. When a producer with a given transactional.id initializes, the broker looks up any pending (uncommitted, unaborted) transactions from a previous producer with the same ID and fences them — preventing the zombie producer from completing a transaction that the new producer will reprocess. This handles the restart case that idempotency alone can't.
Key insight: The transactional.id must be stable and unique per logical partition of work. In a distributed consumer with 8 partitions being consumed by 4 workers, each worker needs a distinct transactional ID that maps to its assigned partitions. A common pattern is {app-name}-{partition-range}. If two workers share a transactional ID, they'll fence each other, causing transaction aborts.
Consumer-Side: `isolation.level`
There's one more piece that's easy to miss on the consumer side. When you write transactionally to Kafka, the messages are physically written to the log immediately but are not visible to consumers until the transaction commits. This is controlled by isolation.level:
read_uncommitted (default): consumers see all messages, including those in open or aborted transactions. This breaks the EOS guarantee for downstream consumers.read_committed: consumers only see messages from committed transactions and non-transactional messages. Aborted transaction messages are skipped.
If your Flink job or downstream consumer reads from an output topic produced by a transactional producer and doesn't set isolation.level=read_committed, it will read messages that may later be aborted — which means it's processing data that doesn't exist from the transaction's perspective.
Flink's Two-Phase Commit: The Checkpoint-Transaction Bridge
Flink's approach to exactly-once is architecturally elegant: it integrates Kafka transaction commits with Flink's own checkpointing mechanism using a two-phase commit (2PC) protocol.
If you're not deeply familiar with Flink checkpointing, the short version is: Flink periodically injects "checkpoint barriers" into the data stream, which flow through the operator DAG. When a barrier reaches an operator, that operator saves its state snapshot and forwards the barrier. When all operators have acknowledged a checkpoint, it's considered complete and durable. On restart, Flink restores state from the last complete checkpoint and replays input from the corresponding Kafka offsets.
The two-phase commit wraps around this:
Phase 1 — Pre-commit: When a checkpoint barrier is received, the Kafka sink operator calls producer.flush() (flushing all in-flight messages to the broker but not committing the transaction) and then snapshots the transaction state (the producer ID, epoch, and the set of partitions involved). This snapshot is stored in Flink's state backend as part of the checkpoint.
Phase 2 — Commit: When Flink notifies all operators that the checkpoint is complete (meaning all state is durably stored), the Kafka sink operator calls producer.commit_transaction(). The messages that were pre-committed are now visible to read_committed consumers.
If a failure occurs between Phase 1 and Phase 2 (after pre-commit but before the commit notification), Flink recovers from the checkpoint. The transaction state is restored from the snapshot, and the recovered operator re-issues the commit. The broker still has the pre-committed messages, so the commit succeeds without re-sending any data.
If a failure occurs before Phase 1 completes (during data processing), Flink recovers from the previous checkpoint. The incomplete transaction is aborted (because the transactional.id epoch is bumped on recovery, fencing the old producer), and the data is reprocessed from the Kafka source offsets that were captured in that previous checkpoint.
This is why exactly-once in Flink requires all three to be configured correctly: the Kafka source (offsets captured in checkpoints), the Flink checkpoint itself (durable state backend), and the Kafka sink (transactional writes coordinated with checkpoint lifecycle).
Configuring the Flink Kafka Connector
Here's a realistic Flink job configuration for end-to-end exactly-once:
import org.apache.flink.api.common.restartstrategy.RestartStrategies;
import org.apache.flink.api.common.time.Time;
import org.apache.flink.connector.kafka.sink.KafkaRecordSerializationSchema;
import org.apache.flink.connector.kafka.sink.KafkaSink;
import org.apache.flink.connector.kafka.source.KafkaSource;
import org.apache.flink.connector.kafka.source.enumerator.initializer.OffsetsInitializer;
import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
// Checkpoint configuration - this is non-negotiable for EOS
env.enableCheckpointing(30_000L, CheckpointingMode.EXACTLY_ONCE);
env.getCheckpointConfig().setMinPauseBetweenCheckpoints(5_000L);
env.getCheckpointConfig().setCheckpointTimeout(120_000L);
env.getCheckpointConfig().setMaxConcurrentCheckpoints(1); // Required for EOS
// Restart strategy - bounded retries prevent infinite restart loops
env.setRestartStrategy(
RestartStrategies.fixedDelayRestart(5, Time.seconds(30))
);
// Source - checkpointing captures the offset position automatically
KafkaSource<PaymentEvent> source = KafkaSource.<PaymentEvent>builder()
.setBootstrapServers("kafka-broker:9092")
.setTopics("payments.raw")
.setGroupId("payment-flink-job")
.setStartingOffsets(OffsetsInitializer.committedOffsets())
.setDeserializer(new PaymentEventDeserializer())
.setProperty("isolation.level", "read_committed") // Critical!
.build();
// Sink - transactional.id.prefix + subtask index = unique transactional ID per task
KafkaSink<EnrichedPayment> sink = KafkaSink.<EnrichedPayment>builder()
.setBootstrapServers("kafka-broker:9092")
.setRecordSerializer(
KafkaRecordSerializationSchema.builder()
.setTopic("payments.enriched")
.setValueSerializationSchema(new EnrichedPaymentSerializer())
.build()
)
.setDeliveryGuarantee(DeliveryGuarantee.EXACTLY_ONCE)
.setTransactionalIdPrefix("payment-enricher") // Each task appends its ID
.setKafkaProducerConfig(producerProps)
.build();
DataStream<PaymentEvent> payments = env.fromSource(
source, WatermarkStrategy.noWatermarks(), "Kafka Payment Source"
);
payments
.map(new EnrichmentFunction())
.sinkTo(sink);
Warning: setMaxConcurrentCheckpoints(1) is required for exactly-once with the Kafka sink. If two checkpoints are in-flight simultaneously, you can have two transactions open simultaneously, and the two-phase commit protocol breaks down because the sink can't maintain the correct transaction boundary per checkpoint.
Transaction Timeout: The Silent Killer
One configuration that causes mysterious failures is transaction.timeout.ms. The Kafka broker has a maximum transaction timeout (transaction.max.timeout.ms, default 15 minutes). If a Flink checkpoint takes longer than the transaction timeout, the broker will abort the transaction — and you'll see a ProducerFencedException or InvalidProducerEpochException when Flink tries to commit it.
The implication: your transaction.timeout.ms must be greater than your maximum expected checkpoint duration, including checkpoint timeout. If checkpoints can take up to 2 minutes and you're using the default transaction.timeout.ms of 60 seconds, you have a time bomb.
// In your KafkaSink producer properties
Properties producerProps = new Properties();
producerProps.setProperty("transaction.timeout.ms", "300000"); // 5 minutes
// Also increase on the broker if needed:
// transaction.max.timeout.ms=600000
Sink Systems: The Hard Part Nobody Talks About
Getting Kafka-to-Kafka exactly-once right is relatively well-documented. The much harder problem is implementing exactly-once to sink systems that aren't Kafka — your relational database, your object storage lake, your search index. Each requires a different strategy.
PostgreSQL: Transactional Upserts
PostgreSQL is ACID-compliant, which makes it the easiest non-Kafka sink to get right. The core pattern is idempotent upsert: your write operation is designed so that executing it twice produces the same result as executing it once.
For a payment records table:
-- Schema with a natural idempotency key
CREATE TABLE payment_records (
payment_id UUID PRIMARY KEY,
order_id UUID NOT NULL,
amount_cents BIGINT NOT NULL,
status VARCHAR(50) NOT NULL,
processed_at TIMESTAMPTZ NOT NULL,
pipeline_run_id UUID NOT NULL, -- Audit: which pipeline run wrote this
updated_at TIMESTAMPTZ DEFAULT NOW()
);
-- Idempotent upsert: re-running with same payment_id is safe
INSERT INTO payment_records (payment_id, order_id, amount_cents, status, processed_at, pipeline_run_id)
VALUES (%s, %s, %s, %s, %s, %s)
ON CONFLICT (payment_id)
DO UPDATE SET
status = EXCLUDED.status,
processed_at = EXCLUDED.processed_at,
pipeline_run_id = EXCLUDED.pipeline_run_id,
updated_at = NOW()
WHERE payment_records.status != EXCLUDED.status -- Only update if something changed
;
This ON CONFLICT DO UPDATE pattern (PostgreSQL's UPSERT) means that if Flink reprocesses a payment event after a restart, the second write is a no-op (or a safe overwrite) rather than a duplicate row.
The batch version, which is far more efficient for Flink sinks:
import psycopg2
from psycopg2.extras import execute_values
def write_payment_batch(conn, records: list[dict]):
"""
Writes a batch of payment records with idempotent upsert semantics.
Safe to call multiple times with the same records.
"""
with conn.cursor() as cur:
execute_values(
cur,
"""
INSERT INTO payment_records
(payment_id, order_id, amount_cents, status, processed_at, pipeline_run_id)
VALUES %s
ON CONFLICT (payment_id) DO UPDATE SET
status = EXCLUDED.status,
processed_at = EXCLUDED.processed_at,
pipeline_run_id = EXCLUDED.pipeline_run_id,
updated_at = NOW()
""",
[(r['payment_id'], r['order_id'], r['amount_cents'],
r['status'], r['processed_at'], r['pipeline_run_id'])
for r in records],
template="(%s, %s, %s, %s, %s, %s)"
)
conn.commit()
Tip: Always include a pipeline_run_id or similar audit column in your sink tables. When you need to investigate whether a duplicate write occurred, this column tells you which pipeline execution wrote each row. It's the difference between a 5-minute investigation and a 2-hour one. This ties into broader data quality validation, testing, and monitoring practices.
For aggregated or derived tables where upsert semantics don't naturally apply (like a daily_revenue rollup), you need a different approach: write-and-replace with a transactional swap.
BEGIN;
-- Write new data to a staging table
INSERT INTO daily_revenue_staging SELECT ... FROM payment_records WHERE ...;
-- Atomic swap - downstream queries never see a partial state
DELETE FROM daily_revenue WHERE report_date = '2024-01-15';
INSERT INTO daily_revenue SELECT * FROM daily_revenue_staging WHERE report_date = '2024-01-15';
TRUNCATE daily_revenue_staging;
COMMIT;
S3 / Object Storage: The Atomic Rename Pattern
Object storage systems like S3, GCS, and Azure Blob Storage don't support transactions in the relational sense. The standard exactly-once pattern for object storage is the atomic rename (or copy-and-delete) approach, combined with a staging prefix.
The idea:
- Write output files to a
_staging/ prefix with a checkpoint-specific path
- On checkpoint completion, atomically "commit" by moving (renaming) files from
_staging/ to the final prefix
In practice, S3 doesn't have atomic rename — you copy then delete. The key is making the commit phase idempotent: if you crash mid-commit, you can recover by re-running the commit using the checkpoint state that recorded which files need to be moved.
Flink's FileSink with EXACTLY_ONCE rolling policy implements exactly this:
import org.apache.flink.connector.file.sink.FileSink;
import org.apache.flink.core.fs.Path;
import org.apache.flink.formats.parquet.avro.ParquetAvroWriters;
import org.apache.flink.streaming.api.functions.sink.filesystem.rollingpolicies.OnCheckpointRollingPolicy;
FileSink<EnrichedPayment> fileSink = FileSink
.forBulkFormat(
new Path("s3://my-data-lake/payments/enriched/"),
ParquetAvroWriters.forReflectRecord(EnrichedPayment.class)
)
.withRollingPolicy(OnCheckpointRollingPolicy.build()) // Roll files on checkpoint
.withOutputFileConfig(
OutputFileConfig.builder()
.withPartPrefix("part")
.withPartSuffix(".parquet")
.build()
)
.build();
The OnCheckpointRollingPolicy ensures files are only finalized (moved from in-progress to committed state) when a checkpoint completes. In-progress files use a .inprogress suffix, and Flink tracks them in checkpoint state. On recovery, uncommitted in-progress files are either committed (if the checkpoint that should have committed them completed) or deleted (if they belong to a failed checkpoint).
Warning: The S3 implementation has a subtle gotcha: S3's eventual consistency model (now largely resolved in AWS, but still relevant for some S3-compatible stores like MinIO or older GCS) means that a file you just wrote might not be immediately visible to a subsequent list operation. If your commit phase relies on listing files to move, you may miss recently written files. Use explicit tracking in Flink's checkpoint state rather than directory listing.
Elasticsearch: The Idempotency Key Approach
Elasticsearch doesn't support multi-document transactions (except for the limited painless scripting approach). Exactly-once writes to Elasticsearch are achieved through document ID idempotency: every document you write has a deterministic, content-derived ID.
When you index a document with a specific _id, Elasticsearch uses the last-write-wins semantics for that ID. If Flink reprocesses a message and re-indexes the same document, the second write replaces the first with identical content — net effect is exactly once.
from elasticsearch import Elasticsearch
from elasticsearch.helpers import bulk
import hashlib
def generate_idempotent_id(payment_id: str, event_type: str, pipeline_version: str) -> str:
"""
Deterministic document ID based on business key + pipeline version.
Including pipeline_version allows intentional re-indexing during migrations.
"""
key = f"{payment_id}:{event_type}:{pipeline_version}"
return hashlib.sha256(key.encode()).hexdigest()[:24]
def index_payments_bulk(es: Elasticsearch, records: list[dict]):
actions = [
{
'_index': 'payments_enriched',
'_id': generate_idempotent_id(
r['payment_id'],
r['event_type'],
'v2'
),
'_source': r,
# Use 'index' (not 'create') to allow overwrites
'_op_type': 'index',
}
for r in records
]
success, errors = bulk(es, actions, raise_on_error=False)
if errors:
# Log but don't necessarily fail - version conflicts are expected
# during recovery and are not true errors
non_conflict_errors = [
e for e in errors
if e.get('index', {}).get('status') != 409
]
if non_conflict_errors:
raise RuntimeError(f"Elasticsearch index errors: {non_conflict_errors}")
return success
Key insight: The choice between _op_type: 'create' and _op_type: 'index' matters for idempotency. create fails if the document already exists (version conflict), which is correct for "write-once" semantics but will cause your recovery to fail with errors on retry. index silently overwrites, which is correct for idempotent re-processing. Use create only if you have a separate deduplication layer and want explicit failure on duplicates.
HTTP APIs: The Inherently Non-Transactional Sink
Writing to external HTTP APIs is the hardest case, because APIs rarely provide transactional guarantees. The best you can usually achieve is idempotency keys combined with implementing dead letter queues for poison messages to handle unrecoverable failures.
import httpx
import uuid
def post_payment_webhook_idempotent(
client: httpx.Client,
payment: dict,
idempotency_key: str # Derived from payment_id + pipeline_run_id
) -> dict:
"""
Posts a payment event to an external API with idempotency key.
The API must support idempotency keys (Stripe-style) for this to work.
"""
response = client.post(
'https://api.partner.com/v1/payment-events',
json=payment,
headers={
'Idempotency-Key': idempotency_key,
'Content-Type': 'application/json',
},
timeout=10.0
)
if response.status_code == 200:
return response.json()
elif response.status_code == 409:
# Idempotency key already used - this is a safe duplicate, return cached response
return response.json()
elif response.status_code in (400, 422):
# Unrecoverable - route to dead letter queue, don't retry
raise UnrecoverableAPIError(f"Bad request: {response.text}")
else:
# Transient failure - raise to trigger retry
raise TransientAPIError(f"API error {response.status_code}: {response.text}")
The Idempotency-Key header (popularized by Stripe's API) is a server-side deduplication mechanism. If the server stores the result of a request by idempotency key and returns the cached result on retry, the operation becomes idempotent. But this only works if the downstream API supports it — many don't.
For APIs without idempotency support, you're left with "best effort" semantics. The practical mitigation is to record successful API calls in a separate durable store (like a Redis set or a database table) keyed by the message ID, and skip re-calling the API if the record already exists. This trades API-level idempotency for application-level idempotency.
Distributed Sagas for Multi-Sink Pipelines
When your Flink job writes to multiple sinks simultaneously — say, PostgreSQL and Elasticsearch — you face a new problem: how do you ensure atomicity across sinks that don't share a transaction boundary? If the PostgreSQL write succeeds but the Elasticsearch write fails, you have inconsistent state.
This is the domain of distributed transactions, and the practical answer for most data pipelines is the Saga pattern: rather than two-phase commit across heterogeneous systems (which is fragile and rarely supported), you break the multi-sink write into a sequence of local transactions with explicit compensating transactions for failure.
For deeper treatment of this pattern, see cross-system transaction management with distributed sagas. The key design principle: order your sinks from most to least recoverable. Write to the system you can most easily compensate or replay last.
In practice for a Kafka → Flink → (PostgreSQL + Elasticsearch) pipeline:
- Write to PostgreSQL first (fully ACID, easy to check and compensate)
- Track PostgreSQL write success in Flink state
- Write to Elasticsearch using the same idempotency key
- If Elasticsearch fails, the Flink checkpoint will not complete, the PostgreSQL write is rolled back with the transaction, and the whole thing retries from the last good checkpoint
This only works cleanly if your PostgreSQL write is inside a transaction that isn't committed until the checkpoint completes — which requires a two-phase commit integration similar to what Flink's Kafka sink does. Implementing this correctly requires TwoPhaseCommitSinkFunction in Flink.
public class PostgresSinkFunction
extends TwoPhaseCommitSinkFunction<EnrichedPayment, PostgresTransaction, Void> {
@Override
protected PostgresTransaction beginTransaction() throws Exception {
// Open a database connection and begin a transaction
Connection conn = dataSource.getConnection();
conn.setAutoCommit(false);
return new PostgresTransaction(conn);
}
@Override
protected void invoke(PostgresTransaction transaction,
EnrichedPayment payment,
Context context) throws Exception {
// Write to the open (uncommitted) transaction
transaction.insertPayment(payment);
}
@Override
protected void preCommit(PostgresTransaction transaction) throws Exception {
// Flush all pending writes - don't commit yet
transaction.flush();
}
@Override
protected void commit(PostgresTransaction transaction) {
// Called only after checkpoint completes - safe to commit
try {
transaction.getConnection().commit();
} catch (SQLException e) {
// Commit failures here are serious - the checkpoint state says
// we should have committed, but we couldn't
// This is where you need alerting and manual intervention
log.error("CRITICAL: Failed to commit transaction after checkpoint", e);
}
}
@Override
protected void abort(PostgresTransaction transaction) {
try {
transaction.getConnection().rollback();
} catch (SQLException e) {
log.warn("Failed to rollback transaction", e);
}
}
}
Warning: The commit() method in TwoPhaseCommitSinkFunction must be idempotent and must not throw exceptions that would abort the pipeline. By the time commit() is called, Flink has already durably recorded the checkpoint. If commit() fails, Flink will retry it — but if it fails consistently (e.g., a database is permanently down), you're in a state where your checkpoint says "committed" but your sink says "not committed." This requires manual reconciliation. Always alert on commit failures, and design your system so this state is detectable.
Measuring and Monitoring Exactly-Once Guarantees
Implementing exactly-once is only half the job. You need to continuously verify that your guarantees are holding. Exactly-once isn't a binary state you enable once — it degrades silently when configurations drift, broker upgrades change defaults, or application code changes break idempotency assumptions.
Duplicate Detection Metrics
Build duplicate rate monitoring into your sink tables:
-- Detect if any payment_id appears more than once (should never happen)
SELECT
DATE_TRUNC('hour', processed_at) as hour,
COUNT(*) as total_records,
COUNT(DISTINCT payment_id) as unique_payments,
COUNT(*) - COUNT(DISTINCT payment_id) as duplicate_count,
ROUND(
(COUNT(*) - COUNT(DISTINCT payment_id))::numeric / COUNT(*) * 100,
4
) as duplicate_rate_pct
FROM payment_records
WHERE processed_at > NOW() - INTERVAL '24 hours'
GROUP BY 1
ORDER BY 1;
If duplicate_count is ever nonzero on a table that should have exactly-once semantics, you have a breach to investigate.
Kafka Metrics to Watch
Monitor these Kafka producer metrics in your Flink job's metrics system:
# Transaction abort rate - high values indicate EOS issues
kafka.producer.transaction-abort-rate
# Record retry rate - retries aren't duplicates with idempotency, but high rates indicate instability
kafka.producer.record-retry-rate
# In-flight request count - should stay below max.in.flight.requests.per.connection
kafka.producer.requests-in-flight
And on the consumer side:
# Records-lag should drain to near-zero between checkpoints
kafka.consumer.records-lag-max
# Offset commit failures - these indicate your EOS consumer-offset integration is broken
kafka.consumer.commit-rate
This ties closely to the broader observability practices covered in logging, alerting, and observability for data pipelines. Exactly-once failures are notoriously hard to observe through standard logging because the failure mode is "something happened twice," not "an exception was thrown."
Hands-On Exercise
This exercise simulates a realistic failure scenario and walks you through verifying exactly-once behavior under failure conditions.
Setup: You'll need Docker with Kafka, Flink, and PostgreSQL available. The exercise uses a simplified payment processing scenario.
Step 1: Create the Kafka topics and PostgreSQL schema
# Create topics with appropriate replication
kafka-topics.sh --create \
--bootstrap-server localhost:9092 \
--topic payments.raw \
--partitions 4 \
--replication-factor 3 \
--config min.insync.replicas=2
kafka-topics.sh --create \
--bootstrap-server localhost:9092 \
--topic payments.enriched \
--partitions 4 \
--replication-factor 3 \
--config min.insync.replicas=2
CREATE TABLE payment_records (
payment_id UUID PRIMARY KEY,
amount_cents BIGINT NOT NULL,
merchant_id VARCHAR(100) NOT NULL,
status VARCHAR(50) NOT NULL,
processed_at TIMESTAMPTZ NOT NULL,
checkpoint_id BIGINT,
insert_count INT DEFAULT 1
);
Step 2: Generate test data with known payment IDs
import uuid
import json
import time
from confluent_kafka import Producer
producer = Producer({'bootstrap.servers': 'localhost:9092'})
# Generate 1000 payments with deterministic IDs
payment_ids = [str(uuid.uuid5(uuid.NAMESPACE_DNS, f"payment-{i}")) for i in range(1000)]
for i, payment_id in enumerate(payment_ids):
msg = json.dumps({
'payment_id': payment_id,
'amount_cents': (i % 500) * 100 + 1000, # $10 to $510
'merchant_id': f'merchant-{i % 20:03d}',
'event_time': time.time()
})
producer.produce('payments.raw', key=payment_id, value=msg)
if i % 100 == 0:
producer.flush()
print(f"Produced {i+1} messages")
producer.flush()
print("Done producing")
Step 3: Run a Flink job with exactly-once configured (use the Java configuration from earlier in this lesson)
Step 4: Simulate a failure mid-processing
While the job is running (after 200-300 messages have been processed but before completion), kill the Flink TaskManager:
# Find the TaskManager container/process and kill it
docker kill flink-taskmanager-1
Step 5: Let Flink recover and complete processing
Observe the Flink Web UI as the job recovers from the last checkpoint. Watch the source operator rewind its offset.
Step 6: Verify exactly-once in PostgreSQL
-- Should be exactly 1000, with no duplicates
SELECT COUNT(*) as total, COUNT(DISTINCT payment_id) as unique_count
FROM payment_records;
-- Total and unique should be identical
-- If total > unique, you have duplicates
-- If total < 1000, you have losses
-- Distribution check - all merchants should have records
SELECT merchant_id, COUNT(*) as record_count
FROM payment_records
GROUP BY merchant_id
ORDER BY merchant_id;
What to observe: If your exactly-once configuration is correct, total = unique_count = 1000 regardless of how many times the TaskManager was killed. If you see total > unique_count, your sink doesn't have idempotent writes. If you see total < 1000, you have at-most-once behavior somewhere in the stack.
Common Mistakes & Troubleshooting
Mistake 1: Mixed Delivery Guarantees Within a Job
You configure your Kafka source and sink for exactly-once, but you have a stateful operator in the middle (like a window aggregate) that doesn't properly participate in checkpointing. The state is partially applied, and on recovery, the aggregation is inconsistent with the outputs already written.
Fix: Every stateful operator in a Flink job using exactly-once must store its state using Flink's managed state API (ValueState, MapState, etc.). Never use in-memory data structures for state that needs to survive checkpoint/recovery cycles.
Mistake 2: Transactional ID Collisions in Scaled Deployments
You scale your Flink job from 4 to 8 parallelism. The previous transactional IDs (based on subtask index 0-3) now overlap with new subtasks. The new subtasks fence the old ones, causing aborted transactions and processing gaps.
Fix: Include a version or epoch in your transactionalIdPrefix. When changing job parallelism, bump the prefix version: payment-enricher-v3. Alternatively, use Flink's job ID in the prefix, though this requires updating Kafka's transactional.id ACLs.
Mistake 3: Checkpoint Storage Inadequacy
You're using in-memory state backend (MemoryStateBackend) for development and forget to switch to RocksDB + S3 for production. The Flink JM runs out of memory storing large state, checkpoints fail consistently, and the job never makes durable progress.
Fix: Always use EmbeddedRocksDBStateBackend with a durable remote storage path for production exactly-once workloads. See the checkpointing and state management in long-running data pipelines lesson for configuration details.
Mistake 4: Ignoring `NotLeaderOrFollowerException` During Broker Failover
During a Kafka broker leadership election, your transactional producer may receive a NotLeaderOrFollowerException. With a standard retry, this is handled automatically. But if your application catches all exceptions and routes them to a dead letter queue without distinguishing transient from fatal errors, you'll route valid messages to the DLQ and create a false gap in your pipeline.
Fix: Separate your exception handling for transient Kafka exceptions (which should trigger retries, not DLQ routing) from application-level processing errors. The Kafka client library classifies most retriable exceptions with isRetriable().
Mistake 5: The "Write Succeeded, Commit Failed" Deadlock
Your TwoPhaseCommitSinkFunction writes to an external database successfully in preCommit(), but the downstream system is unavailable during commit(). Flink keeps retrying commit() but it never succeeds. The job appears stuck, not failed.
Fix: Implement a circuit breaker in your commit() method. After N consecutive commit failures, alert loudly and consider whether to allow the job to continue processing (which will create an ever-growing backlog of uncommitted transactions) or stop it for manual intervention. For external systems, consider whether you actually need 2PC or whether idempotent writes at the application layer are sufficient (they usually are).
This failure mode is exactly what implementing pipeline circuit breakers is designed to address.
Troubleshooting Checklist
When you suspect exactly-once is broken, work through these systematically:
□ Is isolation.level=read_committed set on ALL consumers downstream of transactional producers?
□ Is transaction.timeout.ms > max checkpoint duration + checkpoint timeout?
□ Is max.concurrent.checkpoints=1 set on the Flink job?
□ Does every transactional.id prefix uniquely identify a task (prefix + subtask index)?
□ Are all stateful operators using Flink managed state, not in-memory structures?
□ Does the sink write operation have idempotent semantics (upsert, not insert)?
□ Are you monitoring duplicate_count and consumer lag metrics?
□ Is the checkpoint storage backend durable (not MemoryStateBackend)?
□ Are partition reassignments or rebalances happening frequently? (Check consumer group lag)
The Real Cost of Exactly-Once (And When to Skip It)
Exactly-once isn't free. Understanding the performance implications helps you make deliberate trade-offs rather than cargo-culting EOS everywhere.
Latency overhead: Transactional Kafka producers add approximately 20-50ms of latency per transaction due to the commit_transaction() round trip. For streaming jobs with 30-second checkpoint intervals, this is negligible. For jobs trying to achieve sub-second end-to-end latency, it matters.
Throughput overhead: The transactional protocol adds writes to the __transaction_state topic and requires acks=all. Benchmarks typically show 10-30% throughput reduction compared to at-least-once producers, depending on batch size and message size. Larger batches amortize the per-transaction overhead.
Checkpoint state overhead: For jobs with large state (windowed aggregations over long windows, large hash maps in keyed state), checkpoints can become expensive. With OnCheckpointRollingPolicy on the FileSink, you're rolling output files every 30 seconds regardless of their size, which can create small file proliferation on S3 (addressed by pipeline compaction and merge strategies).
When to use at-least-once instead:
- Idempotent consumers: If your downstream consumer is idempotent by nature (e.g., writing sensor readings to a time-series database that overwrites on duplicate timestamps), at-least-once with idempotent writes gives you the same effective semantics with lower overhead.
- Analytical workloads with deduplication: If your pipeline feeds a data warehouse query layer that runs daily aggregations, a small duplicate rate is often acceptable — or you can add a
SELECT DISTINCT / QUALIFY ROW_NUMBER() = 1 layer at query time.
- High-throughput, low-value events: Clickstream or telemetry data where individual record loss or duplication is statistically irrelevant can often use at-least-once safely.
Key insight: The question to ask isn't "should I implement exactly-once?" but "what is the cost to my business of one duplicate or one missed record per million events?" For payment processing: very high cost, pay for exactly-once. For ad click tracking: low cost, at-least-once is fine. For IoT sensor aggregations: evaluate per use case. The guarantees should match the business requirements, not the other way around.
The designing idempotent data pipelines article explores the broader philosophy of building for idempotency first, which often eliminates the need for distributed transaction machinery altogether.
Summary & Next Steps
Let's consolidate what you've learned:
Kafka layer: Exactly-once between Kafka topics requires idempotent producers (PID + sequence numbers) for single-session deduplication, combined with the transactional API (stable transactional.id, send_offsets_to_transaction) for durability across restarts. Consumers must set isolation.level=read_committed to avoid reading aborted messages.
Flink layer: Flink achieves end-to-end EOS by integrating Kafka transaction commits with its two-phase commit checkpointing. Pre-commit happens at checkpoint barrier, commit happens after checkpoint completion. This requires EXACTLY_ONCE checkpointing mode, a single concurrent checkpoint, and a transaction.timeout.ms larger than your maximum checkpoint duration.
Sink layer: Each sink type requires a different strategy. PostgreSQL uses upsert with conflict handling. S3 uses the staging-and-commit pattern via OnCheckpointRollingPolicy. Elasticsearch uses deterministic document IDs. HTTP APIs use idempotency keys when available, application-level deduplication when not.
Multi-sink coordination: When writing to multiple heterogeneous sinks, implement the Saga pattern with compensating transactions ordered from most to least recoverable, rather than attempting distributed two-phase commit across heterogeneous systems.
Monitoring: Exactly-once degrades silently. Build duplicate rate monitoring into your sink schemas and watch Kafka transaction abort metrics proactively.
Next steps to deepen your expertise:
