5.3 Exactly-Once Semantics (Producer Side)

Module Overview

Goal: Understand transactional producers for exactly-once delivery across partitions and producer sessions.

What You'll Learn:

• What transactions provide beyond idempotence
• Transactional ID and its role
• Transaction lifecycle and API
• Atomic multi-partition writes
• Exactly-once across producer restarts
• Transactional producer configuration
• Read committed vs read uncommitted isolation
• Performance implications

Prerequisites:

• Module 5.1 (Delivery Semantics)
• Module 5.2 (Idempotent Producer)

1. Beyond Idempotence: Why Transactions?

Idempotent Producer Limitations

Idempotent producers prevent duplicates but have limitations:

❌ Session-scoped — Lost on restart
❌ Per-partition — No cross-partition atomicity
❌ Single producer instance — Multiple instances can duplicate

What Transactions Add

✅ Cross-partition atomicity — All-or-nothing writes
✅ Survive producer restarts — Exactly-once across sessions
✅ Application-level identity — Multiple producers with same identity
✅ Read isolation — Consumers see committed data only

Use Cases for Transactions

1. Read-Process-Write Pattern

   Read from Kafka → Process → Write to Kafka
   (Exactly-once stream processing)
   

2. Atomic Multi-Partition Writes

   Write to topic A, B, C atomically
   (All succeed or all fail)
   

3. Exactly-Once Across Restarts

   Producer crashes → Restarts → No duplicates
   

4. Transactional Outbox Pattern

   Database update + Kafka write atomically
   

2. Transactional Producer Fundamentals

Core Concept

A transactional producer groups multiple sends into an atomic transaction:

BEGIN TRANSACTION
  ├─ send(record1) → topic-A, partition-0
  ├─ send(record2) → topic-A, partition-1
  ├─ send(record3) → topic-B, partition-0
  └─ send(record4) → topic-C, partition-2
COMMIT TRANSACTION

Result: All 4 records written atomically (or none)

Transaction Guarantees

1. Atomicity: All records in transaction written, or none
2. Durability: Committed transactions survive broker failures
3. Isolation: Consumers see committed transactions only (with read_committed)
4. Exactly-once: No duplicates, even across restarts

3. Transactional ID

What is Transactional ID?

A transactional ID (transactional.id) is a unique, stable identifier for a producer application.

props.put("transactional.id", "order-processor-1");

Properties:

• User-defined (you choose it)
• Stable across restarts (same ID every time)
• Unique per producer instance (different instances → different IDs)

Transactional ID vs Producer ID (PID)

Relationship:

Transactional ID → Mapped to → Producer ID (PID) + Epoch

Producer restarts → New PID + Epoch++ (same transactional ID)

Epoch Number

Epoch: Generation number incremented on each producer restart.

Producer 1 (tx-id-1): PID=100, Epoch=0
Producer 1 restarts:  PID=200, Epoch=1
Producer 1 restarts:  PID=300, Epoch=2

Purpose: Detect zombie producers (old instances).

Zombie Fencing:

Old producer (Epoch=0) tries to write → REJECTED (old epoch)
New producer (Epoch=1) writes       → ACCEPTED (current epoch)

4. Enabling Transactional Producer

Configuration

Properties props = new Properties();
props.put("bootstrap.servers", "localhost:9092");
props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");

// Enable transactions
props.put("transactional.id", "my-app-producer-1"); // Unique ID
props.put("enable.idempotence", "true");             // Required (auto-set)

// These are set automatically:
// props.put("acks", "all");
// props.put("retries", Integer.MAX_VALUE);

KafkaProducer<String, String> producer = new KafkaProducer<>(props);

Key Points:

• transactional.id must be set
• Idempotence automatically enabled
• acks=all and retries=MAX_VALUE automatically set

Initializing Transactions

// Initialize producer (registers transactional ID with broker)
producer.initTransactions();

What happens:

1. Producer requests PID from broker with transactional ID
2. Broker checks if transactional ID exists:
   • If new → Assign PID, Epoch=0
   • If exists → Increment Epoch, assign new PID
3. Broker fences any old producers with same transactional ID

Call once at startup, before sending any records.

5. Transaction Lifecycle

