How do Kafka transactional messages work?

Kafka Transactional Messages

Kafka transactional messages are an important feature introduced in Kafka 0.11, allowing Producers to send multiple messages to multiple Topics and Partitions, ensuring that these messages either all succeed or all fail. This is crucial for scenarios that require data consistency guarantees.

Basic Concepts of Transactional Messages

1. Transaction Definition

A Kafka transaction is an atomic operation on a group of messages, where all messages in the transaction either all succeed or all fail.

Characteristics:

• Atomicity: All messages in the transaction either all succeed or all fail
• Consistency: System state remains consistent after transaction execution
• Isolation: Other transactions do not see intermediate states during transaction execution
• Durability: Transaction results are permanently saved after commit

2. Transactional ID

Each Producer needs to configure a unique transactional ID.

Purpose:

• Identify the Producer's transaction identity
• Used for failure recovery and idempotence guarantees
• Ensure Producer can recover incomplete transactions after restart

Configuration:

properties
# Transactional ID
transactional.id=my-transactional-id-1

Working Principle of Transactional Messages

1. Transaction Initialization

When the Producer starts, it registers the transactional ID with the Coordinator.

Process:

1. Producer sends registration request to Coordinator
2. Coordinator records mapping between transactional ID and Producer
3. Coordinator assigns a PID (Producer ID) to the Producer

2. Transaction Start

Producer calls beginTransaction() to start a new transaction.

Process:

1. Producer requests to start transaction from Coordinator
2. Coordinator records transaction start time
3. Producer starts collecting messages

3. Send Messages

Producer sends messages to multiple Topics and Partitions within the transaction.

Process:

1. Producer sends messages to Broker
2. Broker writes messages to log but does not mark as consumable
3. Broker records that messages belong to current transaction

4. Transaction Commit or Abort

Producer calls commitTransaction() or abortTransaction().

Commit Process:

1. Producer sends commit request to Coordinator
2. Coordinator sends commit markers to all relevant Brokers
3. Broker marks messages in transaction as consumable
4. Coordinator records transaction completion

Abort Process:

1. Producer sends abort request to Coordinator
2. Coordinator sends abort markers to all relevant Brokers
3. Broker deletes messages in transaction
4. Coordinator records transaction abort

Transactional Message Configuration

Producer Configuration

properties
# Enable transaction support
enable.idempotence=true

# Transactional ID
transactional.id=my-transactional-id-1

# Transaction timeout
transaction.timeout.ms=60000

# Retry count
retries=Integer.MAX_VALUE

# Maximum in-flight requests
max.in.flight.requests.per.connection=5

Broker Configuration

properties
# Transaction state log replica factor
transaction.state.log.replication.factor=3

# Transaction state log minimum in-sync replicas
transaction.state.log.min.isr=2

# Transaction state log segment size
transaction.state.log.segment.bytes=104857600

# Transaction timeout
transactional.id.expiration.ms=604800000

Using Transactional Messages

Basic Usage Example

java
// Create Producer
Properties props = new Properties();
props.put("bootstrap.servers", "localhost:9092");
props.put("transactional.id", "my-transactional-id-1");
props.put("enable.idempotence", "true");

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

// Initialize transactions
producer.initTransactions();

try {
    // Begin transaction
    producer.beginTransaction();
    
    // Send message to Topic1
    producer.send(new ProducerRecord<>("topic1", "key1", "value1"));
    
    // Send message to Topic2
    producer.send(new ProducerRecord<>("topic2", "key2", "value2"));
    
    // Commit transaction
    producer.commitTransaction();
} catch (Exception e) {
    // Abort transaction
    producer.abortTransaction();
}

Integration with Database Transactions

java
// Create Producer
KafkaProducer<String, String> producer = new KafkaProducer<>(props);
producer.initTransactions();

// Get database connection
Connection conn = dataSource.getConnection();

