Walk through the exact producer/consumer call sequence for a transactional read-process-write loop, and explain what sendOffsetsToTransaction does.

## One-time setup ``` Producer config: transactional.id = <stable, unique per logical task> (enable.idempotence=true is implied) Consumer config: enable.auto.commit = false isolation.level = read_committed // for the *next* stage ``` Call **`producer.initTransactions()`** exactly once at startup. This: - Registers the `transactional.id` with the **transaction coordinator** (a broker role). - **Fences** any earlier producer instance that used the same `transactional.id` by bumping the **producer epoch**, so a zombie predecessor can no longer write. - Aborts or completes any in-flight transaction left over from a prior crash, giving you a clean slate. ## The per-batch loop ``` while (running) { records = consumer.poll(timeout) producer.beginTransaction() try { Map<TopicPartition, OffsetAndMetadata> offsets = new HashMap<>() for (record : records) { // transform producer.send(new ProducerRecord<>(outTopic, ...)) offsets.put( new TopicPartition(record.topic(), record.partition()), new OffsetAndMetadata(record.offset() + 1)) // NEXT offset to read } producer.sendOffsetsToTransaction(offsets, consumer.groupMetadata()) producer.commitTransaction() } catch (Exception e) { producer.abortTransaction() } } ``` ### Key details - **`record.offset() + 1`** — a committed offset is the position of the *next* record to consume, so you store last-processed + 1, exactly as a normal `commitSync` would. - **`sendOffsetsToTransaction(offsets, groupMetadata)`** — instead of the consumer committing offsets itself, the *producer* sends them to the transaction coordinator. They are buffered and only written to `__consumer_offsets` when the transaction commits. So the offset advance is **part of** the transaction. - **`consumer.groupMetadata()`** — supplies the consumer's group id plus its **member id** and **generation id**. The coordinator uses these to fence a stale (rebalanced-out) consumer: if this consumer's generation is no longer current, the offset commit is rejected, preventing a zombie from corrupting offsets. (The deprecated overload taking a bare `group.id` string lacks this fencing.) - **`commitTransaction()`** writes transaction *commit markers* to every involved partition (output topics and `__consumer_offsets`). It blocks until those are durable. **`abortTransaction()`** writes abort markers; `read_committed` consumers will skip the aborted records. ## Why this is atomic The output records and the offset records are tied to the same producer transaction. The coordinator's two-phase commit makes both sets of commit markers appear together. A crash before `commitTransaction()` leaves the whole transaction aborted, so neither the output nor the offset advance is visible, and the loop safely reprocesses. ## Common mistakes - Calling `consumer.commitSync()` anywhere — that commits offsets *outside* the transaction and breaks atomicity. - Storing `record.offset()` instead of `+ 1` — causes one record of reprocessing each restart. - Creating a new `transactional.id` per run — defeats fencing and recovery.