Basic Transaction Flow

producer.initTransactions(); // Once at startup

try {
    // BEGIN TRANSACTION
    producer.beginTransaction();
    
    // Send records
    producer.send(new ProducerRecord<>("topic-A", "key1", "value1"));
    producer.send(new ProducerRecord<>("topic-B", "key2", "value2"));
    producer.send(new ProducerRecord<>("topic-C", "key3", "value3"));
    
    // COMMIT TRANSACTION
    producer.commitTransaction();
    
} catch (Exception e) {
    // ABORT TRANSACTION
    producer.abortTransaction();
    throw e;
}

Transaction States

[UNINITIALIZED]
      │
      ├─ initTransactions()
      ↓
[READY]
      │
      ├─ beginTransaction()
      ↓
[IN_TRANSACTION]
      │
      ├─ send(record) ...
      ├─ send(record) ...
      │
      ├─ commitTransaction() ──→ [READY] (success)
      │
      └─ abortTransaction() ──→ [READY] (rollback)

Complete Example

import org.apache.kafka.clients.producer.*;
import java.util.Properties;

public class TransactionalProducerExample {
    
    public static void main(String[] args) {
        Properties props = new Properties();
        props.put("bootstrap.servers", "localhost:9092");
        props.put("key.serializer", 
            "org.apache.kafka.common.serialization.StringSerializer");
        props.put("value.serializer", 
            "org.apache.kafka.common.serialization.StringSerializer");
        props.put("transactional.id", "order-processor-1");
        
        KafkaProducer<String, String> producer = new KafkaProducer<>(props);
        
        // Initialize (once)
        producer.initTransactions();
        
        // Process orders
        for (int i = 0; i < 100; i++) {
            try {
                producer.beginTransaction();
                
                // Send order event
                producer.send(new ProducerRecord<>(
                    "orders", 
                    "order-" + i, 
                    "order-data-" + i
                ));
                
                // Send inventory update
                producer.send(new ProducerRecord<>(
                    "inventory", 
                    "product-" + i, 
                    "decrement-1"
                ));
                
                // Send notification
                producer.send(new ProducerRecord<>(
                    "notifications", 
                    "user-" + i, 
                    "order-confirmation"
                ));
                
                // Commit atomically
                producer.commitTransaction();
                
            } catch (ProducerFencedException | OutOfOrderSequenceException | 
                     AuthorizationException e) {
                // Fatal error — cannot recover
                producer.close();
                throw e;
                
            } catch (KafkaException e) {
                // Retriable error — abort and retry
                producer.abortTransaction();
            }
        }
        
        producer.close();
    }
}

6. Read-Process-Write Pattern

Use Case: Stream Processing

Pattern: Read from Kafka, process, write back to Kafka (exactly-once).

public void processRecords() {
    // Consumer reads from input topic
    ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
    
    try {
        producer.beginTransaction();
        
        for (ConsumerRecord<String, String> record : records) {
            // Process record
            String result = process(record.value());
            
            // Write to output topic
            producer.send(new ProducerRecord<>("output-topic", record.key(), result));
        }
        
        // Commit consumer offsets to transaction
        producer.sendOffsetsToTransaction(
            getOffsets(records), 
            consumer.groupMetadata()
        );
        
        // Commit transaction
        producer.commitTransaction();
        
    } catch (Exception e) {
        producer.abortTransaction();
    }
}

sendOffsetsToTransaction()

Purpose: Include consumer offset commits in the transaction.

producer.sendOffsetsToTransaction(
    offsets,                    // Map<TopicPartition, OffsetAndMetadata>
    consumer.groupMetadata()    // Consumer group info
);

Effect:

• Consumer offsets committed atomically with producer writes
• If transaction aborts → Offsets NOT committed
• Exactly-once: Read → Process → Write

Result: No duplicates, no data loss in stream processing.

Why This Matters

Without Transactions:

1. Read record (offset 100)
2. Process
3. Write output
4. Write succeeds
5. CRASH before committing offset
6. Restart → Read record (offset 100) again
7. Duplicate output!

With Transactions:

