Kafka Delivery Semantics and Transactions

Data Engineering & Modern Data Platforms · Lesson 41 · Real-Time Data Engineering and Kafka

Lesson Overview

The landing job commits its group offset after an offset-named object is durable. The player enables idempotence and waits for acks=all. The activity tile writes a file sink from a checkpoint. A design review still asked for a single phrase on the architecture slide: "exactly-once Kafka." The phrase does not say exactly-once of what, across which crash, or including which system. An idempotent producer does not collapse two send calls of the same event_id. A Kafka transaction does not commit the PostgreSQL enrollment row. A commitSync after a lake put does not become a transaction because someone wraps the loop in beginTransaction and forgets that the object store is not a participant.

This lesson names the three delivery semantics in pipeline language, scopes the idempotent producer to the retries of one producer session, and scopes transactions to read-process-write inside Kafka. The lake landing stays at-least-once processing with an idempotent object. The tile keeps the file-sink claim from Structured Streaming with Spark. The enrollment publisher stays the at-least-once handoff Change Data Capture Fundamentals already described, until a later lesson puts a connector on it. The specialty pages Delivery Guarantees, Idempotent Producers, Transactions, and Exactly Once Processing are the protocol references. This lesson decides which hop of the TechLearningPro bus is allowed to use which sentence.

Learning Objectives

By the end of this lesson you will be able to:

1. Define at-most-once, at-least-once, and exactly-once as statements about a named hop and a named crash.
2. Explain what enable.idempotence deduplicates, and what a new send of the same event_id still does.
3. Implement a transactional read-process-write between two Kafka topics with sendOffsetsToTransaction and read_committed.
4. Explain producer fencing with transactional.id and why two instances must not share one id unless one should replace the other.
5. Keep the raw-zone landing on idempotent object keys rather than a Kafka transaction that cannot include the object store.
6. Say why a transaction does not enlist PostgreSQL, and what the outbox hop actually guarantees.
7. Choose the semantics for the player, the landing job, the tile, a derived topic, and the enrollment publisher without using one label for all five.

Prerequisites

• Kafka Producers and Consumers: callbacks, acks=all, the DLQ produce, and the offset-named object.
• Kafka Consumer Groups and Offset Management: next-offset commits, CommitFailedException, and the rule that Spark's checkpoint is a different cursor.
• Kafka Topics, Partitions and Replication: the ISR policy idempotent producers and acks=all assume.
• Change Data Capture Fundamentals: a committed database change and a later publish are two steps.
• Official client types: KafkaProducer.initTransactions, beginTransaction, sendOffsetsToTransaction, commitTransaction, abortTransaction, and consumer isolation.level. Primer: /learn/kafka/duplicate-events and /learn/kafka/message-loss.

Detailed Core Concepts

Semantics are a sentence with a subject

At-most-once means a crash can drop the effect and will not repeat it. At-least-once means a crash can repeat the effect and will not drop a successful one. Exactly-once means a crash neither drops nor repeats the effect, inside the boundary you named. Without the boundary, the words are a slogan. The boundaries on this bus are: player process to partition log; partition log to raw object; partition log to tile files; partition log to a derived Kafka topic; PostgreSQL commit to partition log.

A hop can be at-least-once while the next hop removes the duplicate. That composition is how landing works. It is not a license to call the first hop exactly-once. Say both sentences. Reviewers can then see where a bug in the object name reintroduces duplicates, and where a bug in the producer callback reintroduces loss.

At-most-once on a consumer

Commit the offset, then do the work. A crash after the commit and before the work loses the record. The group will not read it again. Auto-commit can fall into this pattern when the commit timer records a poll that you have not finished. The landing job does not want this. A metrics sample that is allowed to vanish might. Progress events are not that sample. The player can also cause loss without a consumer involved: acks=0, or a process exit that drops the producer buffer, or a callback failure that the UI treats as success. Those are at-most-once publishes, or simple loss, depending on whether a retry exists. The player retries on callback failure and therefore chooses the other side of the trade.

At-least-once on a consumer

Do the work, then commit. A crash after the work and before the commit repeats the work. Lesson 38's loop is this pattern. The repeat is safe if the work replaces an object named by topic, partition, and offset. The repeat is a double count if the work appends to a date-stamped file. The semantics of the hop are at-least-once either way. The semantics of the raw prefix are exactly-once per offset only when the name makes the second write identical to the first. Say "at-least-once delivery into an idempotent sink." Do not shorten it to "exactly-once Kafka" on a slide that also covers the player.

