KafkaIQ

10.1.2 – Producer Transactions

Module: M10 – Advanced Topics & Architecture
Section: 10.1 – Transactions & Exactly-Once Semantics
Goal: Master the producer transactional API and implementation patterns

Introduction

This module covers the practical implementation of transactional producers in Kafka. You'll learn how to configure, initialize, and use the transaction API to achieve exactly-once semantics.

Prerequisites: Understanding of transaction protocol (10.1.1)

Configuring a Transactional Producer

Required Configuration

Minimum config for transactions:

# Enable idempotence (required for transactions)
enable.idempotence=true

# Unique transactional ID (must be unique per producer instance)
transactional.id=payment-processor-1

# Bootstrap servers
bootstrap.servers=localhost:9092

# Serializers
key.serializer=org.apache.kafka.common.serialization.StringSerializer
value.serializer=org.apache.kafka.common.serialization.StringSerializer

Key Configuration Parameters

1. transactional.id

Purpose: Uniquely identifies the producer instance, enabling exactly-once guarantees across restarts.

Rules:

  • Must be unique per logical producer instance
  • Must be stable across restarts (same ID for same logical producer)
  • Enables fencing of old producer instances

Example Naming:

transactional.id = "payment-processor-1"      # Single-threaded processor
transactional.id = "order-service-thread-0"   # Multi-threaded (one ID per thread)
transactional.id = "etl-job-partition-5"      # Partitioned batch job

Anti-Pattern:

transactional.id = UUID.randomUUID().toString()  # ❌ Different ID each time

Why Bad?: Each restart gets a new ID, defeating the purpose of transactions (can't fence zombies, can't recover state).

2. transaction.timeout.ms

Purpose: Maximum time a transaction can remain open before the coordinator aborts it.

Default: 60,000 ms (1 minute)

Configuration:

transaction.timeout.ms=300000  # 5 minutes

When to Increase:

  • Slow processing (complex transformations)
  • Large batch sizes
  • External system calls (DB, API)

When to Keep Low:

  • Fast, simple transformations
  • High-throughput pipelines
  • Want quick zombie detection

3. acks

Required: acks=all (transactions implicitly set this)

Why: Transactions require all in-sync replicas to acknowledge writes for durability.

4. max.in.flight.requests.per.connection

Recommendation: max.in.flight.requests.per.connection=5 (default with idempotence)

Why: Enables pipelining while maintaining order guarantees.

Full Example Configuration

Properties props = new Properties();

// Required for transactions
props.put("bootstrap.servers", "broker1:9092,broker2:9092,broker3:9092");
props.put("transactional.id", "payment-processor-1");
props.put("enable.idempotence", "true");

// Serializers
props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");

// Optional: Tuning
props.put("transaction.timeout.ms", "300000");  // 5 minutes
props.put("max.block.ms", "60000");             // Max time to block on send
props.put("retries", Integer.MAX_VALUE);        // Infinite retries (default with idempotence)

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

Initializing Transactions

initTransactions()

Purpose: Initializes the producer for transactions by registering with the transaction coordinator.

What It Does:

  1. Contacts transaction coordinator
  2. Obtains or recovers Producer ID (PID) and epoch
  3. Fences any previous instances with the same transactional.id
  4. Aborts any incomplete transactions from previous sessions

Usage:

producer.initTransactions();  // Must be called once before any transactions

Blocking Behavior: This call blocks until initialization completes (can take several seconds).

Initialization Sequence:

Producer starts
  ↓
Calls initTransactions()
  ↓
Sends InitProducerIdRequest to coordinator
  ↓
Coordinator checks __transaction_state:
  - If transactional.id exists: Increment epoch, fence old producer
  - If new: Assign new PID
  ↓
Coordinator aborts any incomplete transactions from old producer
  ↓
Returns {PID, epoch} to producer
  ↓
Producer ready to start transactions

Error Handling:

try {
    producer.initTransactions();
} catch (TimeoutException e) {
    // Coordinator unreachable or slow
    log.error("Failed to initialize transactions", e);
    // Retry with backoff or fail fast
} catch (KafkaException e) {
    // Unrecoverable error (invalid config, auth failure)
    log.error("Transaction initialization failed", e);
    throw e;
}

Transaction Lifecycle

Step-by-Step Transaction Flow

1. Begin Transaction

producer.beginTransaction();

What It Does:

  • Marks the producer as "in transaction" (local state)
  • No network call to coordinator yet

2. Produce Messages

ProducerRecord<String, String> record1 = new ProducerRecord<>("orders", "order-1", "paid");
ProducerRecord<String, String> record2 = new ProducerRecord<>("inventory", "item-1", "reserved");

producer.send(record1);
producer.send(record2);

What It Does:

  • First send() triggers AddPartitionsToTxnRequest to coordinator
  • Coordinator adds partitions to transaction state
  • Messages sent to brokers with transaction metadata (PID, epoch, sequence)
  • Messages marked as uncommitted (not visible to read_committed consumers)

3. Commit Transaction

producer.commitTransaction();

What It Does:

  • Sends EndTxnRequest(commit=true) to coordinator
  • Coordinator writes COMMIT markers to all partitions
  • Messages become visible to read_committed consumers
  • Transaction completes

Blocking Behavior: Blocks until all markers are written and acknowledged.

4. Abort Transaction

producer.abortTransaction();

What It Does:

  • Sends EndTxnRequest(commit=false) to coordinator
  • Coordinator writes ABORT markers to all partitions
  • Messages become invisible (discarded by read_committed consumers)
  • Transaction rolls back

Complete Transaction Example

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

try {
    producer.beginTransaction();
    
    // Produce messages to multiple topics
    producer.send(new ProducerRecord<>("orders", "order-1", "paid"));
    producer.send(new ProducerRecord<>("payments", "payment-1", "processed"));
    producer.send(new ProducerRecord<>("notifications", "email-1", "sent"));
    
    // Commit transaction (all-or-nothing)
    producer.commitTransaction();
    
    log.info("Transaction committed successfully");
    
} catch (ProducerFencedException | OutOfOrderSequenceException | AuthorizationException e) {
    // Fatal errors - cannot recover
    log.error("Fatal producer error", e);
    producer.close();
    throw e;
    
} catch (KafkaException e) {
    // Transient error - abort and retry
    log.warn("Transaction failed, aborting", e);
    producer.abortTransaction();
    // Retry logic here
    
} finally {
    producer.close();
}

Transactional Producer Patterns

Pattern 1: Simple Transactional Write

Use Case: Write to multiple topics atomically.

public void processOrder(Order order) {
    try {
        producer.beginTransaction();
        
        producer.send(new ProducerRecord<>("orders", order.getId(), order.toJson()));
        producer.send(new ProducerRecord<>("inventory", order.getItemId(), "reserved"));
        producer.send(new ProducerRecord<>("audit-log", order.getId(), "order-created"));
        
        producer.commitTransaction();
        
    } catch (KafkaException e) {
        log.error("Failed to process order", e);
        producer.abortTransaction();
        throw e;
    }
}

Guarantee: All three messages are written atomically, or none are.

Pattern 2: Read-Process-Write (Consumer + Producer)

Use Case: Consume from one topic, transform, produce to another, commit offsets atomically.

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

consumer.subscribe(Arrays.asList("input-topic"));
producer.initTransactions();