1. Read record (offset 100)
2. BEGIN TRANSACTION
3. Write output
4. Commit offset (101) to transaction
5. COMMIT TRANSACTION
6. CRASH
7. Restart → Read from offset 101 (no duplicate)

7. Transaction Coordinator

What is Transaction Coordinator?

Transaction Coordinator: Broker component that manages transactions.

Responsibilities:

• Track transaction state
• Coordinate commit/abort across partitions
• Fence zombie producers
• Manage transactional ID → PID mapping

Transaction Log (__transaction_state)

Kafka stores transaction metadata in internal topic:

__transaction_state (50 partitions by default)

Stores:
  - Transactional ID → PID, Epoch mapping
  - Transaction state (ongoing, committed, aborted)
  - Partitions involved in transaction

Similar to: __consumer_offsets (for consumer groups)

8. Consumer Isolation Levels

Read Committed vs Read Uncommitted

Consumer Configuration:

props.put("isolation.level", "read_committed"); // or "read_uncommitted"

read_uncommitted (Default for non-transactional)

Behavior: Read all messages, including those in ongoing transactions.

Producer writes (transaction ongoing):
  Offset 0: msg-A (uncommitted)
  Offset 1: msg-B (uncommitted)
  Offset 2: msg-C (uncommitted)

Consumer reads: msg-A, msg-B, msg-C (immediately)

Producer aborts transaction.

Result: Consumer read aborted data!

read_committed

Behavior: Read only committed messages.

Producer writes (transaction ongoing):
  Offset 0: msg-A (uncommitted)
  Offset 1: msg-B (uncommitted)
  Offset 2: msg-C (uncommitted)

Consumer reads: (blocked, waiting for commit)

Producer commits transaction.

Consumer reads: msg-A, msg-B, msg-C (after commit)

Guarantees: Consumers never see aborted data.

Latency Impact

read_uncommitted:
  Latency: Low (read immediately)

read_committed:
  Latency: Higher (wait for commit)

Typical overhead: < 100ms (time to commit transaction)

9. Exactly-Once Across Restarts

Scenario: Producer Crashes Mid-Transaction

Time  Producer (tx-id-1, PID=100, Epoch=0)
─────┼──────────────────────────────────────
  1  │ beginTransaction()
  2  │ send(msg-A) → topic-1
  3  │ send(msg-B) → topic-2
  4  │ CRASH (before commit)

  ── Producer restarts ──

  5  │ New instance (tx-id-1, PID=200, Epoch=1)
  6  │ initTransactions()
  7  │   → Broker increments epoch
  8  │   → Broker aborts previous transaction (PID=100, Epoch=0)
  9  │ beginTransaction()
 10  │ send(msg-A) → topic-1 (no duplicate)
 11  │ send(msg-B) → topic-2 (no duplicate)
 12  │ commitTransaction()

Result: No duplicates, even though producer crashed.

How It Works

1. Producer restarts with same transactional.id
2. initTransactions() increments epoch
3. Broker aborts any ongoing transaction from previous epoch
4. Old messages not committed (consumers with read_committed never see them)
5. New producer continues with new epoch (no duplicates)

10. Zombie Fencing

The Zombie Problem

Scenario: Network partition causes multiple producer instances.

Time  Producer A (Epoch=0)     Producer B (Epoch=1)
─────┼─────────────────────────┼─────────────────────
  1  │ Running                 │
  2  │ Network partition       │
  3  │ (Appears dead)          │
  4  │                         │ Starts (same tx-id)
  5  │                         │ initTransactions()
  6  │                         │ Epoch=1 assigned
  7  │ Network recovers        │
  8  │ Tries to write          │
  9  │ ❌ REJECTED (Epoch=0 < 1) │
 10  │                         │ ✅ Writes (Epoch=1)

Result: Old producer (zombie) cannot write.

ProducerFencedException

Thrown when: Producer with old epoch tries to write.

try {
    producer.commitTransaction();
} catch (ProducerFencedException e) {
    // Fatal: Another producer with same transactional ID started
    // Cannot recover — must close
    producer.close();
    throw e;
}

Action: Close producer immediately (cannot continue).

11. Error Handling

Fatal Errors (Cannot Recover)