Processing a record and then failing the DLQ publish, without committing, is also at-least-once. The record returns. If the first attempt wrote a partial object, the second attempt must replace it. Committing even though the DLQ publish failed is at-most-once for the quarantine and a silent drop for the contract. The loop does not do that.

The idempotent producer

enable.idempotence=true is the default since Apache Kafka 3.0. The broker assigns the producer a producer id and remembers a sequence number per partition. A retried batch whose sequence the broker has already applied is acknowledged again and not appended twice. Leader changes during those retries are covered. This removes a real class of duplicates: the network timed out, the broker had appended, the client sent the batch again.

It does not remove duplicates from a second send of the same bytes. The sequence number advances. Both records are new. The player's HTTP retry, after a lost response, is a second send with the same event_id and a new sequence. Both offsets exist. Downstream dedupe, if the product definition wants one observation, keys on event_id. The landing job keeps both offsets so the duplicate is visible. Idempotence also does not span two producer processes. A restart creates a new producer id unless you set transactional.id, which is the transactional producer, a stronger protocol, not a synonym for "please dedupe."

Idempotence requires acks=all, retries enabled, and at most five in-flight requests per connection. The client enforces those when the flag is on. Turning the flag off to allow a different in-flight setting reopens duplicate appends on retry. Leave it on for the player and for the landing process's DLQ producer.

Transactions inside Kafka

A transactional producer has a transactional.id. initTransactions registers that id and fences any previous producer that used it, by bumping an epoch. A zombie process that still holds the old epoch receives ProducerFencedException and must stop. Two live instances configured with the same transactional.id will fence each other. Use one id for one logical owner. A landing member that should be replaced on restart can use a stable id. Two members that run at once need different ids, or one of them will be fenced as soon as the other initializes.

The produce path is beginTransaction, one or more send calls, then commitTransaction or abortTransaction. Either the records in that transaction become visible to transactional readers, or none of them do. Consumers that want that visibility set isolation.level=read_committed. The default is read_uncommitted, which returns records from transactions that later abort, as well as ordinary records. A derived-topic consumer that has not set read_committed will see aborted attempts and will count them. If no producer on that topic uses transactions, the two isolation levels show the same log, because there are no transaction markers to honor. Set read_committed on any consumer of a topic that transactional producers write. Leave the progress landing consumer at the default only if learning.progress itself is not transactionally produced. The player in this course is idempotent and not transactional, so landing does not need read_committed to avoid aborted player transactions. A derived topic written with transactions does.

sendOffsetsToTransaction writes the consumer group's next offsets into the same transaction as the output records. The overload that takes ConsumerGroupMetadata includes the generation. At commit, a stale generation aborts the transaction. That is how a zombie member, which lost its partitions in a rebalance, fails the commit instead of publishing a second copy and advancing offsets it no longer owns. The output records and the offset commit succeed or fail together. This is the exactly-once boundary Kafka documents for read-process-write: one input topic, one output topic, offsets committed in the transaction, downstream consumers on read_committed. It is exactly-once of that pipe. It is not exactly-once of a lake put that happened beside it.

transaction.timeout.ms bounds how long a transaction may stay open. The broker will abort it past the timeout. Keep the transaction inside one poll batch. Do not begin a transaction, call a slow external system, and hope the timeout waits. The lake put is the slow external system. It does not belong between beginTransaction and commitTransaction.

What a transaction will not enlist

The Kafka transaction coordinator coordinates Kafka partitions and the offset topic. It does not prepare a PostgreSQL transaction. There is no KafkaProducer method that takes a JDBC connection. The enrollment path remains: the database commits, a publisher reads the change or the outbox row, the publisher appends to learning.enrollment, the publisher marks the outbox. A crash after the append and before the mark publishes again. Consumers apply current state in an idempotent fold on (learner_id, course_id) and source_lsn, which lesson 16 specified. That is at-least-once publish into an idempotent fold. Calling beginTransaction around the Kafka send does not roll back PostgreSQL when the send aborts, and it does not commit PostgreSQL when the send commits. Anyone who wants a single commit across both systems is designing a distributed transaction this platform is not running. The outbox is the design that keeps the database as the system of record for "the change happened" and the log as a copy.

The object store is the same kind of boundary. commitTransaction will not delete an object you already put, and it will not put an object you only planned to write. Order the lake write so that a repeated offset replaces the same key, and commit the Kafka offset afterward, as lessons 38 and 40 did. If you also produce a derived Kafka record in the same step, decide which output is the one inside the transaction. A practical split: landing writes objects idempotently and commits offsets with commitSync; a different job, if you need a validated topic, reads the raw topic and writes learning.progress.validated transactionally. One job that tries to do the object put inside the transaction will time out or will commit Kafka without a durable object, depending on the order of mistakes. Split the jobs.

