Exactly-Once Processing #

When we design real-time data processing systems for critical use cases — like financial transaction processing, logistics inventory calculations, or usage-based billing systems — tolerance for data errors is zero. Data loss is unacceptable, and data duplication from network failures or cluster rebalances is also very dangerous. To automatically solve this challenge, Apache Kafka Streams provides the Exactly-Once Processing (EOS) guarantee. This feature ensures every input record is processed exactly once, even if server failures happen mid-way. Through this article, we’ll dissect the internal EOS architecture in Kafka Streams, how atomic transactions integrate with Two-Phase Commit (2PC), EOS v1 vs EOS v2 differences, safe Java code configuration, performance implications, EOS limitations, and a production readiness checklist.

How Exactly-Once Works in Kafka Streams #

The Exactly-Once Processing guarantee in Kafka Streams is achieved through very tight integration between three basic Apache Kafka components: the Idempotent Producer, the Transaction Coordinator, and the Consumer Isolation Level.

Inside Kafka Streams, the data processing workflow always follows the Read-Process-Write pattern:

  1. Read: Internal consumers read records from input topics.
  2. Process: Business logic executes those records, updates local state stores (RocksDB), and prepares to back up state to changelogs.
  3. Write: Internal producers send transformed result records to output topics and write state changes to changelog topics.

The Two-Phase Commit (2PC) Protocol in Kafka #

To unite the three steps above into one All-or-Nothing atomic transaction, Kafka Streams uses a variation of the classic distributed Two-Phase Commit (2PC) protocol:

  • Phase 1: Prepare Commit When the commit interval is exceeded (set by commit.interval.ms), the Kafka Streams thread sends a transaction closing request to the Transaction Coordinator. The Coordinator writes a special record labeled PrepareCommit into the internal transaction log topic __transaction_state. At this stage, the transaction status is permanently locked on the broker. If the coordinator crashes after this point, the active replacement coordinator automatically continues the transaction commit.
  • Phase 2: Commit (Completion / Commit Markers) After the PrepareCommit status is safely written to the log, the Transaction Coordinator sends WriteTxnMarkerRequest instructions to all leader brokers of the involved output topic partitions and changelog topics. Those brokers write special control records called Commit Markers (or Abort Markers if aborted) into physical disk partition logs. After all markers are sent, the coordinator marks the transaction complete by writing the CompleteCommit status to the transaction status topic.
sequenceDiagram
    autonumber
    participant Cons as KStreams Consumer (Read)
    participant TC as Transaction Coordinator
    participant Out as Output Topic & Changelog (Write)
    participant Offset as __consumer_offsets Topic (Commit)

    Note over Cons, TC: "Transaction Cycle Starts"
    Cons->>TC: Start Transaction (beginTransaction)
    Cons->>Out: Send processed data (Produce)
    Cons->>Out: Send RocksDB update backup (Changelog)
    
    Note over Cons, Offset: "Locking Consumption Offsets"
    Cons->>TC: Register consumer offsets (sendOffsetsToTransaction)
    TC->>Offset: Write temporary offset status (Uncommitted)

    Note over Cons, TC: "Atomic Commit Phase (2-Phase Commit)"
    Cons->>TC: Commit Transaction (commitTransaction)
    TC->>Out: Write Commit Marker to the Log
    TC->>Offset: Change Offset status to Committed
    TC-->>Cons: Transaction Finished & Successful!

EOS Evolution: EOS v1 (Alpha) vs EOS v2 (Exactly-Once Beta/Production) #

Apache Kafka first introduced the EOS guarantee in version 0.11 (known as EOS v1), and dramatically refined it in version 2.5 through the introduction of EOS v2 (enabled via the exactly_once_v2 configuration).

1. EOS v1 (Using transactional.id per Task) #

In EOS v1, Kafka Streams creates one separate transactional producer instance for every StreamTask running in the application.

  • High Overhead: If our application processes 50 partitions, there are 50 active transactional producer instances. Each producer must independently interact with the Transaction Coordinator, triggering abundant TCP socket connection creation and burdening JVM memory.
  • Rebalance Latency: Every time a rebalance happens, the closing coordination of old transactions for dozens of tasks takes a very long time, slowing cluster recovery processes.

2. EOS v2 (Using transactional.id per StreamThread) #

