Producer Consumer Transaction #

In event-driven microservices architecture, the processing pattern we most often encounter is the Read-Process-Write pattern. In this pattern, an application consumes messages from an input topic, performs some business logic transformations or data calculations, then sends the results to another output topic. To guarantee this system’s reliability comprehensively, we must ensure that reading the source messages, writing new messages, and updating the read position (offset commit) all execute atomically as one single transaction unit. This is where the transaction integration between producers and consumers in Apache Kafka plays a vital role. This article will deeply dissect how to design safe Read-Process-Write transactions, why regular offset commit methods don’t apply here, how to stream offsets into producer transactions, and production-ready Java implementation code examples.

The Read-Process-Write Pattern Challenge and Why Regular Commits Fail #

In non-transactional applications, the Read-Process-Write processing cycle operates by separating consumer and producer transactions:

  1. The consumer reads a message with offset 300 from the input-orders topic.
  2. The application processes that message.
  3. The producer sends the new processed message to the output-invoices topic.
  4. The consumer sends an offset 300 commit to the broker separately.

Where’s the failure point? If step 3 succeeds (the invoice message was sent to the broker), but the server crashes in step 4 before offset 300 is committed. When the application recovers, the consumer re-reads the offset 300 message from the input-orders topic, processes it again, and the producer sends a duplicate invoice to the output-invoices topic. This damages the exactly-once semantic.

To overcome this partial failure, we must not use standard offset commit methods (like consumer.commitSync() or auto-commit). Instead, consumer offsets must be committed inside the producer transaction. This way, the coordinator broker treats writing offsets to the internal __consumer_offsets topic the same as writing regular messages to the output topic. If the transaction succeeds, the new messages and new offsets are committed simultaneously. If the transaction is aborted, the new messages aren’t visible to downstream consumers and the input consumer offsets don’t move, letting our application safely reprocess those input messages from the original position.

flowchart TD
    subgraph Input["1. Read Phase"]
        Cons["Kafka Consumer"] -->|"Fetch data (Offset 100)"| InTopic["Input Topic: orders"]
    end
    
    subgraph Process["2. Processing Phase"]
        App["Application Thread (Data Transformation)"]
    end
    
    subgraph Output["3. Atomic Write Phase"]
        Prod["Kafka Producer (Transactional)"]
        OutTopic["Output Topic: invoices"]
        OffsetTopic["Internal Topic: __consumer_offsets"]
        
        Prod -->|"A. Send New Invoice"| OutTopic
        Prod -->|"B. Send Offset 100 Commit"| OffsetTopic
    end
    
    InTopic --> Cons
    Cons --> App
    App --> Prod
    
    style InTopic stroke:#333,stroke-width:2px
    style OutTopic stroke:#2e7d32,stroke-width:2px
    style OffsetTopic stroke:#0288d1,stroke-width:2px

Streaming Offsets into the Producer Transaction #

The main key to Read-Process-Write transaction atomicity lies in the producer API function:

  • producer.sendOffsetsToTransaction(Map<TopicPartition, OffsetAndMetadata> offsets, ConsumerGroupMetadata groupMetadata)

This function tells the Transaction Coordinator to register consumer offsets into the active running transaction.

Why Does It Need ConsumerGroupMetadata? #

Since Apache Kafka 2.5, the ConsumerGroupMetadata parameter (obtained via consumer.groupMetadata()) must be included in this API call.

  • Purpose: This parameter carries important information like the consumer’s Member ID and the consumer group’s Generation ID value.
  • Consumer-Side Zombie Fencing: This metadata information is used by the coordinator broker to verify that the consumer instance requesting the offset commit is a legitimate group member. If a rebalance happened and that consumer was removed from the group (because of a long GC pause) and replaced by a new consumer, the coordinator detects the Generation ID difference and rejects that offset commit by throwing FencedInstanceIdException or CommitFailedException. This prevents zombie consumers from disrupting group coordination.

Fencing Protocol Mechanism Details #

When sendOffsetsToTransaction executes, the producer client library sends a TxnOffsetCommitRequest request to the Group Coordinator broker (the broker managing our consumer group, not the Transaction Coordinator). This request wraps the offset data, group ID, Member ID, and Generation ID.

The Group Coordinator validates those parameters:

  1. If the Generation ID sent by the producer on behalf of the consumer is smaller than the consumer group’s current active Generation ID on the broker, the coordinator realizes a rebalance has completed and this consumer is now stale.
  2. The coordinator immediately rejects that offset commit and returns the FencedState error.
  3. The producer client catches this error and throws a transactional exception to abort the entire transaction, ensuring no output data gets committed while the input offset failed to commit.

Scaling Scenarios in Kafka Streams #

For developers using the Kafka Streams framework, this distributed Read-Process-Write coordination is dramatically simplified. Kafka Streams uses the processing.guarantee="exactly_once_v2" parameter.

