Implementing Transactions

1. Understanding Transactional Concept

Kafka transactions provide atomicity across multiple producers and consumer groups, ensuring that either all operations succeed or all fail. This is crucial for maintaining data consistency in distributed systems.

Concept	Description	Use Case
Idempotent Producer	Ensures that messages are delivered exactly once, even in case of retries.	Preventing duplicate messages in high-throughput systems
Transactional Producer	Supports multi-partition writes with atomic commit/abort semantics.	Ensuring consistency across multiple topics or partitions
Consumer Group Offset Management	Allows consumers to commit offsets only after processing is complete.	Maintaining exactly-once processing semantics

Example: Transactional Producer Setup

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

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

Note: The transactional.id must be unique per producer instance and remain consistent across restarts to ensure idempotency.

2. Enabling Transactional Producer

Property	Value	Description
enable.idempotence	`true`	Enables idempotent producer to ensure exactly-once delivery semantics
transactional.id	`string`	Unique identifier for the transactional producer; required when `enable.idempotence` is true
max.in.flight.requests.per.connection	`5`	Must be set to 5 or less when idempotence is enabled

Example: Configuring Transactional Producer

const { Kafka } = require('kafkajs');

const kafka = new Kafka({
  clientId: 'my-app',
  brokers: ['localhost:9092']
});

const producer = kafka.producer({
  idempotent: true,
  transactionalId: 'my-transactional-producer'
});

await producer.connect();

Note: The transactionalId must be unique per producer instance and remain consistent across restarts to maintain transaction state.

Warning: Setting max.in.flight.requests.per.connection to a value greater than 5 will cause the producer to fail to initialize when idempotence is enabled.

3. Initializing Transactions

Property	Description	Default Value	Notes
enable.idempotence	Enables idempotent producer behavior	false	Required for transactional producers
transactional.id	Unique identifier for the transactional producer	null	Must be set when idempotence is enabled
max.in.flight.requests.per.connection	Maximum number of unacknowledged requests	5	Must be 1 when idempotence is enabled

Example: Initializing Transactional Producer

const { Kafka } = require('kafkajs');

const kafka = new Kafka({
  clientId: 'my-app',
  brokers: ['localhost:9092']
});

const producer = kafka.producer({
  idempotent: true,
  transactionalId: 'my-transactional-producer'
});

await producer.connect();
await producer.initTransactions();

Note: The initTransactions() method must be called before any transactional operations can begin.

4. Beginning Transaction

Method	Description	Parameters	Notes
beginTransaction()	Begins a new transaction	None	Must be called before any produce or consume operations within the transaction

Example: Beginning a Transaction

const { Kafka } = require('kafkajs');

const kafka = new Kafka({
  clientId: 'my-app',
  brokers: ['localhost:9092']
});

const producer = kafka.producer();

await producer.connect();
await producer.beginTransaction(); // Start the transaction

// Perform operations within the transaction
await producer.send({
  topic: 'my-topic',
  messages: [
    { value: 'message-1' },
    { value: 'message-2' }
  ]
});

await producer.commitTransaction(); // Commit the transaction

Note: Transactions in Kafka require the transactional.id to be configured on the producer.

5. Sending Transactional Messages

Property	Description	Default Value
enable.idempotence	Enables idempotent producer to ensure messages are not duplicated	false
transactional.id	Unique identifier for transactional producer	null
max.in.flight.requests.per.connection	Maximum number of unacknowledged requests on a connection	5

Example: Sending Transactional Messages

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("enable.idempotence", "true");
props.put("transactional.id", "my-transactional-id");

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

producer.initTransactions();

try {
    producer.beginTransaction();
    producer.send(new ProducerRecord<>("my-topic", "key1", "value1"));
    producer.send(new ProducerRecord<>("my-topic", "key2", "value2"));
    producer.commitTransaction();
} catch (KafkaException e) {
    producer.abortTransaction();
}

Note: Transactional messages must be sent within a transaction boundary using beginTransaction(), commitTransaction(), and abortTransaction().

Warning: If enable.idempotence is not set to true, the producer will not be able to send transactional messages.

6. Sending Offsets in Transaction

Property	Description	Default Value	Notes
enable.idempotence	Enables idempotent producer to ensure no duplicates	false	Required for transactional producers
transactional.id	Unique identifier for the transactional producer	null	Must be set when idempotence is enabled
max.in.flight.requests.per.connection	Maximum number of unacknowledged requests	5	Must be set to 1 when idempotence is enabled

Example: Sending Offsets in Transaction

const { Kafka } = require('kafkajs');

const kafka = new Kafka({
  clientId: 'my-app',
  brokers: ['localhost:9092']
});

const producer = kafka.producer();
const consumer = kafka.consumer({ groupId: 'my-group' });