EOS v2 is designed to remove the scalability obstacles above by mapping transactions at the StreamThread level, not the StreamTask level anymore.

  • One Producer per Thread: If our application instance has num.stream.threads=2, only 2 transactional producer instances are created, no matter how many hundreds of tasks those threads manage.
  • Extraordinary Scalability: Reduces TCP connection load to brokers by more than 90%, lowers broker CPU usage, and massively speeds up cluster rebalancing processes.
  • Since Kafka 3.0, EOS v1 has been marked deprecated. We must use EOS v2 for all new applications in production.

EOS Guarantee Limitations in Kafka Streams #

It’s important for us to be aware of the theoretical boundaries where this Exactly-Once guarantee applies. EOS in Kafka isn’t a magic solution solving all distributed system consistency problems:

  • Only Applies to Kafka-to-Kafka Flows: The Exactly-Once guarantee is only fully guaranteed if data is read from Kafka topics and written back to Kafka topics. If our Kafka Streams topology writes to external systems (like making HTTP POST calls to external REST APIs or INSERT queries to PostgreSQL) mid-processing, Kafka transactions can’t roll back that external database if the transaction is aborted on the broker.
  • External System Mitigation: For external system interactions, we must implement the Idempotent Consumer pattern (e.g., using unique key constraints in databases) or use the Transactional Outbox Pattern combined with Kafka Connect.

Implementation Code: Anti-Pattern vs the Managed Exactly-Once Solution #

Let’s compare dangerous approaches trying to handle duplication manually with the official EOS implementation in Kafka Streams.

Use Case #

We process payment transaction balance data (account-transactions). We want to ensure running balance additions for every account are processed exactly once without duplication from cluster rebalances.

Anti-Pattern: Managing Manual Duplicate Detection with Redis Locks #

Trying to create an external database (like Redis) as a storage place for transactional locking keys (distributed locks) to manually filter duplicate records is a performance-destroying anti-pattern.

// ANTI-PATTERN: Building a manual idempotency mechanism using Redis Distributed Locks.
// ✗ Very slow (adds network latency overhead), prone to deadlocks, and vulnerable to lock leaks.
public class VolatileIdempotencyProcessor {
    private static final JedisPool jedisPool = new JedisPool("localhost", 6379);

    public static void build(StreamsBuilder builder) {
        builder.<String, String>stream("account-transactions")
            .filter((userId, transactionJson) -> {
                String txId = extractTxId(transactionJson);
                try (Jedis jedis = jedisPool.getResource()) {
                    // ✗ VERY BAD: Blocking write locks to Redis per event!
                    Long isNew = jedis.setnx("lock:" + txId, "locked");
                    if (isNew == 1) {
                        // Lock set for 10 safety minutes
                        jedis.expire("lock:" + txId, 600);
                        return true; // Pass through for processing
                    }
                    return false; // Duplicate detected at the Redis level, discard the event
                }
            })
            .to("processed-transactions");
    }

    private static String extractTxId(String json) { return "tx_123"; }
}

Practical Solution: Enabling EOS v2 in an Integrated Way #

Below is the correct, officially recommended way. We enable the Exactly-Once guarantee by setting the processing.guarantee parameter at the configuration level. Kafka Streams coordinates atomic RocksDB changelog transactions and offset commits transparently in local memory without needing additional external infrastructure.

// CORRECT: Enabling the Exactly-Once processing guarantee (EOS v2) in Kafka Streams.
// ✓ Safe from data loss, automatically scaled, microsecond latency.
public class ResilientExactlyOnceApp {
    
    public static Properties createConfiguration() {
        Properties config = new Properties();
        
        config.put(StreamsConfig.APPLICATION_ID_CONFIG, "finance-accounting-service");
        config.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");

        // =====================================================================
        // IMPORTANT EXACTLY-ONCE CONFIGURATION
        // =====================================================================
        // ✓ Enabling the Exactly-Once v2 guarantee (EOS v2) globally
        // Automatically enables the acks=all parameter on internal producers
        // and forces the read_committed isolation level on internal consumers
        config.put(StreamsConfig.PROCESSING_GUARANTEE_CONFIG, StreamsConfig.EXACTLY_ONCE_V2);

        // ✓ Commit Interval Synergy
        // In EOS, the default commit interval drops from 30 seconds to 100 milliseconds
        // to ensure transactions are immediately committed and minimize downstream consumer lag
        config.put(StreamsConfig.COMMIT_INTERVAL_MS_CONFIG, 100);

        // Setting the internal transaction timeout limit to 1 minute for LSO Blockage mitigation
        config.put(StreamsConfig.producerPrefix(ProducerConfig.TRANSACTION_TIMEOUT_CONFIG), 60000);

        return config;
    }

