In this lab, you trigger failures to observe each delivery guarantee, then add protection where the application needs stronger results. A consumer that commits before it processes loses an order. A consumer that processes before it commits counts an order twice. You repair the second with a processed-events marker, prevent supported producer-retry duplicates with idempotence, and build a transactional read-process-write processor whose aborted output stays invisible to read_committed readers. The last phase shows where the guarantee stops: a simulated email entry written inside the processing loop appears twice in a local file.

The contract is fixed. orders.placed has one partition. On this fresh non-transactional input topic, the twenty records occupy offsets 0 through 19. Twenty OrderPlaced events, evt-7001 through evt-7020, carry amountMinor equal to 1000 times their sequence number, with one orderDate and one currency. A correct daily_sales total is 210000. Losing event 10 gives 200000; counting it twice gives 220000. The early phases crash at evt-7010; Phase 6 uses evt-7022.

Environment: E1, the single broker kafka-local, plus E3, the Java client project, plus a PostgreSQL 17 container named postgres-local. Docker runs Kafka and PostgreSQL. Kafka tools run inside kafka-local through docker exec, and psql runs inside postgres-local. Java 21, Maven, and your three programs run on the host against localhost:9092 and localhost:5432. Open four terminals: Terminal A for Java programs, Terminal B for a second Java instance or console consumer, Terminal C for Kafka tools and Docker, and Terminal D for psql.

The image's single-node configuration sets transaction.state.log.replication.factor and transaction.state.log.min.isr to 1, which Phase 4 needs. Confirm it after the first transaction by describing __transaction_state with kafka-topics.sh --describe; the replication factor shown is 1.

Start from the consumer project the consumers module set up, and copy it to kafka-labs/exactly-once/:

Add one dependency to pom.xml, next to kafka-clients:

Build with mvn compile dependency:copy-dependencies -DoutputDirectory=target/lib and run each program with java -cp 'target/classes:target/lib/*' com.example.kafka.<Class>. System properties such as -Dlab.crashAt=evt-7010 go before -cp.

Start a fresh broker and the database from Terminal C:

Create both topics with one partition each:

Save this as sql/schema.sql and load it with docker exec -i postgres-local psql -U postgres -d ordersdb < sql/schema.sql:

Write LabProducer. For each sequence number n from 1 through 20, it sends a record to orders.placed with key "ord-" + (1041 + n). Construct the JSON value with eventId = "evt-" + (7000 + n), that same orderId, numeric amountMinor = 1000 * n, eventType = "OrderPlaced", orderDate = "2026-09-12", and currency = "INR". Serialize these fields as valid JSON. It waits on each send() future, sleeps 500 ms between sends, and prints the confirmed offset. Configure acks=all, request.timeout.ms=5000, delivery.timeout.ms=60000, and StringSerializer for both. Two system properties control it: lab.idempotence sets enable.idempotence, default true, and lab.count sets the last event number, default 20; a value above 20 sends only that one event. Run it once now and check that the last line reports offset 19.

Write LabConsumer. It takes the mode as its argument and subscribes to orders.placed in the group named by the lab.group property, default order-reporting, with enable.auto.commit=false, auto.offset.reset=earliest, and max.poll.records=1, so each nonempty poll returns at most one record and each batch commit covers that record. For each record it extracts eventId from the JSON and follows the mode:

crashIf calls Runtime.getRuntime().halt(1) when eventId equals the lab.crashAt property. addToDailySales runs one autocommit statement over JDBC at jdbc:postgresql://localhost:5432/ordersdb, user postgres, password lab:

Run the first variant in Terminal A:

After the process dies, describe the group from Terminal C and read the total from Terminal D:

Restart the consumer without -Dlab.crashAt. The replacement may wait up to the session timeout for the partition, because the crashed member never left the group. Wait until CURRENT-OFFSET reaches 20 and lag reaches zero, then record the description and the total in observations/01-commit-first.txt.

What you should see: after the crash, CURRENT-OFFSET is 10 and the total is 45000, the sum of events 1 through 9. After the restart, CURRENT-OFFSET is 20 and the total is 200000. Event 10 is still at offset 9 in the log, but the group committed past it before the update ran. This is the at-most-once window of the commit-before-processing sequence.