async function run() {
  await producer.connect();
  await consumer.connect();

  // Start transaction
  await producer.sendOffsets({
    groupId: 'my-group',
    topics: [
      {
        topic: 'my-topic',
        partitions: [
          { partition: 0, offset: '123' }
        ]
      }
    ]
  });

  // Commit offsets within transaction
  await producer.commitOffsets({
    groupId: 'my-group',
    topics: [
      {
        topic: 'my-topic',
        partitions: [
          { partition: 0, offset: '123' }
        ]
      }
    ]
  });
}

run().catch(console.error);

Note: Transactional offsets must be committed within the same transaction as the messages they reference.

Warning: If a transaction is aborted, all offsets committed within that transaction are discarded.

7. Committing Transaction

Method	Description	Notes
commitTransaction()	Commits the ongoing transaction	Must be called after all produce, update, or delete operations are completed

Example: Committing a Transaction

const { Kafka } = require('kafkajs');

const kafka = new Kafka({
  clientId: 'my-app',
  brokers: ['localhost:9092']
});

const producer = kafka.producer();

await producer.connect();
await producer.send({
  topic: 'my-topic',
  messages: [
    { value: 'Hello Kafka' }
  ]
});

await producer.commitTransaction();

console.log('Transaction committed successfully');

Note: If a transaction is not explicitly committed or aborted, it will timeout and be automatically aborted by the broker.

8. Aborting Transaction

Method	Description	Notes
abortTransaction()	Aborts the current transaction and rolls back all messages produced within it.	Must be called before the transaction timeout or after a successful commit.

Example: Aborting a Transaction

const { Kafka } = require('kafkajs');

const kafka = new Kafka({ brokers: ['localhost:9092'] });
const producer = kafka.producer();

await producer.connect();
await producer.transaction().then(async (transaction) => {
  try {
    await transaction.send('topic', { key: 'key1', value: 'value1' });
    // Simulate an error condition
    throw new Error('Simulated failure');
  } catch (err) {
    await transaction.abort();
    console.log('Transaction aborted');
  }
});

Note: If abortTransaction() is not called explicitly, the transaction will timeout and be automatically aborted by Kafka.

9. Configuring Transaction Timeout

Property	Default Value	Description
transaction.timeout.ms	`60000`	Timeout for transaction commits and aborts. Used by the producer to determine how long to wait before considering a transaction as failed.
transactional.id	`null`	Unique identifier for a transactional producer. Required when using idempotent producers with transactions.
max.in.flight.requests.per.connection	`5`	Maximum number of unacknowledged requests on a connection before the producer blocks. Helps ensure transactional order.

Example: Setting Transaction Timeout

const producer = kafka.producer({
  idempotent: true,
  transactionalId: 'my-transactional-producer',
  transactionTimeout: 60000
});

Note: The transaction.timeout.ms must be less than the max.poll.interval.ms to prevent consumer rebalances during transaction processing.

10. Setting Read Committed

Property	Values	Description
isolation.level	`read_committed`	Ensures that only committed transactions are read, preventing reading uncommitted data.
isolation.level	`read_uncommitted`	Allows reading uncommitted data, which can lead to dirty reads.

Example: Configuring Read Committed in Producer

{
  "bootstrap.servers": "localhost:9092",
  "isolation.level": "read_committed",
  "enable.idempotence": true
}

Example: Configuring Read Committed in Consumer

{
  "bootstrap.servers": "localhost:9092",
  "isolation.level": "read_committed",
  "group.id": "transactional-group"
}

11. Understanding Exactly-Once Semantics

Exactly-once semantics (EOS) in Apache Kafka ensures that messages are processed exactly once by consumers, even in the presence of failures or retries. This is critical for systems where duplicate processing can lead to incorrect state or financial loss.

Aspect	Description	Implementation
Idempotent Producers	Producer ensures that messages are not duplicated during retries.	Enabled via `enable.idempotence=true` in producer config.
Transactional Producers	Producer can send messages in a transaction, ensuring atomicity across partitions.	Uses `beginTransaction()`, `send()`, and `commitTransaction()`.
Consumer Offset Management	Consumers commit offsets only after successful processing.	Uses `enable.auto.commit=false` and manual offset commits.

Example: Producer with Idempotent and Transactional Features

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("enable.idempotence", "true");
props.put("transactional.id", "my-transactional-id");

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

try {
    producer.beginTransaction();
    producer.send(new ProducerRecord<>("topic", "key", "value"));
    producer.commitTransaction();
} catch (Exception e) {
    producer.abortTransaction();
}

Note: EOS requires a minimum Kafka version of 0.11.0 for idempotent producers and transactional support.

Key Benefits of Exactly-Once Semantics

Eliminates duplicate processing in case of retries or failures.
Ensures data consistency across systems using Kafka as a message broker.
Enables reliable processing in financial or inventory systems where duplicates are costly.