try {
    producer.commitTransaction();
    
} catch (ProducerFencedException e) {
    // Another producer fenced this one
    producer.close();
    throw e;
    
} catch (OutOfOrderSequenceException e) {
    // Sequence number issue (bug or corruption)
    producer.close();
    throw e;
    
} catch (AuthorizationException e) {
    // Permission denied
    producer.close();
    throw e;
}

Must close producer — Cannot recover.

Retriable Errors

try {
    producer.commitTransaction();
    
} catch (KafkaException e) {
    // Retriable error (network, broker unavailable, etc.)
    producer.abortTransaction();
    // Can retry transaction
}

Can retry — Abort and try again.

12. Performance Implications

Overhead

Transactional producer adds:

• Transaction coordinator RPCs
• Transaction log writes
• Consumer isolation overhead

Benchmark Example

Configuration: acks=all, 100 byte messages, 3 replicas

Non-transactional (idempotent):
  Throughput: 102,000 msg/sec
  Latency (p99): 11ms

Transactional (small transactions, 10 msgs each):
  Throughput: 75,000 msg/sec (-26%)
  Latency (p99): 18ms (+7ms)

Transactional (large transactions, 1000 msgs each):
  Throughput: 95,000 msg/sec (-7%)
  Latency (p99): 50ms (+39ms)

Key Insight: Larger transactions amortize overhead (better throughput, higher latency).

Optimization Tips

1. Batch multiple sends per transaction

   producer.beginTransaction();
   for (int i = 0; i < 1000; i++) {
       producer.send(record);
   }
   producer.commitTransaction();
   

2. Don't over-transaction

   // ❌ BAD: One transaction per message
   for (int i = 0; i < 1000; i++) {
       producer.beginTransaction();
       producer.send(record);
       producer.commitTransaction();
   }
   

3. Balance latency vs throughput

   • Small transactions: Lower latency, lower throughput
   • Large transactions: Higher latency, higher throughput

13. Configuration Summary

Minimum Configuration

Properties props = new Properties();
props.put("bootstrap.servers", "localhost:9092");
props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");

// Transactional configuration
props.put("transactional.id", "unique-app-id-1");

That's it! Everything else is automatic.

Advanced Configuration

// Transaction timeout (broker aborts if transaction open too long)
props.put("transaction.timeout.ms", 60000); // 1 minute (default)

// Delivery timeout
props.put("delivery.timeout.ms", 120000);   // 2 minutes

// Compression (recommended)
props.put("compression.type", "lz4");

14. When to Use Transactional Producer

Use Transactional Producer When:

✅ Atomic multi-partition writes required
✅ Exactly-once across producer restarts needed
✅ Read-process-write pattern (stream processing)
✅ No duplicates acceptable (e.g., financial transactions)
✅ Can tolerate slightly higher latency (< 100ms)

Use Idempotent Producer When:

✅ Single partition writes (or independent partitions)
✅ OK with duplicates on restart (rare)
✅ Lowest latency required
✅ Simple use case (produce events)

Summary

Key Takeaways

✅ Transactional producer provides exactly-once beyond idempotence
✅ transactional.id is stable, user-defined identifier
✅ Epoch fencing prevents zombie producers
✅ Atomic multi-partition writes (all-or-nothing)
✅ Survives producer restarts (no duplicates)
✅ read_committed isolation for consumers
✅ sendOffsetsToTransaction() for read-process-write
✅ Performance overhead: ~10-30% (depends on transaction size)
✅ Fatal errors: Must close producer
✅ Retriable errors: Abort and retry

Transaction API Summary

Configuration Checklist

• transactional.id set (unique per producer instance)
• initTransactions() called once at startup
• beginTransaction() before sending
• commitTransaction() or abortTransaction() after sending
• Error handling for fatal errors (ProducerFencedException, etc.)
• Consumer isolation.level=read_committed (if needed)
• Transaction size optimized (batch multiple sends)

What's Next

• Module 6.1: Error Handling & Failure Scenarios
  Handle producer errors comprehensively

• Module 7.1: Producer Internals
  Understand what happens inside the producer

Previous Module: ← 5.2 - Idempotent Producer
Next Module: 6.1 - Error Handling & Failure Scenarios →