The five hops, named

The DLQ hop is a produce from the landing job. It should be idempotent on the origin offset in the header so a replay does not require five copies for an operator to ignore. A transactional inclusion of the DLQ produce together with the group offset is possible only if you are not also pretending the object put is in that transaction. The straightforward landing loop keeps the DLQ get before commitSync and uses an origin-offset header so a second publish is detectable. That stays at-least-once and operable. Move it into a transaction when the DLQ consumer cannot tolerate duplicates and you have removed the external put from the same transaction.

Fencing and the operator

ProducerFencedException and InvalidProducerEpochException mean this process is no longer the owner of the transactional.id. The correct response is to close the producer and stop the process. Recreating a producer in a loop with the same id will fence the other live owner, including the one that should be working. A crash loop that calls initTransactions on a shared id is a fencing fight. Give each member its own id, derived from a stable instance id, and alert on fencing rather than retrying it.

Aborted transactions leave markers in the log. A read_committed consumer skips the aborted records. A consumer that copies the topic with read_uncommitted into the lake will persist aborted records as if they had committed. The validated-topic landing, if you ever build one, sets read_committed. The progress landing of the player's non-transactional topic does not have aborted markers from the player. Do not set isolation levels at random on every consumer because a platform standard says "we use exactly-once." Set them on the consumers of transactional topics.

Architecture and Design Diagrams

mermaidCopy
flowchart TB
    subgraph hop1 [Player hop]
        player[Idempotent producer]
        progress[learning.progress]
        player -->|retry may append a second event_id| progress
    end

    subgraph hop2 [Lake hop]
        progress --> land[Landing commitSync after object replace]
        land --> raw[One object per offset]
    end

    subgraph hop3 [Inside Kafka]
        progress --> tx[Transactional processor]
        tx --> validated[learning.progress.validated]
        tx --> commits[Offsets in the same transaction]
        validated --> rc[Downstream read_committed]
    end

    subgraph hop4 [Database hop]
        pg[(PostgreSQL commit)]
        outbox[Outbox row]
        pg --> outbox
        outbox -->|at-least-once publish| enroll[learning.enrollment]
    end

The arrows are different protocols. A slide that merges them into one padlock is the slide this lesson replaces.

mermaidCopy
sequenceDiagram
    participant C as Consumer read_committed
    participant P as Transactional producer
    participant In as Input topic
    participant Out as Output topic
    participant Coord as Transaction coordinator

    C->>In: poll batch
    P->>Coord: beginTransaction
    P->>Out: send derived records
    P->>Coord: sendOffsetsToTransaction with group metadata
    P->>Coord: commitTransaction
    Note over Coord: stale generation aborts the whole transaction
    C->>Out: downstream sees records only after commit

If the process crashes after send and before commitTransaction, the transaction aborts or times out. read_committed consumers do not see the derived records. The input offsets are not committed in the transaction, so the batch is polled again. That is the exactly-once retry. A downstream consumer left on read_uncommitted sees the sends that later abort, which breaks the claim.

mermaidCopy
sequenceDiagram
    participant L as Landing thread
    participant Store as Object store
    participant K as Group coordinator

    L->>Store: put object key for offsets 100-102
    alt crash before commitSync
        L->>Store: restart puts the same key again
        L->>K: commitSync next offset 103
    else commitSync succeeds
        L->>K: next poll starts at 103
    end

There is no commitTransaction in this second sequence. Adding one around the put does not enlist the object store. The key is the guarantee.

Technical Explanation

Configuration for a transactional processor

A lab processor reads learning.progress and writes learning.progress.validated with the same key and the same bytes when classify would accept them, and writes nothing for rejects. Rejects stay the landing job's DLQ concern. This processor is optional in the platform. It exists so the transactional API is real, not theoretical. It is not a second landing path to the lake.

javaCopy
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.ConsumerGroupMetadata;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.clients.consumer.OffsetAndMetadata;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.common.TopicPartition;
import org.apache.kafka.common.errors.InvalidProducerEpochException;
import org.apache.kafka.common.errors.ProducerFencedException;
import org.apache.kafka.common.serialization.ByteArrayDeserializer;
import org.apache.kafka.common.serialization.ByteArraySerializer;
import org.apache.kafka.common.serialization.StringDeserializer;
import org.apache.kafka.common.serialization.StringSerializer;

import java.time.Duration;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Properties;

public final class ValidatingProcessor {
    private final KafkaConsumer<String, byte[]> consumer;
    private final KafkaProducer<String, byte[]> producer;