    public static void main(String[] args) {
        Properties config = createConfiguration();
        StreamsBuilder builder = new StreamsBuilder();
        
        // Reading the financial transaction stream
        KStream<String, String> transactions = builder.stream(
            "account-transactions",
            Consumed.with(Serdes.String(), Serdes.String())
        );

        // Doing stateful running balance accumulation
        // RocksDB changelogs and output data deliveries are guaranteed atomic
        transactions
            .groupByKey(Grouped.with(Serdes.String(), Serdes.String()))
            .aggregate(
                () -> 0.0,
                (userId, txJson, currentBalance) -> currentBalance + parseAmount(txJson),
                Materialized.<String, Double, KeyValueStore<Bytes, byte[]>>as("account-balance-store")
                    .withKeySerde(Serdes.String())
                    .withValueSerde(Serdes.Double())
            )
            .toStream()
            .to("account-balance-alerts", Produced.with(Serdes.String(), Serdes.Double()));

        Topology topology = builder.build();
        KafkaStreams streams = new KafkaStreams(topology, config);
        
        streams.start();
    }

    private static double parseAmount(String json) { return 150000.0; }
}

Performance Implications and the Danger of LSO Blockage #

Although EOS v2 provides extraordinary convenience in maintaining data consistency, we must anticipate two main performance consequences when enabling it in production:

1. Maximum Throughput Decrease #

  • Marker Overhead: Every committed transaction writes one additional small record called a Commit/Abort Marker to broker partition physical logs.
  • Impact: Enabling EOS v2 generally lowers maximum write throughput by 10% - 15% because of the extra load for internal Two-Phase Commit (2PC).

2. The Danger of LSO Blockage (Hanging Last Stable Offsets) #

Downstream consumers reading Kafka Streams processed results with read_committed isolation can’t read any records past the Last Stable Offset (LSO) boundary. LSO is the offset of the oldest active uncommitted transaction record.

  • Problem Scenario: If one Kafka Streams instance experiences freezing (long GC Pauses or hangs) mid-active-transaction, that transaction never calls commitTransaction() or abortTransaction() before the timeout passes.
  • Impact: The LSO boundary on broker partitions is held at old positions. As a result, downstream (read_committed) consumers stop reading (stuck) and experience severe lag spikes, even though other producer applications keep sending thousands of new messages to the same partitions. Downstream consumers only flow smoothly again after that hanging transaction is automatically aborted once the transaction.timeout.ms duration passes.

Exactly-Once (EOS v2) Production Readiness Checklist #

Before releasing Kafka Streams applications with EOS v2 to production, we must verify that broker cluster and client parameters are safely configured:

1. Broker-Level Configuration #

  • transaction.state.log.replication.factor=3: Ensuring internal cluster transaction log replication is safe from single broker deaths.
  • transaction.state.log.min.isr=2: Ensuring new transaction statuses are only successfully written if the minimum ISR count is two brokers.
  • transaction.id.expiration.ms=604800000 (7 days): The transaction ID storage time limit before automatically being deleted from brokers.

2. Client-Level Configuration (KStreams Client Configuration) #

  • processing.guarantee=exactly_once_v2: The mandatory parameter for enabling compressed EOS v2.
  • commit.interval.ms=100 (or maximum 500ms): Guaranteeing short transaction status delivery latency.
  • transaction.timeout.ms=60000 (1 minute): The LSO blockage prevention time limit so downstream systems don’t hang too long during crashes.

Summary #

  • Exactly-Once Processing (EOS) — The processing guarantee where every input event is processed exactly once without data loss or duplication even during failures.
  • Read-Process-Write Atomicity — The protocol binding consumer offsets, state store changelog writes, and data output into one distributed atomic transaction.
  • processing.guarantee=“exactly_once_v2” — The industry standard configuration since Kafka 2.5 for enabling EOS v2 which uses one producer per StreamThread.
  • read_committed — The downstream consumer isolation level ensuring consumers discard aborted transaction messages and only read successful (committed) messages.
  • Last Stable Offset (LSO) — The latest offset boundary on brokers indicating the oldest unclosed active transaction message, acting as a safety filter for consumers.
  • LSO Blockage — The downstream consumption congestion phenomenon from hanging transactions on brokers holding the LSO boundary from moving past normal limits.
← Previous: Join Stream and Table
About | Author | Content Scope | Editorial Policy | Privacy Policy | Disclaimer | Contact