What a transaction actually writes to the log

Key points

• Aborted records are not removed from the log. They are written, and a control record - a transaction marker - is appended afterwards saying commit or abort.
• A read_committed consumer reads the marker and skips the aborted batches. That means transactions add write amplification and give consumers extra work, which is the real cost of exactly-once.
• The last stable offset is the offset before the earliest still-open transaction. A read_committed consumer cannot read past it, because those records might yet abort.
• This is the failure that surprises people: a single long-running open transaction holds the LSO back while the high watermark advances, so transactional consumers stall on a topic that is visibly receiving data.
• transaction.timeout.ms exists precisely to bound that. A transaction the coordinator has not heard about is aborted, releasing the LSO.
• Fencing is what makes this survive a crash. A restarted producer with the same transactional.id bumps the epoch, so a zombie instance still holding the old epoch has its writes rejected - only one live producer per id, guaranteed.

Example

// Read-process-write, atomically, entirely inside Kafka.
producer.initTransactions();          // bumps the epoch, fencing any zombie

while (running) {
    ConsumerRecords<String, Order> records = consumer.poll(Duration.ofMillis(200));
    if (records.isEmpty()) { continue; }

    producer.beginTransaction();
    try {
        for (ConsumerRecord<String, Order> r : records) {
            producer.send(new ProducerRecord<>("orders-enriched", r.key(), enrich(r.value())));
        }

        // The offsets join the transaction - this is what makes it atomic.
        // Without it, output could be committed while progress was not.
        producer.sendOffsetsToTransaction(offsetsOf(records), consumer.groupMetadata());

        producer.commitTransaction();
    } catch (ProducerFencedException e) {
        producer.close();             // another instance took over - exit, do not retry
        throw e;
    } catch (KafkaException e) {
        producer.abortTransaction();  // records stay in the log; a marker hides them
    }
}

/*
 * In the log after an abort - nothing is deleted:
 *
 *   offset 5001  [record]        <- written, then hidden
 *   offset 5002  [record]        <- written, then hidden
 *   offset 5003  ABORT marker    <- read_committed consumers skip 5001-5002
 *
 * And the stall to watch for:
 *   high watermark      9000   (data is arriving)
 *   last stable offset  5000   (one transaction open since offset 5000)
 *   -> a read_committed consumer sees NOTHING past 5000 until it resolves
 */

Aborted records stay in the log behind a marker, and one open transaction blocks every read_committed consumer at the last stable offset.