    public ValidatingProcessor(String bootstrap, String transactionalId) {
        Properties consumerProps = new Properties();
        consumerProps.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrap);
        consumerProps.put(ConsumerConfig.GROUP_ID_CONFIG, "learning-progress-validated");
        consumerProps.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        consumerProps.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, ByteArrayDeserializer.class.getName());
        consumerProps.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, false);
        consumerProps.put(ConsumerConfig.ISOLATION_LEVEL_CONFIG, "read_committed");
        consumerProps.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
        this.consumer = new KafkaConsumer<>(consumerProps);

        Properties producerProps = new Properties();
        producerProps.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrap);
        producerProps.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
        producerProps.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, ByteArraySerializer.class.getName());
        producerProps.put(ProducerConfig.ACKS_CONFIG, "all");
        producerProps.put(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG, true);
        producerProps.put(ProducerConfig.TRANSACTIONAL_ID_CONFIG, transactionalId);
        producerProps.put(ProducerConfig.TRANSACTION_TIMEOUT_MS_CONFIG, 15_000);
        this.producer = new KafkaProducer<>(producerProps);
        this.producer.initTransactions();
        this.consumer.subscribe(List.of("learning.progress"));
    }

    public void pollOnce() {
        ConsumerRecords<String, byte[]> records = consumer.poll(Duration.ofMillis(500));
        if (records.isEmpty()) {
            return;
        }
        producer.beginTransaction();
        try {
            Map<TopicPartition, OffsetAndMetadata> offsets = new HashMap<>();
            for (ConsumerRecord<String, byte[]> record : records) {
                if (accept(record.value())) {
                    producer.send(new ProducerRecord<>(
                            "learning.progress.validated", record.key(), record.value()));
                }
                offsets.put(
                        new TopicPartition(record.topic(), record.partition()),
                        new OffsetAndMetadata(record.offset() + 1));
            }
            ConsumerGroupMetadata metadata = consumer.groupMetadata();
            producer.sendOffsetsToTransaction(offsets, metadata);
            producer.commitTransaction();
        } catch (ProducerFencedException | InvalidProducerEpochException fenced) {
            // This process no longer owns transactional.id. abortTransaction can throw again.
            throw fenced;
        } catch (RuntimeException error) {
            producer.abortTransaction();
            throw error;
        }
    }

    private static boolean accept(byte[] value) {
        return value != null && value.length > 0;
    }
}

accept in production is the contract classifier, not a length check. The group id is learning-progress-validated, not the landing group. transactional.id is unique per instance, for example progress-validated- plus the static group.instance.id. read_committed on this consumer matters when learning.progress contains transactional writes. With a non-transactional player it is harmless. The downstream consumer of learning.progress.validated must set read_committed, and that is the setting that protects the claim.

transaction.timeout.ms of 15 seconds is inside the broker's maximum and inside one quick poll. If commitTransaction throws ProducerFencedException, do not abort and retry in a loop. Exit. The fenced producer cannot abort reliably. The process supervisor starts a new process only if it is supposed to own that id.

The offset map uses record.offset() + 1 because sendOffsetsToTransaction stores the same next-offset contract as commitSync. Committing record.offset() without the increment rereads the last record and writes it again. Inside a transaction, that duplicate is still a duplicate the downstream will see once the transaction commits. The arithmetic does not become optional because the API is transactional.

The lake hop, unchanged and named

The landing loop does not gain initTransactions. Its claim is the table row "at-least-once processing, one object per offset." A test kills the process after the put and before commitSync, restarts, and checks that the object ETag or the file bytes match a single copy of the batch. A second test publishes the same event_id twice from the player, two send calls, and checks that two offsets exist in the raw prefix. The morning model may count distinct event_id. The landing prefix must not collapse them. If a test shows one offset for two sends, the producer was retried by the client library and idempotence did its job, or the test accidentally reused one send. Log the offsets from the two callbacks. They must differ when the application called send twice.

Enrollment, unchanged and named

The publisher writes learning.enrollment after PostgreSQL commits. The cohort consumer folds source_lsn. A duplicate publish with the same LSN is a no-op in the fold. A publish that never happens is lag, and lesson 16's slot still retains WAL until the publisher confirms. Kafka transactions are not a substitute for the slot monitor. They are not a substitute for the fold. Lesson 46 can change the publisher's implementation. It should not change this sentence without a new design review.

Spark, unchanged and named

The tile's exactly-once claim is the file sink plus the checkpoint, for micro-batches that the sink commits. foreachBatch remains at-least-once unless the batch id is part of an idempotent write. Nothing in initTransactions moves the checkpoint. A proposal to "turn on Kafka transactions so the tile is exactly-once" misunderstands which system already made the claim. If the tile later writes to a Kafka topic, that write is a new hop and needs its own row in the table, including read_committed consumers of that topic. The Parquet sink does not become transactional by adjacency.

