Exactly-Once Processing

The idea in one sentence

The classic exactly-once use case is consume-transform-produce — read from an input topic, transform, write to an output topic — and Kafka makes it exactly-once by committing the input offsets inside the same transaction as the output records.

Why offsets must join the transaction

Consider reading orders, enriching each, and writing to orders-enriched. If you commit the output in a transaction but commit the input offset separately, a crash between the two either reprocesses (duplicates the output) or skips (loses input). The fix: commit the consumer’s offsets as part of the producer’s transaction with txn.sendOffsets(). Now output records and input progress commit or abort together.

const producer = kafka.producer({ transactionalId: 'enricher-1', idempotent: true, maxInFlightRequests: 1 })
const consumer = kafka.consumer({ groupId: 'enricher', readUncommitted: false })
await producer.connect()
await consumer.connect()
await consumer.subscribe({ topic: 'orders' })

await consumer.run({
  autoCommit: false,                               // offsets are committed inside the transaction
  eachBatch: async ({ batch }) => {
    const txn = await producer.transaction()       // begin
    try {
      for (const message of batch.messages) {
        await txn.send({ topic: 'orders-enriched', messages: [{ key: message.key, value: enrich(message.value) }] })
      }
      // Commit the input offsets in the SAME transaction as the output
      await txn.sendOffsets({
        consumerGroupId: 'enricher',
        topics: [{ topic: batch.topic, partitions: [{ partition: batch.partition, offset: (Number(batch.lastOffset()) + 1).toString() }] }],
      })
      await txn.commit()                           // outputs + offsets are atomic
    } catch (e) {
      await txn.abort()
    }
  },
})

Handling failures correctly

KafkaJS throws if a transactional operation fails, and every thrown error carries a retriable flag that decides your recovery. The common case is retriable: txn.abort() and let the batch run again from the last committed offsets. A non-retriable error (e.retriable === false) means a newer producer has taken over your transactionalId — you cannot continue, so disconnect() this instance and let a healthy one take over:

} catch (e) {
  await txn.abort()                  // roll back and retry the batch
  if (!e.retriable) {                // a zombie was fenced — this producer cannot recover
    await producer.disconnect()      // give up this instance
    throw e
  }
}

flowchart LR
  in["consume from orders"] --> begin["producer.transaction()"]
  begin --> transform["transform + txn.send to orders-enriched"]
  transform --> off["txn.sendOffsets(input offsets)"]
  off --> commit["txn.commit(): outputs + offsets atomic"]
  commit --> in

Consume-transform-produce as one atomic unit

🌊 The one-line takeaway

Exactly-once consume-transform-produce works by putting the consumer’s input offsets inside the producer’s transaction via txn.sendOffsets(), so output records and input progress commit atomically. On failure, txn.abort() and retry; if the thrown error is non-retriable (e.retriable === false, a zombie was fenced) disconnect() and let another instance take over. This gives exactly-once for pipelines that stay inside Kafka.

In consume-transform-produce, why commit input offsets inside the transaction?

To make it faster So output records and input progress commit or abort together — otherwise a crash duplicates or loses data Because offsets cannot be committed any other way To reduce the partition count

Which call commits consumer offsets as part of the producer transaction?

consumer.commitOffsets() txn.sendOffsets({ consumerGroupId, topics }) on the transaction producer.flush() consumer.seek()

How should you handle a non-retriable transaction error (`e.retriable === false`)?

Retry the transaction immediately Treat it as unrecoverable — abort, disconnect the producer, and let another instance take over (a zombie was fenced) Ignore it Switch to auto-commit

What is the correct response to a retriable error inside a transaction?

Commit anyway Abort the transaction and retry the batch from the last committed offsets Delete the topic Disable idempotence

Submit

Mark as complete