while (true) {
    ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
    
    if (!records.isEmpty()) {
        try {
            producer.beginTransaction();
            
            for (ConsumerRecord<String, String> record : records) {
                // Transform
                String transformed = transform(record.value());
                
                // Produce to output topic
                producer.send(new ProducerRecord<>("output-topic", record.key(), transformed));
            }
            
            // Commit offsets as part of transaction
            Map<TopicPartition, OffsetAndMetadata> offsets = new HashMap<>();
            for (TopicPartition partition : records.partitions()) {
                long lastOffset = records.records(partition).get(records.records(partition).size() - 1).offset();
                offsets.put(partition, new OffsetAndMetadata(lastOffset + 1));
            }
            producer.sendOffsetsToTransaction(offsets, consumer.groupMetadata());
            
            // Commit transaction
            producer.commitTransaction();
            
        } catch (ProducerFencedException | OutOfOrderSequenceException | AuthorizationException e) {
            log.error("Fatal error, closing producer", e);
            producer.close();
            break;
            
        } catch (KafkaException e) {
            log.warn("Transaction failed, aborting", e);
            producer.abortTransaction();
        }
    }
}

Guarantee: Processed messages and consumer offsets are committed atomically. If producer crashes, consumer resumes from last committed offset and reprocesses (no data loss, no duplicates).

Pattern 3: Batch Processing with Transactions

Use Case: Process large batches atomically.

public void processBatch(List<Record> batch) {
    try {
        producer.beginTransaction();
        
        for (Record record : batch) {
            producer.send(new ProducerRecord<>("output-topic", record.getKey(), record.getValue()));
        }
        
        producer.commitTransaction();
        log.info("Batch of {} records committed", batch.size());
        
    } catch (KafkaException e) {
        log.error("Batch failed, aborting", e);
        producer.abortTransaction();
        throw e;
    }
}

Tuning:

transaction.timeout.ms=600000  # 10 minutes for large batches

Pattern 4: External System Integration (Database + Kafka)

Use Case: Write to database and Kafka atomically (requires XA transactions or idempotency).

Approach 1: Idempotent Database Writes

public void processPayment(Payment payment) {
    try {
        producer.beginTransaction();
        
        // Write to Kafka
        producer.send(new ProducerRecord<>("payments", payment.getId(), payment.toJson()));
        
        // Write to database (with idempotent UPDATE)
        database.execute("UPDATE accounts SET balance = balance - ? WHERE id = ? AND balance >= ?",
                         payment.getAmount(), payment.getAccountId(), payment.getAmount());
        
        producer.commitTransaction();
        
    } catch (Exception e) {
        producer.abortTransaction();
        throw e;
    }
}

Guarantee: If Kafka commits but DB fails, retry is safe (idempotent UPDATE).

Approach 2: Outbox Pattern

// Write to DB outbox table + commit
database.beginTransaction();
database.insert("INSERT INTO outbox (event_id, payload) VALUES (?, ?)", eventId, payload);
database.commit();

// Separate process reads outbox and sends to Kafka transactionally

Guarantee: Database is source of truth, Kafka is eventually consistent.

Error Handling

Fatal Errors (Cannot Recover)

1. ProducerFencedException

Cause: Another producer with the same transactional.id has fenced this producer (higher epoch).

Action: Close producer immediately, do NOT retry.

catch (ProducerFencedException e) {
    log.error("Producer fenced by newer instance", e);
    producer.close();
    System.exit(1);  // Let orchestration restart
}

2. OutOfOrderSequenceException

Cause: Broker detected out-of-order sequence numbers (data corruption or bug).

Action: Close producer, investigate.

catch (OutOfOrderSequenceException e) {
    log.error("Sequence error, data corruption possible", e);
    producer.close();
    throw e;
}

3. AuthorizationException

Cause: Producer lacks permissions to write to topic or use transactions.

Action: Fix ACLs, restart.

catch (AuthorizationException e) {
    log.error("Authorization failed, check ACLs", e);
    producer.close();
    throw e;
}

Transient Errors (Retry Allowed)

1. KafkaException (generic)

Cause: Network issues, broker unavailable, coordinator failover.

Action: Abort transaction, retry with backoff.

catch (KafkaException e) {
    log.warn("Transient error, aborting and retrying", e);
    producer.abortTransaction();
    
    // Exponential backoff
    Thread.sleep(retryBackoffMs);
    retryBackoffMs = Math.min(retryBackoffMs * 2, maxBackoffMs);
}