Cost and failure

Transactions add a coordinator round trip and they pin transactional.id state on the cluster. For a derived topic that must not show duplicates to a consumer you do not control, the cost is justified. For landing, the object key is cheaper and covers the system the transaction cannot see. Idempotence on the player is cheap relative to a duplicated append on every leader change, and it is already the default. Turning it off does not save money you can find on a bill. It spends duplicates into course_day if the morning job forgets to dedupe event_id.

Fencing failures are operationally loud. Page them. A quiet retry hides a split brain between two processors writing the validated topic. Aborted-transaction rate on a topic that should commit is a bug in the processor, often a timeout because work crept inside the transaction. Move the work out or shorten the batch. Do not raise transaction.timeout.ms into minutes so a lake put can sit inside it. That put is the wrong work.

Practical Implementation or Design Walkthrough

Problem

Prove three claims in a lab, with the official client, against learning.progress and a derived topic you create with the same replication policy as lesson 39.

1. Two application-level send calls with one event_id produce two offsets even though idempotence is on.
2. Killing landing after the object put and before commitSync leaves one object for that offset range.
3. A transactional processor's aborted transaction is invisible to a read_committed consumer and visible to a read_uncommitted consumer.

Steps

Create learning.progress.validated with replication factor 3, min.insync.replicas 2, and delete cleanup. Do not point the morning job at it.

For claim 1, call publish twice with the same bytes and the same event_id, wait for both callbacks, and store both offsets. Fetch those offsets with a console consumer or a short assign plus seek that does not use the landing group id. You should see two records. Then induce a duplicate the idempotent producer should collapse: for example, rely on a retried batch under fault injection if your lab can drop a response, or read the producer metric for record retries and confirm the log end did not advance by the retry count. If you cannot inject a fault, write the sentence you would expect and do not claim you demonstrated it. The two-send result is the one this bus must not misunderstand.

For claim 2, use the landing loop from lesson 38. Kill after fsync. Restart. Compare bytes.

For claim 3, run ValidatingProcessor with a test hook that calls abortTransaction instead of commitTransaction after send. Consume the output topic twice, once with isolation.level=read_committed and once with read_uncommitted, using two throwaway group ids. The uncommitted consumer can see the aborted records. The committed consumer does not. Then run a committing pass and show the committed consumer receives the records once. Reset those throwaway groups rather than the landing group.

What failure looks like

ProducerFencedException during the lab usually means two processors started with the same transactional.id. Stop one. InvalidProducerEpochException is the same family. TimeoutException on commit means the transaction exceeded transaction.timeout.ms or the coordinator was unreachable. Abort if you still own the epoch, and poll again. Do not commit the consumer offsets on the side with commitSync after a failed transaction. The transactional commit is the only offset commit for this group. A side commitSync publishes a position without the output records, which is at-most-once on the derived topic. The code above does not call consumer.commitSync. Keep it that way.

Expected evidence

The two offsets for one event_id, the single object after the kill, and a note of which isolation level saw the aborted batch. A paragraph that maps each evidence item to a row in the hop table. The enrollment publisher is out of scope for the lab and in scope for the paragraph: state the outbox claim you did not pretend to test.

Enterprise Real-World Scenarios

Scenario 1: Finance asks the morning job to count events, and the player retries

A flaky hour on the edge loses HTTP responses. The player retries with the same event_id. Idempotence is on. Callbacks succeed twice because send was called twice. The raw prefix has two offsets. Finance's first draft of course_day counts lines. The count is high by the retry rate. The tile, counting events in a window, is high for the same reason. Neither bug is a broker duplicate. The broker did what two acknowledged sends ask it to do.

The model change is to count distinct event_id where the product means observations, and to keep the duplicate lines in raw so the retry rate is measurable. The player change is to retry the same in-flight request, not to mint a new event_id per click of a spinner. Security is unchanged: retries are still the player's principal. Cost is a slightly larger raw prefix on bad days, which is cheaper than a silent under-count from a producer that drops on timeout. The incident write-up uses the hop table. It does not say "exactly-once failed."

Scenario 2: A validated topic for a team that cannot dedupe

A recommendations job wants progress events and will double-count if it sees duplicates. Its owners will not implement an offset-named sink. They will consume a topic. The platform offers learning.progress.validated, produced by ValidatingProcessor, with their group on read_committed. Their lag is their group, not the landing group. The lake remains the landing hop, unchanged, because the recommendations job is not the morning table.