Behind the scenes, Kafka Streams divides the processing topology into small units called Tasks. Each Task is assigned to process specific input partitions. To guarantee EOS, Kafka Streams allocates a separate transactional producer for each Stream Thread, with transactional.id formatted structurally: <application.id>-<task.id>.

This static per-Task unique format is crucial: if a task crashes and is rescheduled on another thread or server, the new task instance initializes the producer with the same transactional.id. This initialization automatically fences old task instances that might still be running (zombies), thanks to the epoch fencing mechanism in the Transaction Coordinator.

read_committed Isolation Level Behavior on Downstream Consumers #

So transactionally processed data can be safely read by downstream systems, the downstream consumers reading from output-invoices must be configured with the isolation.level=read_committed property.

If downstream consumers use the default (read_uncommitted), they immediately read new messages in the output topic as soon as the producer writes them, without waiting for the transaction to be committed. This violates the atomicity guarantee because if the producer transaction above is eventually aborted, the downstream consumers have already processed wrong dirty data (dirty reads).

Offset Reading Comparison Visualization #

Let’s look at the following partition log structure to understand the reading boundaries between the two isolation levels:

PARTITION LOG STRUCTURE:
Offset:  │ 100   │ 101   │ 102   │ 103   │ 104   │ 105   │ 106
Message: │ Msg A │ Msg B │ Msg C │ Msg D │ Msg E │ Msg F │ COMMIT marker
Status:  │ Normal│ Tx_01 │ Tx_02 │ Tx_01 │ Tx_02 │ Normal│ for Tx_01
         └───────┴───────┴───────┴───────┴───────┴───────┴──────────────
                 ▲                                       ▲
                 │                                       │
                LSO                                     HW (High Watermark)

In the scenario above:

  • Tx_01 is a transaction successfully committed at offset 106.
  • Tx_02 (Msg C at offset 102 and Msg E at offset 104) is an ongoing transaction that hasn’t finished.
  • High Watermark (HW) is the last offset successfully written to all replicas (offset 106).
  • Last Stable Offset (LSO) is at offset 101 because Msg B (Tx_01) and Msg C (Tx_02) are stacked, and Tx_02 is still actively running. LSO can’t move past offset 102.

How do consumers read this data?

  1. read_uncommitted consumers: Read all messages from offset 100 to 106 (including Tx_02’s ongoing transaction data).
  2. read_committed consumers: Only read Msg A at offset 100. The consumer is held back from reading Msg B (offset 101) or Msg F (offset 105) because the log is blocked by the LSO at offset 102 waiting for Tx_02’s status clarity. After Tx_02 receives its commit/abort marker, the LSO jumps forward and the consumer can immediately continue reading.

The Last Stable Offset (LSO) Concept and LSO Blockage #

On the broker side, for each topic partition, Kafka maintains a read pointer boundary called the Last Stable Offset (LSO).

  • Definition: LSO is the smallest offset of the transaction currently still ongoing/active.
  • Consumer Behavior: Consumers with read_committed are only allowed to read messages up to the LSO boundary. If there’s a long-running uncommitted transaction on a partition, that partition’s LSO is held at the position where that transaction started. Even if hundreds of new non-transactional messages are written after that, read_committed consumers are held (blocked/paused) at the LSO boundary and can’t read those new messages until the ongoing transaction is committed or aborted. This phenomenon is called LSO Blockage. Therefore, we must keep every transaction duration as short as possible.

Java Implementation Code: Read-Process-Write Implementation #

Here’s a complete implementation example of the Read-Process-Write transaction processing pattern using the Java SDK. This code safely handles the transaction lifecycle, streams offsets into producer transactions, includes consumer group metadata, and does rollback if processing failures occur.

import org.apache.kafka.clients.consumer.*;
import org.apache.kafka.clients.producer.*;
import org.apache.kafka.common.TopicPartition;
import org.apache.kafka.common.errors.ProducerFencedException;
import org.apache.kafka.common.serialization.StringDeserializer;
import org.apache.kafka.common.serialization.StringSerializer;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.time.Duration;
import java.util.Collections;
import java.util.HashMap;
import java.util.Map;
import java.util.Properties;

public class TransactionalStreamProcessor {
    private static final Logger log = LoggerFactory.getLogger(TransactionalStreamProcessor.class);
    private static final String INPUT_TOPIC = "input-orders";
    private static final String OUTPUT_TOPIC = "output-invoices";
    private static final String TRANSACTIONAL_ID = "stream-processor-tx-01";
    private static final String GROUP_ID = "orders-processor-group";

    public static void main(String[] args) {
        // 1. Input Consumer Configuration
        Properties consumerProps = new Properties();
        consumerProps.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
        consumerProps.put(ConsumerConfig.GROUP_ID_CONFIG, GROUP_ID);
        consumerProps.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        consumerProps.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        
        // Must disable auto-commit because we're streaming offsets into the producer transaction
        consumerProps.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "false");
        consumerProps.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
        // Use read_committed if this consumer also reads from other transactional topics
        consumerProps.put(ConsumerConfig.ISOLATION_LEVEL_CONFIG, "read_committed");

