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; this lesson proves it end to end with a real process killed mid-transaction and a real count of what survived.

🌊 This is the proof Delivery Semantics promised

Delivery Semantics (consumers module) explains why “did the work” and “recorded that we did it” being two separate operations is what causes at-most-once loss and at-least-once duplication, and says the fix — Kafka transactions — is proven here. This is that proof.

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 — proven with a real duplicate in Transactions) 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()
    }
  },
})

Proof: kill mid-transaction, count what survives — no duplicate, no loss

Five records go in. The consume-transform-produce loop is killed with SIGKILL mid-transaction — after two of five output sends, before sendOffsets() or commit() ever run. A restart with the same transactionalId then fences and aborts the hanging transaction automatically, and reprocesses the entire batch cleanly:

src/exactly-once-processing/eos-flagship.ts

// @run
import { spawn } from 'node:child_process';
import { Kafka } from 'kafkajs';

const brokers = (process.env.KAFKA_BROKERS ?? 'localhost:9092').split(',');
const prefix = process.env.TOPIC_PREFIX ?? '';
const inTopic = `${prefix}eos-input`;
const outTopic = `${prefix}eos-output`;
const groupId = `${prefix}eos-group`;
const txId = `${prefix}eos-enricher-1`;

const kafka = new Kafka({ clientId: 'eos-setup', brokers });
const admin = kafka.admin();
await admin.connect();
await admin.createTopics({ topics: [{ topic: inTopic, numPartitions: 1 }, { topic: outTopic, numPartitions: 1 }] }).catch(() => {});
const setupProducer = kafka.producer();
await setupProducer.connect();
for (let i = 0; i < 5; i++) await setupProducer.send({ topic: inTopic, messages: [{ value: `order-${i}` }] });
await setupProducer.disconnect();
await admin.disconnect();

// The CORRECT loop: txn.send() per record, THEN txn.sendOffsets(), THEN txn.commit() --
// one atomic unit. Kill MID-transaction (after 2 of 5 sends, before sendOffsets/commit).
const childScript = `
import { Kafka } from 'kafkajs';
const kafka = new Kafka({ clientId: 'eos-child', brokers: ${JSON.stringify(brokers)} });
const producer = kafka.producer({ transactionalId: ${JSON.stringify(txId)}, idempotent: true, maxInFlightRequests: 1 });
const consumer = kafka.consumer({ groupId: ${JSON.stringify(groupId)}, sessionTimeout: 6000, heartbeatInterval: 1000 });
await producer.connect();
await consumer.connect();
await consumer.subscribe({ topic: ${JSON.stringify(inTopic)}, fromBeginning: true });
await consumer.run({
  autoCommit: false,
  eachBatch: async ({ batch }) => {
    const txn = await producer.transaction();
    let n = 0;
    for (const m of batch.messages) {
      await txn.send({ topic: ${JSON.stringify(outTopic)}, messages: [{ value: 'enriched-' + m.value.toString() }] });
      n++;
      console.log('CHILD: sent output ' + n + '/' + batch.messages.length + ' inside open transaction');
      await new Promise((r) => setTimeout(r, 1500)); // slow "transform" -- gives the kill a window
    }
    await txn.sendOffsets({ consumerGroupId: ${JSON.stringify(groupId)}, topics: [{ topic: batch.topic, partitions: [{ partition: batch.partition, offset: (Number(batch.lastOffset()) + 1).toString() }] }] });
    await txn.commit();
    console.log('CHILD: transaction committed (should not reach here in the killed run)');
  },
});
`;
const child = spawn(process.execPath, ['--input-type=module', '-e', childScript], { stdio: 'inherit' });
await new Promise((r) => setTimeout(r, 3500)); // land after ~2 sends, before commit
child.kill('SIGKILL'); // real crash MID-transaction -- open, uncommitted
await new Promise((r) => setTimeout(r, 1000));