The transactional id is per instance. A deploy that starts the new instance before the old one exits will fence one of them. The runbook starts the new process only after the old process has closed, or uses static membership and a restart that reuses one id after the previous process is dead. A fencing alert pages the processor owner, not the player team. The recommendations consumer sets isolation.level=read_committed in the same pull request that creates the group. Forgetting it means they consume aborted attempts during the first bad deploy and the exactly-once claim is false for them specifically. The platform review checks the consumer config, not only the producer.

Cost is another topic, another group, and transaction coordinator load proportional to the poll rate. It is justified for a consumer that cannot dedupe. It is not justified as a second copy of landing.

Scenario 3: Enrollment double-publish and a cohort that stays correct

The outbox publisher crashes after Kafka acknowledges an enrollment_changed delete and before it marks the outbox row. On restart it sends the same source_lsn again. Two records sit on learning.enrollment. The cohort fold keeps the greater LSN per key and treats an equal LSN as already applied. The seat count does not drop twice. A Kafka transaction around only the second send would not have removed the need for the fold, because the first send was already committed in Kafka before the crash, and PostgreSQL had committed before the first send. The transaction boundary people wished for would have included the database, the outbox mark, and the log. That boundary is not on offer. The fold is the boundary. The slot monitor remains, because a publisher that never restarts does not double-publish; it stalls, and WAL accumulates. Exactly-once language in the enrollment ticket gets rewritten to this paragraph before the ticket is accepted.

Best Practices

• Write the hop, the crash, and the claim. Reject a slide that says exactly-once without them.
• Leave producer idempotence on. Do not describe it as dedupe of event_id.
• Use transactions for Kafka-to-Kafka pipes whose consumers set read_committed. Commit offsets only with sendOffsetsToTransaction in that pipe.
• Keep lake landing on offset-named objects and commitSync. Do not open a transaction around the put.
• Give each transactional instance its own transactional.id. On ProducerFencedException, exit.
• Keep transactions shorter than transaction.timeout.ms, and keep external I/O out of them.
• Deduplicate the morning model on event_id when the product counts observations. Retain both offsets in raw.
• Fold enrollment on source_lsn. Do not ask a Kafka transaction to commit PostgreSQL.
• Leave the Spark file-sink claim where lesson 32 put it. A group commit does not improve it.
• Test the kill before commit, the double send, and an aborted transaction under both isolation levels.

Common Mistakes and Anti-Patterns

"Idempotent producer means exactly-once end to end." It means retried batches in one producer session are not appended twice. A second send, a second process, and a non-idempotent sink are all outside that sentence. The hop table is the correction.

Wrapping the lake put in beginTransaction. The object store does not join the transaction. You will time out, or you will commit Kafka while the put is still ambiguous. Put first, with a stable key, then commit the offset.

commitSync plus a transactional send. The offset can commit when the transaction aborts, or the reverse if you order them independently. One path. For the derived topic, only sendOffsetsToTransaction.

A downstream consumer on read_uncommitted. Aborted transactions appear as data. The producer did the work and the consumer undid the claim. Set read_committed on every reader of a transactional topic.

Sharing one transactional.id across two live members. They fence each other. The validated topic stalls or flops. One owner per id.

Retrying forever after ProducerFencedException. Each initTransactions may fence the rightful owner. Stop the process. Page a human.

Collapsing two event_id copies in landing. You hide player retries and make the duplicate rate unmeasurable. Land both. Dedupe in the model that defines an observation.

Using a transaction to avoid the enrollment fold. The database commit and the log append are not one transaction. The fold on LSN is what makes the second publish safe. Removing the fold because "we enabled transactions" double-applies or drops, depending on which side you forgot.

Turning off idempotence to "simplify." You reintroduce duplicates on every ambiguous produce timeout. The default is the simple setting. Leave it.

Troubleshooting and Design Review

Symptom: Distinct event_id in raw is lower than the line count by a few percent after a flaky edge hour. Broker retry metrics are low. Player retry metrics are high.

Likely cause: application-level resend, two send calls, idempotence working as designed and not deduping them.

Investigation: pairs of offsets with the same event_id and increasing producer timestamps. Callback logs that recorded two metadata offsets.

Corrective action: count distinct event_id in course_day if that is the product. Keep the lines. Fix the player if it mints a new event_id per retry when it should reuse one. Do not disable idempotence.

Symptom: course_day is high versus distinct event_id, and producer retry metrics are high, and the two lines share an event_id but the client insists it called send once.

Likely cause: idempotence is off, and a timed-out batch was appended and then retried.