2. TimeoutException

Cause: Coordinator or broker didn't respond in time.

Action: Abort transaction, retry.

catch (TimeoutException e) {
    log.warn("Transaction timeout, aborting", e);
    producer.abortTransaction();
}

Error Handling Best Practices

1. Always Abort on Transient Errors

try {
    producer.beginTransaction();
    // ... produce messages ...
    producer.commitTransaction();
} catch (ProducerFencedException | OutOfOrderSequenceException | AuthorizationException e) {
    // Fatal - close and exit
    producer.close();
    throw e;
} catch (KafkaException e) {
    // Transient - abort and retry
    producer.abortTransaction();
    // Retry logic
}

2. Implement Retry with Exponential Backoff

int maxRetries = 3;
int retryCount = 0;
long backoff = 1000;  // 1 second

while (retryCount < maxRetries) {
    try {
        // Transaction logic
        break;  // Success
    } catch (KafkaException e) {
        retryCount++;
        if (retryCount >= maxRetries) throw e;
        Thread.sleep(backoff);
        backoff *= 2;
    }
}

3. Monitor Transaction Failures

Metrics.counter("transactions.aborted").increment();
Metrics.counter("transactions.committed").increment();
Metrics.timer("transaction.duration").record(duration);

Performance Considerations

Transaction Overhead

Latency Impact:

  • Idempotent producer: ~1ms overhead (sequence numbers)
  • Transactional producer: ~5-10ms overhead (coordinator communication, markers)

Throughput Impact:

  • ~10-20% throughput reduction compared to non-transactional producers

Batching to Amortize Overhead

Anti-Pattern: One transaction per message

for (Message msg : messages) {
    producer.beginTransaction();
    producer.send(msg);
    producer.commitTransaction();  // ❌ Expensive (1 transaction = 5-10ms overhead)
}

Best Practice: Batch multiple messages per transaction

producer.beginTransaction();
for (Message msg : messages) {
    producer.send(msg);  // Batch 100-1000 messages
}
producer.commitTransaction();  // ✅ Overhead amortized

Rule of Thumb: 100-1000 messages per transaction for optimal throughput.

Tuning for Throughput

# Increase batch size to reduce per-message overhead
batch.size=32768        # 32 KB (default: 16 KB)
linger.ms=10            # Wait 10ms to fill batch

# Increase transaction timeout for large batches
transaction.timeout.ms=300000  # 5 minutes

# Enable compression to reduce network/disk I/O
compression.type=snappy

Hands-On Lab: Implementing EOS Step-by-Step

This section provides complete, runnable code examples for implementing transactional producers in real applications.

Lab 1: Transactional Producer-Consumer Pipeline

Goal: Build a Java pipeline where a producer sends messages transactionally, and a consumer reads them with EOS guarantees. We'll simulate failures to test atomicity.

Step 1: Setup

Prerequisites:

  • Kafka cluster (local or Confluent)
  • Java 8+ and Maven
  • Dependencies:
    <dependency>
  <groupId>org.apache.kafka</groupId>
  <artifactId>kafka-clients</artifactId>
  <version>3.4.0</version>
</dependency>

Create Topics:

kafka-topics --create --topic orders --partitions 3 --replication-factor 1 --bootstrap-server localhost:9092
kafka-topics --create --topic payments --partitions 3 --replication-factor 1 --bootstrap-server localhost:9092

Step 2: Transactional Producer

Code: Sends two messages atomically ("order created" + "payment processed").

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

public class TransactionalProducer {
    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");
        
        // Transaction configuration
        props.put("acks", "all");
        props.put("enable.idempotence", "true");
        props.put("transactional.id", "eos-demo-1"); // Unique per producer instance
        props.put("max.in.flight.requests.per.connection", "5");

        KafkaProducer<String, String> producer = new KafkaProducer<>(props);
        producer.initTransactions(); // Critical!