try {
    // Begin Kafka transaction
    producer.beginTransaction();
    
    // Begin database transaction
    conn.setAutoCommit(false);
    
    // Send Kafka message
    producer.send(new ProducerRecord<>("topic1", "key1", "value1"));
    
    // Execute database operation
    Statement stmt = conn.createStatement();
    stmt.executeUpdate("INSERT INTO table1 VALUES (1, 'data')");
    
    // Commit database transaction
    conn.commit();
    
    // Commit Kafka transaction
    producer.commitTransaction();
} catch (Exception e) {
    // Rollback database transaction
    conn.rollback();
    
    // Abort Kafka transaction
    producer.abortTransaction();
} finally {
    conn.close();
}

Transactional Message Isolation Levels

Read Committed (Default)

Consumers can only read messages from committed transactions.

Characteristics:

• Ensures data consistency
• Avoids reading uncommitted data
• Suitable for most scenarios

Configuration:

properties
# Consumer configuration
isolation.level=read_committed

Read Uncommitted

Consumers can read all messages, including messages from uncommitted transactions.

Characteristics:

• Better performance
• May read uncommitted data
• Suitable for scenarios with low consistency requirements

Configuration:

properties
# Consumer configuration
isolation.level=read_uncommitted

Transactional Message Use Cases

1. Data Consistency Guarantee

Scenario Description: Need to ensure data consistency between multiple systems.

Examples:

• Order system and inventory system
• Payment system and accounting system
• User center and permission system

2. Idempotence Guarantee

Scenario Description: Need to ensure messages are not processed multiple times.

Examples:

• Payment notifications
• Order status updates
• Inventory deductions

3. Event Sourcing

Scenario Description: Need to record all state change events.

Examples:

• Account transaction records
• Order status flow
• System operation logs

Performance Impact of Transactional Messages

Performance Overhead

1. Network Overhead

   • Need to communicate with Coordinator
   • Need to communicate with multiple Brokers
   • Increased network round trips

2. Storage Overhead

   • Need to store transaction state
   • Need to store transaction logs
   • Increased disk I/O

3. Latency Overhead

   • Need to wait for transaction commit
   • Need to wait for all Broker acknowledgments
   • Increased end-to-end latency

Performance Optimization

1. Batch Commit

   • Send multiple messages in one transaction
   • Reduce transaction commit frequency
   • Improve throughput

2. Reasonably Set Timeout

   • Set transaction timeout based on business requirements
   • Avoid excessively long transaction timeouts
   • Balance reliability and performance

3. Optimize Network Configuration

   • Increase Broker network bandwidth
   • Reduce network latency between Coordinator and Brokers
   • Optimize network topology

Monitoring Transactional Messages

Monitoring Metrics

• TransactionStarted: Transaction start count
• TransactionCommitted: Transaction commit count
• TransactionAborted: Transaction abort count
• TransactionTimeout: Transaction timeout count

Monitoring Commands

bash
# View transaction status
kafka-transactions --bootstrap-server localhost:9092 \
  --describe --transactional-id my-transactional-id-1

# View Producer transaction status
kafka-producer-perf-test --topic test-topic \
  --num-records 1000 --record-size 1000 \
  --throughput 10000 --producer-props \
  enable.idempotence=true \
  transactional.id=my-transactional-id-1

Best Practices

1. Reasonably Design Transaction Scope

• Moderate number of messages in transaction
• Avoid long-running transactions
• Design transaction boundaries based on business requirements

2. Handle Transaction Failures

• Implement transaction failure retry mechanism
• Record transaction failure logs
• Establish transaction compensation mechanism

3. Monitor Transaction Status

• Real-time monitoring of transaction commits and aborts
• Monitor transaction timeouts
• Timely discover and handle exceptions

4. Optimize Performance

• Batch send messages
• Reasonably set transaction timeout
• Optimize network and storage configuration

By properly using Kafka transactional messages, strong consistency guarantees can be achieved in distributed systems while maintaining high system performance and availability.