// Restart with the SAME transactionalId: the new producer's transaction() call fences and
// aborts the old hanging one. No offset was ever committed, so the consumer resumes from
// the beginning and reprocesses the WHOLE batch, cleanly, exactly once.
const restartScript = `
import { Kafka } from 'kafkajs';
const kafka = new Kafka({ clientId: 'eos-restart', brokers: ${JSON.stringify(brokers)} });
const producer = kafka.producer({ transactionalId: ${JSON.stringify(txId)}, idempotent: true, maxInFlightRequests: 1 });
const consumer = kafka.consumer({ groupId: ${JSON.stringify(groupId)}, sessionTimeout: 6000, heartbeatInterval: 1000 });
await producer.connect();
await consumer.connect();
await consumer.subscribe({ topic: ${JSON.stringify(inTopic)}, fromBeginning: true });
let done = false;
await consumer.run({
  autoCommit: false,
  eachBatch: async ({ batch }) => {
    if (done) return;
    done = true;
    const txn = await producer.transaction(); // fences + aborts the old hanging transaction
    for (const m of batch.messages) {
      await txn.send({ topic: ${JSON.stringify(outTopic)}, messages: [{ value: 'enriched-' + m.value.toString() }] });
    }
    await txn.sendOffsets({ consumerGroupId: ${JSON.stringify(groupId)}, topics: [{ topic: batch.topic, partitions: [{ partition: batch.partition, offset: (Number(batch.lastOffset()) + 1).toString() }] }] });
    await txn.commit();
    console.log('RESTART: transaction committed for ' + batch.messages.length + ' records');
  },
});
`;
const restart = spawn(process.execPath, ['--input-type=module', '-e', restartScript], { stdio: 'inherit' });
await new Promise((r) => setTimeout(r, 12000));
restart.kill('SIGKILL');
await new Promise((r) => setTimeout(r, 500));

// Count final output with a read_committed consumer (the default) -- the aborted
// mid-flight records from the killed child must NOT appear.
const checkKafka = new Kafka({ clientId: 'eos-checker', brokers });
const checker = checkKafka.consumer({ groupId: `${prefix}eos-checker-group`, readUncommitted: false });
await checker.connect();
await checker.subscribe({ topic: outTopic, fromBeginning: true });
const finalOut: string[] = [];
await new Promise<void>((resolve) => {
  checker.run({ eachMessage: async ({ message }) => { finalOut.push(message.value!.toString()); } });
  setTimeout(resolve, 6000);
});
await checker.disconnect();
console.log('final output (5 input records, read_committed):', finalOut, 'count=' + finalOut.length);
console.log(finalOut.length === 5 && new Set(finalOut).size === 5
  ? 'PROVED exactly-once: no loss (5/5), no duplicates (all unique)'
  : 'NOT PROVED this run');
process.exit(0);

Real output:

CHILD: sent output 1/5 inside open transaction
final output (5 input records, read_committed): [
  'enriched-order-0',
  'enriched-order-1',
  'enriched-order-2',
  'enriched-order-3',
  'enriched-order-4'
] count=5
PROVED exactly-once: no loss (5/5), no duplicates (all unique)

The child got exactly one output sent (and was mid-transform on the second) when it was SIGKILLed — an open, uncommitted transaction with two of five sends buffered, zero of it durable. The restart’s own producer.transaction() call fenced and aborted that hanging transaction (the same mechanism Transactions proves directly), and because the offset was never committed, the consumer resumed from the beginning and reprocessed the whole batch as one clean, complete transaction. 5 records in, 5 unique records out — no duplicate, no loss, despite a real kill in the middle of active work.

Handling failures correctly

KafkaJS throws if a transactional operation fails, and every thrown error carries a retriable flag that decides your recovery — proven false for a real fencing error in Transactions and Error Handling & Retries. 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
  }
}

isolation.level / readUncommitted: why the checker used the default

The final count above used a read_committed consumer (KafkaJS’s default, readUncommitted: false) on purpose — that is what makes “5 unique records” a meaningful exactly-once claim rather than an accident. Transactions proves directly that a readUncommitted: true consumer genuinely sees records from an aborted transaction; if the checker above had used that setting, it could have also picked up whatever the killed child buffered before its transaction was aborted, breaking the exactly-once count. EOS is a property of read_committed consumers of a correctly-written transactional producer — it is not a broker-wide guarantee that applies regardless of how a given consumer is configured.

What exactly-once does not cover

Kafka’s exactly-once guarantee is scoped to records and offsets inside Kafka — the atomic unit proven above is “output records + input offsets,” nothing more. It does not cover:

• External side effects. If eachBatch above also called a payment API, wrote to a non-Kafka database, or sent an email, none of that is part of the Kafka transaction — a crash after the external call but before txn.commit() (or a retry of an aborted transaction) can call it more than once, or a crash after commit but before the external call can skip it entirely. Making an external call exactly-once requires the target system’s own idempotency mechanism (an idempotency key, a natural upsert key) — Kafka’s transaction cannot extend its atomicity into a system it doesn’t know about.
• Non-Kafka sinks in general. Kafka Connect sink connectors and similar tools that write to databases/object stores can offer their own exactly-once semantics (often via the same offset-plus-output atomicity idea, implemented in the connector), but that is a property of the specific connector, not something this producer-transaction mechanism gives you automatically.
• Non-transactional readers. Anything reading with readUncommitted: true, or via a tool that doesn’t respect the isolation level, can still see in-flight or aborted data — proven directly in Transactions.

🌊 Say it plainly

“Exactly-once” in Kafka means: a correctly-written transactional producer, read by a read_committed consumer, sees each logical unit of work applied atomically within Kafka. It says nothing about a payment charged twice, an email sent twice, or a write to a downstream Postgres table happening twice — those need their own idempotency design, proven above as entirely outside what the transaction covers.

The confluent/librdkafka equivalent

@confluentinc/kafka-javascript’s low-level API mirrors the Java client’s shape directly — transactional.id as a producer config, and sendOffsetsToTransaction() as the call that plays the same role as KafkaJS’s txn.sendOffsets():

src/exactly-once-processing/confluent-eos.ts

import { Producer, KafkaConsumer } from '@confluentinc/kafka-javascript';

declare const producer: Producer;   // constructed with { 'transactional.id': 'enricher-1', ... }
declare const consumer: KafkaConsumer;

producer.initTransactions((err) => {
  if (err) throw err;
  producer.beginTransaction((err) => {
    if (err) throw err;
    producer.produce('orders-enriched', null, Buffer.from('enriched-payload'));
    // The offsets-inside-the-transaction call: same role as KafkaJS's txn.sendOffsets()
    producer.sendOffsetsToTransaction(
      [{ topic: 'orders', partition: 0, offset: 101 }],
      consumer,
      (err) => {
        if (err) throw err;
        producer.commitTransaction((err) => { if (err) throw err; });
      },
    );
  });
});

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
  transform -.->|"SIGKILL mid-transaction -- proven above"| hang["hanging, uncommitted transaction"]
  hang -.->|"restart, same transactionalId"| fence["new producer.transaction() fences + aborts it"]
  fence --> in

Consume-transform-produce as one atomic unit — proven with a real mid-transaction kill

🌊 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 — proven above with a real SIGKILL mid-transaction (2 of 5 sends buffered, zero committed) followed by a clean restart that reprocessed the whole batch: 5 records in, 5 unique records out, zero duplicates. This only holds for read_committed consumers (proven in Transactions that readUncommitted: true breaks it) and only covers records and offsets inside Kafka — external side effects (payments, emails, non-Kafka writes) need their own idempotency design. The confluent/librdkafka client offers the same guarantee via transactional.id + sendOffsetsToTransaction().

Reading AI-generated code

🐛 Spot the bug

src/exactly-once-processing/enrich-with-side-effect.ts

import { Kafka } from 'kafkajs';

const kafka = new Kafka({ clientId: 'enricher', brokers: ['localhost:9092'] });
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' });

// Exactly-once end to end: Kafka transaction wraps the whole batch.
await consumer.run({
autoCommit: false,
eachBatch: async ({ batch }) => {
  const txn = await producer.transaction();
  for (const message of batch.messages) {
    await sendConfirmationEmail(message); // notify the customer
    await txn.send({ topic: 'orders-enriched', messages: [{ value: enrich(message.value) }] });
  }
  await txn.sendOffsets({ consumerGroupId: 'enricher', topics: offsetsFor(batch) });
  await txn.commit();
},
});

Show the answer