        try {
            producer.beginTransaction();
            
            // Send to two topics atomically
            producer.send(new ProducerRecord<>("orders", "order-123", "Order created"));
            producer.send(new ProducerRecord<>("payments", "order-123", "Payment processed"));
            
            producer.commitTransaction(); // Both or none
            System.out.println("Transaction committed successfully");
            
        } catch (ProducerFencedException | OutOfOrderSequenceException | AuthorizationException e) {
            // Fatal errors - close producer
            System.err.println("Fatal error: " + e.getMessage());
            producer.close();
            throw e;
            
        } catch (KafkaException e) {
            // Transient error - abort and retry
            System.err.println("Transaction failed: " + e.getMessage());
            producer.abortTransaction(); // Rollback on failure
            
        } finally {
            producer.close();
        }
    }
}

Key Points:

  • transactional.id must be unique and stable across restarts
  • initTransactions() registers the producer with Kafka's transaction coordinator
  • If the producer crashes before commitTransaction(), Kafka discards the messages

Step 3: Transactional Consumer

Code: Reads from payments and commits offsets transactionally.

import org.apache.kafka.clients.consumer.*;
import org.apache.kafka.clients.producer.*;
import org.apache.kafka.common.TopicPartition;
import java.time.Duration;
import java.util.*;

public class TransactionalConsumer {
    public static void main(String[] args) {
        // Consumer configuration
        Properties consumerProps = new Properties();
        consumerProps.put("bootstrap.servers", "localhost:9092");
        consumerProps.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        consumerProps.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        consumerProps.put("group.id", "eos-consumer-group");
        consumerProps.put("isolation.level", "read_committed"); // Ignore aborted messages
        consumerProps.put("enable.auto.commit", "false"); // Manual offset commits

        KafkaConsumer<String, String> consumer = new KafkaConsumer<>(consumerProps);
        consumer.subscribe(Arrays.asList("payments"));

        // Producer configuration for transactional offset commits
        Properties producerProps = new Properties();
        producerProps.put("bootstrap.servers", "localhost:9092");
        producerProps.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
        producerProps.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
        producerProps.put("transactional.id", "eos-consumer-producer-1");
        producerProps.put("enable.idempotence", "true");

        KafkaProducer<String, String> producer = new KafkaProducer<>(producerProps);
        producer.initTransactions();

        while (true) {
            ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(1000));
            
            if (!records.isEmpty()) {
                try {
                    producer.beginTransaction();
                    
                    for (ConsumerRecord<String, String> record : records) {
                        System.out.printf("Processed: %s\n", record.value());
                        // Simulate external side effect (e.g., DB write)
                    }
                    
                    // Commit offsets + external state atomically
                    Map<TopicPartition, OffsetAndMetadata> offsets = new HashMap<>();
                    for (TopicPartition partition : records.partitions()) {
                        List<ConsumerRecord<String, String>> partitionRecords = records.records(partition);
                        long lastOffset = partitionRecords.get(partitionRecords.size() - 1).offset();
                        offsets.put(partition, new OffsetAndMetadata(lastOffset + 1));
                    }
                    
                    producer.sendOffsetsToTransaction(offsets, consumer.groupMetadata());
                    producer.commitTransaction();
                    
                } catch (ProducerFencedException | OutOfOrderSequenceException | AuthorizationException e) {
                    System.err.println("Fatal error: " + e.getMessage());
                    producer.close();
                    consumer.close();
                    break;
                    
                } catch (KafkaException e) {
                    System.err.println("Processing failed: " + e.getMessage());
                    producer.abortTransaction();
                }
            }
        }
    }
}

Key Points:

  • isolation.level=read_committed skips aborted transactions
  • sendOffsetsToTransaction() binds offset commits to the transaction
  • If transaction aborts, consumer will reprocess the same messages

Step 4: Simulate Failures

Test 1: Crash Producer Mid-Transaction

// Add before commitTransaction():
Thread.sleep(5000);  // Simulate slow processing

// Kill the producer during sleep (Ctrl+C)
// Verify: No messages appear in the topics