Stop the consumer with Ctrl-C. Wait until the group has no active members, then reset for the second variant:

Repeat the crash and restart with mode process-first, and save the evidence as observations/01-process-first.txt.

What you should see: after the crash, CURRENT-OFFSET is 9 and the total is 55000; the database holds event 10 but Kafka does not know. After the restart, the total is 220000. The replacement resumed at offset 9 and added 10000 again. This is the duplicate processing window of the process-before-committing sequence.

applyOnce runs both statements on one connection with setAutoCommit(false), then commits:

The marker statement is INSERT INTO processed_events (consumer_name, event_id, event_payload) VALUES ('order-reporting', ?, ?::jsonb) ON CONFLICT DO NOTHING. The pause sits between the marker insert and the commit so the race later in this phase can hold a claim open.

Stop all affected consumers and wait until their groups are empty. Reset the relevant group offsets to 0, then truncate both tables. Run mode idempotent with -Dlab.crashAt=evt-7010, restart without it, and collect the group description, the total, and:

Save everything as observations/02-idempotent.txt.

What you should see: the crash leaves CURRENT-OFFSET at 9, exactly as in the process-first run. After the restart, the consumer prints evt-7010 duplicate once, then applied for events 11 through 20. The total is 210000 and processed_events has twenty rows, one for evt-7010. The consumer read the record twice; its effect happened once.

Now race two instances. Stop the consumers and wait until order-reporting is empty, then reset it to offset 0 and truncate both tables. The second instance uses order-fulfillment; if that group already exists from an earlier attempt, stop its members, wait until it is empty, and reset it to offset 0 too. Start one instance in Terminal A with -Dlab.pauseMs=3000 and mode idempotent, and a second in Terminal B with the same command plus -Dlab.group=order-fulfillment. Both groups read every record, so both instances reach the database with the same event, which is what an old partition owner and its replacement look like at the destination.

What you should see: for each event, one terminal prints applied and the other prints duplicate. When the inserts overlap, the second waits for the first transaction. Log timestamps around each insert and commit to measure the wait; the exact delay depends on when the inserts start. The total is 210000 and processed_events has twenty rows. Record which instance won each event in observations/02-race.txt; the split is timing-dependent, the totals are not.

Stop both Phase 2 instances with Ctrl-C, then delete and recreate orders.placed so counts start from zero:

Recreate it with the Setup command. In Terminal A, run LabProducer with -Dlab.idempotence=false. After it confirms three or four offsets, pause the broker from Terminal C, wait ten seconds, and unpause:

If a request remains unresolved beyond request.timeout.ms, the client can retry while its delivery budget remains. Pausing the broker does not guarantee that it will append both an original attempt and a retry. Record successful sends, errors, and retry evidence along with the counts. Count records per event ID:

Delete and recreate the topic again, run LabProducer with idempotence at its default, repeat the pause at the same point, and count again. Save both counts as observations/03-producer.txt.

What to compare: with idempotence off, an event ID can appear more than once if the broker appends both the original attempt and a retry. A run with no duplicates is also valid; record that result and explain why it does not establish a guarantee. To repeat either experiment, delete and recreate the input topic before running the application again. Reusing event IDs in a new application send can create duplicates even with idempotence enabled.

With idempotence on, a successful run on the fresh topic contains one copy of each of the twenty event IDs and ends at offset 19. The leader recognizes supported internal retries from producer identity and sequence information. If the producer reports a failure, preserve and diagnose it before repeating with a fresh topic. Proceed to Phase 4 only after confirming exactly twenty input records, one for each expected ID.

Use the fresh group order-transaction-lab for this phase. If you repeat the phase, stop its members, wait until the group is empty, and reset its input offsets to 0 before starting. For a repeat run, also stop the output readers, delete and recreate orders.events with the Setup settings, and clear emails.log so the observation counts start from zero. Write OrderProcessor. Its consumer joins order-transaction-lab with enable.auto.commit=false, isolation.level=read_committed, auto.offset.reset=earliest, and max.poll.records=1. Its producer sets transactional.id from the first argument, enable.idempotence=true, and acks=all. Skip empty polls. With max.poll.records=1, each nonempty poll becomes one transaction:

accepted parses the input JSON, preserves its actual eventId and orderId, changes eventType to OrderAccepted, and serializes those fields as valid JSON. The flush() forces the output record into the log before the crash point, so the crash lands between an appended record and its commit. Before the loop, main calls producer.initTransactions(); if the property lab.initOnly is true, it stops there and sleeps instead of consuming.

Watch the two reader terminals diverge at evt-7010:

Start the read_uncommitted reader in Terminal A:

Start the same command in Terminal B with --command-property isolation.level=read_committed added. Run the processor from Terminal C with -Dlab.crashAt=evt-7010 and argument order-processor-0. After it dies, describe order-transaction-lab. Restart the processor without the crash property, wait until the input group reaches CURRENT-OFFSET=20 and the committed reader has received twenty records, and describe the group again. Save both readers' output and both descriptions as observations/04-isolation.txt.

What you should see: after the crash, Terminal A has printed ten records, including evt-7010 from the unresolved transaction, and Terminal B has printed nine and stopped at the last stable offset. CURRENT-OFFSET for order-transaction-lab is 9; the offsets sent to the transaction did not commit. On restart, initTransactions() resolves the open transaction as aborted, the processor reprocesses offset 9, and Terminal B prints evt-7010 once. Terminal A prints it twice, at two offsets. Both terminals skip control-marker offsets. Only the read_committed reader also skips the aborted record’s offset. The final CURRENT-OFFSET is 20, and Terminal B has printed exactly twenty records. If you wait longer than transaction.timeout.ms before restarting, the coordinator aborts the transaction itself; the outcome is the same.

Leave the Phase 4 processor running and idle. From Terminal D, start a second instance with the same transactional ID:

Publish one more event by running LabProducer with -Dlab.count=21. Record what the first processor prints, stop the init-only instance, restart the real processor, and save the evidence as observations/05-fencing.txt.

What you should see: the first processor fails with ProducerFencedException on its first transactional call after the second instance initializes. Record the method that reports the error; it may surface while beginning the transaction, sending output, submitting offsets, or committing. It never commits evt-7021. The restarted processor initializes with a newer epoch, processes offset 20 once, and the read_committed reader prints evt-7021 once. Two live producers with one transactional ID cannot both commit; the newer initialization wins.

Stop the processor first. Restart it with -Dlab.crashAt=evt-7022, then publish event 22 with LabProducer and -Dlab.count=22. Let the processor crash, then restart it without the crash flag. Count the evidence:

Count evt-7022 in the read_committed terminal and describe the group. Save everything as observations/06-boundary.txt.

What you should see: emails.log has two lines for evt-7022 and two for evt-7010, one per attempt. The read_committed reader shows one evt-7022 record, and CURRENT-OFFSET is 22. The Kafka output and the offsets committed once; the file append is outside the transaction, and the abort did not remove it. Exactly-once semantics ends at the last operation Kafka can include in the commit.

Stop every Java program and console consumer with Ctrl-C, then remove both containers:

Verify with docker ps -a --filter name=kafka-local and docker ps -a --filter name=postgres-local; both listings are empty. If Phase 3 ended with the broker paused, docker stop still works. Keep kafka-labs/exactly-once/; it is your submission.

The lost and duplicated database contributions came from committing Kafka progress separately from database work. Producer retries and fencing demonstrate different protocol boundaries. Committing the offset first exposed the at-most-once window and dropped 10000 from the total; updating the database first exposed the duplicate processing window and added it twice. The processed-events marker made retries within the second window safe by letting the database recognize its own earlier work, and the unique constraint made that recognition hold when two instances arrived at once. Producer idempotence closed a different window, the retry of a send whose acknowledgment never arrived, using sequence numbers rather than event IDs.

Transactions moved the output record and the input offset into one outcome, and the two console consumers showed what that means: the aborted attempt existed in the log, read_uncommitted printed it, and read_committed waited at the last stable offset and never did. Fencing kept a stale producer from committing alongside its replacement. The email file marked the edge of the guarantee. Kafka committed once, the offsets moved once, and the side effect outside the transaction still happened on every attempt.