Investigation: the running producer's enable.idempotence and acks. A false flag with acks=1 is the misconfiguration to find.

Corrective action: turn idempotence on and set acks=all. Rebuild the affected morning partitions from raw after you decide whether those pairs are one observation. They are one player action and two log records. The model should match the product, and the producer should stop creating new pairs.

Symptom: The validated topic's consumer sees records that later vanish from a read_committed recount, or the counts flicker during deploys.

Likely cause: the consumer is read_uncommitted and is counting aborted transactions.

Investigation: the consumer configuration's isolation.level, and the processor's abort rate.

Corrective action: set read_committed and reset the consumer group to a known offset if it persisted bad counts downstream. Fix the abort cause, often a timeout, rather than lengthening the transaction to cover I/O.

Symptom: ProducerFencedException on both instances of the validating processor.

Likely cause: one transactional.id in a shared config map.

Investigation: the id in each process's logs at initTransactions.

Corrective action: stop both, assign distinct ids, start one, confirm commits, start the other only if it owns a different id. Check the output topic for a gap or a duplicate around the fight, using a read_committed consumer from a throwaway group. Repair by reprocessing from the last good input offset if the output is wrong. Do not continue the fight.

Symptom: Landing objects are correct, and a new transactional landing attempt duplicated an hour into a second prefix.

Likely cause: someone began a transaction, put an object with a non-idempotent key, and also committed offsets transactionally, then retried the put on abort.

Investigation: object keys. If they do not contain the offset, the transaction was never the bug. The key was.

Corrective action: return to commitSync after an offset-named put. Delete or replace the duplicated prefix. Keep transactions on the validated topic only.

A design review asks, for each hop: what is the crash, what is repeated, what is dropped, and which system is inside the claim? Is idempotence on, and is event_id still required? Does any transaction include an external put or a database commit? Do all consumers of a transactional topic set read_committed? What is the transactional.id scheme? What does the morning job count? What does the cohort fold on? A review that accepts the phrase "exactly-once Kafka" has not started.

Interview Questions and Answers

Q1. Define the three semantics, and give each a hop on this bus.

At-most-once can lose and will not repeat. I do not use it for progress. At-least-once can repeat and will not lose a success. The player hop is at-least-once per event_id, and the lake hop is at-least-once processing. Exactly-once neither loses nor repeats inside a boundary. The tile's file sink is exactly-once for Spark's committed batches. A transactional copy from learning.progress to learning.progress.validated is exactly-once inside Kafka when consumers read committed. I will not use one of these words without the hop.

Q2. What does the idempotent producer actually suppress?

It suppresses a second append of a batch the broker already applied for this producer id and sequence, which is what a timed-out retry would otherwise do. It requires acks=all and a bounded number of in-flight requests. It does not suppress a second send from application code, even with the same event_id and the same bytes. It does not survive as the same producer id across a process restart unless you move to a transactional id, which is a different feature. I still put event_id in the payload.

Q3. Why is landing not a Kafka transaction?

The effect that must be exactly once is an object in a store Kafka's transaction coordinator does not write. commitTransaction will not roll back a put and will not create one. I put an object whose key is the topic, partition, and offset range, then commitSync the next offset. A crash between them repeats the put and replaces the key. That is at-least-once processing with an idempotent sink. A transaction wrapped around the put adds coordinator state without adding the object store to the commit.

Q4. What has to be true for exactly-once between two topics?

The producer sets a unique transactional.id, calls initTransactions, and puts the output sends and sendOffsetsToTransaction with the consumer's ConsumerGroupMetadata in one transaction. The input consumer does not also commitSync. Downstream consumers set isolation.level=read_committed. A stale group generation fails the commit, which fences a zombie member's output. Transactions stay shorter than transaction.timeout.ms. If any of those is missing, I do not claim exactly-once for that pipe.

Q5. What is ProducerFencedException, and why is a retry dangerous?

initTransactions on a transactional.id bumps the epoch and fences the previous owner. The old process receives ProducerFencedException on its next transactional call. It should exit. If it starts again with the same id while the new owner is alive, it fences the new owner. Two supervisors restarting each other will take turns aborting the validated topic. I page, I stop one process, and I assign distinct ids to members that are supposed to run concurrently.

Q6. Why must the offset passed to sendOffsetsToTransaction be record.offset() + 1?

Committed positions are the next record to read, the same contract as commitSync. Storing record.offset() asks for that record again. The transaction would commit, downstream would see the derived record, and the next poll would derive it again. The transactional API does not change the meaning of an offset. I use the no-argument commitSync on the non-transactional landing path so the client computes the next offset, and I am explicit with + 1 when I build the map for sendOffsetsToTransaction.