        // 2. Transactional Producer Configuration
        Properties producerProps = new Properties();
        producerProps.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
        producerProps.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
        producerProps.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
        
        // transactional.id must be unique per active producer instance
        producerProps.put(ProducerConfig.TRANSACTIONAL_ID_CONFIG, TRANSACTIONAL_ID);
        producerProps.put(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG, "true");

        KafkaConsumer<String, String> consumer = new KafkaConsumer<>(consumerProps);
        KafkaProducer<String, String> producer = new KafkaProducer<>(producerProps);

        consumer.subscribe(Collections.singletonList(INPUT_TOPIC));

        log.info("Initializing producer transactions...");
        producer.initTransactions();

        try {
            while (true) {
                // Fetch data from the input topic
                ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(200));
                
                if (!records.isEmpty()) {
                    log.info("Received {} records to process.", records.count());
                    
                    try {
                        // ✓ STEP A: Start a new transaction on the producer side
                        producer.beginTransaction();

                        Map<TopicPartition, OffsetAndMetadata> offsetsToCommit = new HashMap<>();

                        for (ConsumerRecord<String, String> record : records) {
                            // ✓ STEP B: Process the data (Business Logic)
                            String outputValue = transformOrderToInvoice(record.value());
                            
                            // ✓ STEP C: Send the processing result to the output topic
                            ProducerRecord<String, String> outRecord = new ProducerRecord<>(
                                    OUTPUT_TOPIC, record.key(), outputValue
                            );
                            producer.send(outRecord);

                            // Record the next offset position (current offset + 1) for each partition
                            offsetsToCommit.put(
                                    new TopicPartition(record.topic(), record.partition()),
                                    new OffsetAndMetadata(record.offset() + 1)
                            );
                        }

                        // ✓ STEP D: Stream the consumer offset commit into the producer transaction.
                        // Include the consumer group metadata to secure the process from zombie consumers.
                        producer.sendOffsetsToTransaction(offsetsToCommit, consumer.groupMetadata());

                        // ✓ STEP E: Commit the transaction atomically (New messages + Offset commits stored together)
                        producer.commitTransaction();
                        log.info("✓ Transaction successfully committed atomically.");

                    } catch (ProducerFencedException e) {
                        // Split-brain happened: This producer was detected as a zombie
                        log.error("✗ Fatal: Producer detected as a zombie. Stopping the process.", e);
                        break;
                    } catch (Exception e) {
                        // A business logic error or ordinary network error happened
                        log.error("✗ Failed to process the batch. Rolling back the transaction...", e);
                        // Cancel message writes and cancel consumer offset updates
                        producer.abortTransaction();
                    }
                }
            }
        } finally {
            log.info("Cleanly closing the consumer and producer connections.");
            consumer.close();
            producer.close();
        }
    }

    private static String transformOrderToInvoice(String orderJson) throws Exception {
        // Simulate business data transformation logic
        if (orderJson.contains("ERROR_TRIGGER")) {
            throw new Exception("Simulated data processing error.");
        }
        return orderJson.replace("ORDER", "INVOICE");
    }
}

When to Use the Read-Process-Write Transaction Pattern? #

Use the following decision guide when determining your system’s transaction architecture:

STILL use Transactional Read-Process-Write if:
  ✓ The data processing pipeline is entirely inside Kafka (Kafka-to-Kafka).
  ✓ Data is calculated in chains across multiple topics and partitions (e.g., e-commerce aggregation).
  ✓ You use the Kafka Streams API and want to guarantee stateful processing (like windowing joins) free from duplication.

DON'T use this pattern (Use manual database deduplication) if:
  ✗ Data processing logic involves writing directly to external databases (SQL/NoSQL).
  ✗ The application makes HTTP REST API calls to third parties (like payment gateways) mid-process.
  ✗ You want to avoid the coordination overhead latency of the Kafka cluster's Two-Phase Commit.

Summary #

  • Read-Process-Write (RPW) — The distributed stream processing pattern where reading from an input topic, processing, and sending results to an output topic are managed in one single transaction.
  • Offsets in Transactions — Input consumer offset commits must be sent through the producer function sendOffsetsToTransaction() so they commit atomically together with the new output messages.
  • ConsumerGroupMetadata — The mandatory parameter in transaction commits securing the system from zombie consumers through Generation ID matching at the Group Coordinator.
  • LSO (Last Stable Offset) — The lowest offset boundary of the oldest still-active transaction on a topic partition, acting as the read boundary for downstream consumers.
  • LSO Blockage — The negative impact where read_committed consumers are held back from reading new messages because of another transaction running too long.
  • Zero Data Loss & Zero Duplication — The main advantage of the transactional RPW pattern ensuring no input messages are missed and no duplicate output messages are emitted to downstream topics.

← Previous: Transaction API Next: EOS Limitation →

About | Author | Content Scope | Editorial Policy | Privacy Policy | Disclaimer | Contact