Verification:

kafka-console-consumer --topic orders --from-beginning --bootstrap-server localhost:9092
kafka-console-consumer --topic payments --from-beginning --bootstrap-server localhost:9092
# Should see no messages (transaction aborted)

Test 2: Crash Consumer Mid-Processing

// Add after printing "Processed: ..."
throw new RuntimeException("Simulated failure");

// Verify: Same message is reprocessed (idempotent handling required)

Lab 2: Kafka Streams EOS Implementation

Goal: Process a stream with built-in EOS guarantees using Kafka Streams.

Code:

import org.apache.kafka.streams.*;
import org.apache.kafka.streams.kstream.*;
import java.util.Properties;

public class KafkaStreamsEOS {
    public static void main(String[] args) {
        Properties props = new Properties();
        props.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
        props.put(StreamsConfig.APPLICATION_ID_CONFIG, "eos-streams-app");
        props.put(StreamsConfig.PROCESSING_GUARANTEE_CONFIG, "exactly_once_v2");
        props.put(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, 
                  org.apache.kafka.common.serialization.Serdes.String().getClass());
        props.put(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG, 
                  org.apache.kafka.common.serialization.Serdes.String().getClass());

        StreamsBuilder builder = new StreamsBuilder();
        
        builder.stream("orders")
               .mapValues(value -> "Processed: " + value)
               .to("processed-orders");

        KafkaStreams streams = new KafkaStreams(builder.build(), props);
        streams.start();

        // Add shutdown hook
        Runtime.getRuntime().addShutdownHook(new Thread(streams::close));
    }
}

Key Points:

  • exactly_once_v2 handles state stores, changelogs, and offsets atomically
  • No need for manual transaction APIs!
  • Works only within Kafka (external systems still need separate handling)

Lab 3: Expected Output & Validation

Check for Duplicates:

kafka-console-consumer --topic payments --from-beginning \
  --property print.key=true --bootstrap-server localhost:9092

Expected: Each order-123 appears exactly once (no duplicates).

Inspect Transaction Logs:

kafka-console-consumer --topic __transaction_state --from-beginning \
  --bootstrap-server localhost:9092 \
  --formatter "kafka.coordinator.transaction.TransactionLog\$TransactionLogMessageFormatter"

Expected: See transaction state transitions (Ongoing → PrepareCommit → CompleteCommit).

Check Consumer Offsets:

kafka-consumer-groups --describe --group eos-consumer-group \
  --bootstrap-server localhost:9092

Expected: Offsets committed only after successful transaction.

Key Takeaways

  1. Configuration: Requires transactional.id (unique per instance) and enable.idempotence=true.

  2. Lifecycle: initTransactions() → beginTransaction() → send() → commitTransaction() or abortTransaction().

  3. Atomicity: All messages in a transaction are committed or aborted atomically across multiple partitions.

  4. Read-Process-Write: Use sendOffsetsToTransaction() to commit consumer offsets atomically with produced messages.

  5. Error Handling: Fatal errors (fencing, sequence errors) require producer close. Transient errors allow abort + retry.

  6. Performance: ~10-20% throughput reduction. Batch 100-1000 messages per transaction to amortize overhead.

  7. Fencing: Old producer instances are automatically fenced when new instance with same transactional.id starts.

  8. Hands-On Practice: The labs above provide runnable code for transactional producers, consumers, and Kafka Streams with EOS.

What's Next?

Next Module: 10.1.3 – Consumer Isolation Levels

What We Covered:

  • ✅ Producer transaction API
  • ✅ Configuration and initialization
  • ✅ Transaction patterns (simple, read-process-write, batch)
  • ✅ Error handling strategies

What's Coming:

  • ❌ Consumer isolation levels (read_committed vs read_uncommitted)
  • ❌ How consumers handle transaction markers
  • ❌ End-to-end EOS pipeline design

References:

  • Transaction protocol → 10.1.1
  • Producer configs → M3: Producers