Q7. What does read_committed change, and when does it change nothing?

It hides records that belong to aborted or open transactions, and it returns them only after commit. read_uncommitted returns them as soon as they are fetched, including ones a later abort should erase. On a topic where every producer is non-transactional, there are no transaction markers, and both settings return the same records. I set read_committed on readers of learning.progress.validated. I do not set it on the player topic's landing consumer in the hope that it will dedupe event_id. It will not.

Q8. How can the morning count be wrong if every hop is "working"?

The player successfully sent two records for one event_id. Landing successfully stored both. The tile successfully counted both. The broker successfully ran an idempotent producer. The model counted lines. The product wanted observations. The fix is distinct event_id in the model, not a change to acks. I would show the two offsets in the review so "working" gets a definition.

Q9. Why doesn't a Kafka transaction solve the enrollment publisher?

PostgreSQL commits the enrollment change before the publisher runs. The publisher then appends to the log and marks an outbox row. Those are three commits in two systems. KafkaProducer does not take the database connection into commitTransaction. A crash between append and mark sends the change twice. The cohort consumer treats an equal source_lsn as already applied. A publisher that dies before the first append stalls, and the replication slot retains WAL. I monitor the slot and the group lag. I do not announce exactly-once CDC because a transactional id was configured on the publisher.

Q10. Where does Spark's exactly-once claim live, relative to this lesson?

It lives in the checkpoint and the file sink, for the micro-batch the sink committed. It does not live in learning-progress-raw-landing's commits, and it does not move when we reset that group. foreachBatch is still at-least-once unless the batch id makes the write idempotent. If a future tile writes to Kafka, that output is a new hop: either at-least-once into a topic whose consumers dedupe, or a transactional write with read_committed readers. I would add a row to the hop table. I would not stretch the Parquet sentence to cover it.

Hands-On Exercises

Exercise: Three claims, three logs

Problem statement. Demonstrate application-level duplicates, an idempotent lake retry, and an aborted transaction under both isolation levels. Map each result to the hop table.

Tasks.

1. With idempotence enabled, send the same event_id twice. Record both offsets and show both records without using the landing group id.
2. Run the landing loop, kill it after the durable put and before commitSync, restart, and show a single object for that range.
3. Create learning.progress.validated with the lesson 39 durability settings. Run a transactional processor that aborts once. Compare read_committed and read_uncommitted consumers on throwaway group ids. Then commit a batch and show a single visible result to the committed consumer.
4. Confirm the processor never calls consumer.commitSync, and confirm its group.id is not the landing group.
5. Force or describe ProducerFencedException with two processes and one transactional.id. State the operator response.
6. Write the enrollment hop in the same table, including the LSN fold, and state that this lab did not test PostgreSQL.
7. State what course_day should count, lines or distinct event_id, and why the raw prefix keeps both.

Expected deliverables. Offset evidence, the object listing, the two isolation outputs, and the completed hop table.

Evaluation rubric.

You pass when the hop table and the lab evidence use the same verbs, and when no step commits the landing group from the transactional processor.

Lesson Summary

Delivery semantics name a hop and a crash. The player is at-least-once per event_id: idempotence removes duplicate appends from broker retries and does not remove a second send. Landing is at-least-once into an object key that makes a repeated offset a replace, with commitSync after the put. The tile's file sink remains the Spark checkpoint's claim. A derived topic can be exactly-once inside Kafka when a transactional producer commits output and input offsets together and every reader uses read_committed. transactional.id fences zombies and must not be shared by two live owners. Transactions do not enlist the object store or PostgreSQL. Enrollment stays an at-least-once publish into a fold on source_lsn. The morning model counts distinct event_id when the product means observations, and the raw log keeps every offset so retries stay visible.

Kafka Connect and Data Integration is the next lesson, for moving these hops with connectors instead of the hand-written landing loop. The protocol primer is Transactions and Exactly Once Processing.

Further Reading

• Producer configurations — enable.idempotence, transactional.id, transaction.timeout.ms, and the constraints idempotence places on acks and in-flight requests.
• Consumer configurations — isolation.level values read_uncommitted and read_committed.
• KafkaProducer Javadoc — initTransactions, beginTransaction, sendOffsetsToTransaction, commitTransaction, abortTransaction, and fencing exceptions.
• Transactions in the Kafka documentation — the project's own scope for exactly-once. Match that scope in design reviews.
• Idempotent Producers, Delivery Guarantees, and Duplicate Events — the specialty primer for the failure catalog behind the hop table.
• Change Data Capture Fundamentals — the database side of the enrollment hop this lesson refuses to absorb into a Kafka transaction.