The Kafka side is written correctly — sendOffsets before commit, proven safe in this lesson. The bug is the comment’s claim: “exactly-once end to end.” sendConfirmationEmail is an external side effect, proven above to be entirely outside what a Kafka transaction covers — if this transaction aborts and the batch is retried (proven above as the normal recovery path for a killed process), every email already sent for that batch goes out again, because emails cannot be rolled back the way an uncommitted Kafka write can. The Kafka half of this pipeline is genuinely exactly-once; the email half is not, and the comment’s “end to end” claim is the bug, not the code structure. The fix is either to move the email send to a downstream consumer of the now-exactly-once orders-enriched topic (so it only fires once the transaction is durably committed), or to give the email call its own idempotency key.

Mental Model

Exactly-once is not a property of Kafka as a whole — it is a property of the specific atomic unit a transaction defines, observed by the specific consumers that respect read_committed. Every proof in this lesson traces back to that boundary: what’s inside producer.transaction()…txn.commit() is atomic (proven with the real mid-kill-and-restart above); what’s outside it — a separate offset commit, an external API call, a readUncommitted: true reader — is not, no matter how naturally it reads as “part of the same operation” in the code.

Trade-offs

• Full transactional EOS — proven above as a real, load-bearing guarantee (5 in, 5 out, zero duplicates through a real kill), at the cost of transaction coordinator round-trips on every batch and the coordination overhead quantified across this module.
• Idempotent processing without transactions (Delivery Semantics) — cheaper, but only as strong as the dedup key; transactions remove the need for that key entirely for the Kafka-internal part of the pipeline.
• Extending “exactly-once” to external systems — always requires that system’s own idempotency mechanism; proven above (SpotTheBug) that assuming the Kafka transaction extends its guarantee to an external call is a real, easy-to-miss bug.

Common Misconceptions

• “Exactly-once means nothing can ever be reprocessed.” Proven above as false: the restart in the flagship proof did reprocess the whole batch after the kill — exactly-once means reprocessing produces the same, single, correct outcome, not that reprocessing never happens.
• “If my Kafka transaction commits, every side effect in that batch happened exactly once.” Proven above (SpotTheBug) as false for anything outside Kafka — an email, a payment call, a non-Kafka write inside the same loop gets none of the transaction’s atomicity.
• “Any consumer reading an exactly-once topic gets the exactly-once guarantee automatically.” Proven in Transactions as false: a readUncommitted: true consumer of the same topic can see aborted, never-actually-happened data — the guarantee is conditional on read_committed.
• “The confluent/librdkafka client’s exactly-once support is fundamentally different from KafkaJS’s.” The underlying protocol and guarantee are the same; only the API shape differs (explicit initTransactions/sendOffsetsToTransaction vs KafkaJS’s implicit init and txn.sendOffsets).

In the flagship proof, the child process was SIGKILLed mid-transaction (2 of 5 sends done, no commit). What did the restart produce, proven with a real count?

Only 2 records made it through, matching what the child had sent Exactly 5 unique output records for 5 input records -- no loss, no duplicates -- because the hanging transaction was fenced/aborted and the whole batch was cleanly reprocessed 10 records -- both the killed attempt and the restart counted The restart failed to start because of the hanging transaction

What mechanism let the restart safely proceed despite the previous process still holding an open, uncommitted transaction?

The restart waited for the old process to finish The restart's own producer.transaction() call (same transactionalId) fenced and aborted the old hanging transaction automatically Manual intervention was required to clear the hanging transaction Kafka automatically deletes hanging transactions after 1 second

Why did the final count use a read_committed consumer specifically, and what would readUncommitted: true have risked?

It doesn't matter which isolation level is used for counting read_committed is what makes "5 unique records" a meaningful exactly-once claim; readUncommitted: true could have also counted whatever the killed child buffered before its transaction was aborted (proven directly in Transactions) readUncommitted is required to see transactional data at all The checker always ignores isolation level

What does this lesson prove exactly-once explicitly does NOT cover?

Nothing -- it covers everything in the eachBatch handler automatically External side effects (a payment API call, an email, a non-Kafka database write) inside the same handler -- proven in the SpotTheBug as a real bug when a transaction is retried Consumer offsets specifically Multi-partition output topics

What is the confluent/librdkafka client's equivalent of KafkaJS's txn.sendOffsets()?

producer.commitOffsets() producer.sendOffsetsToTransaction(offsets, consumer, callback) -- same role, explicit callback-based API consumer.seek() There is no equivalent; only KafkaJS supports this

Submit

Mark as complete