In this lab, you will build a keyed JSON producer for OrderPlaced events, measure how batching settings change the number of Produce requests, break delivery by stopping and pausing the broker, and show what idempotence does and does not protect against. You will finish by routing one hot key to a dedicated partition with a custom partitioner.

Every run publishes the same workload: 1,000 events with ids evt-7001 through evt-8000, spread over 50 order ids ord-1000 through ord-1049, so ord-1042 receives exactly 20 events. Each run prints confirmed and failed counts and a few producer metrics. Keeping the workload fixed helps isolate configuration and failure effects. Repeat runs to account for scheduling, broker state, and measurement variation.

The broker is environment E1: the single kafka-local container running apache/kafka:4.3.1, reachable from the host at localhost:9092. Every Kafka command-line tool runs inside that container through docker exec. The Java code is environment E3 and runs on the host with Java 21 and Maven. This is the first lab in this module that needs Java, so start from the kafka-producer-demo project from the building-a-producer chapter.

Copy the producer project and rename it:

Run the remaining project commands from kafka-labs/reliable-order-producer/. Create observations/ before saving results. Update this project’s pom.xml:

Start the broker and create the first topic with three partitions. If the kafka-local container exists but is stopped, start it:

If no kafka-local container exists (the earlier labs remove it during cleanup), create it:

Later phases need fresh topics so earlier runs do not affect the counts. Create orders.placed.timeout, orders.placed.dup-off, orders.placed.dup-on, and orders.placed.hot with the same command, changing only the name.

Save this as OrderProducerLab.java. The first argument is the topic. Every further key=value argument becomes a producer property, except lab.pace.ms, which sleeps between sends so that you have time to break the broker mid-run.

The callback normally runs on the producer's I/O thread, so it only updates counters and maps. The loop also records exceptions that send() throws immediately in immediateErrors. This finite experiment attempts all 1,000 events, checks that every attempt has an outcome, and exits with an error if any send failed. A long-running service should stop or apply backpressure during a persistent failure instead of continuing indefinitely. The program reads metrics after flush() and before closing the producer. Run the baseline:

Save the last lines of output as observations/phase-1.txt.

What you should see: confirmed=1000, empty errors and immediateErrors maps, three entries in lastOffsets whose values plus one sum to 1,000 on a fresh topic, ord-1042 mapped to a single partition, and splitKeys=0. Record which partition ord-1042 used; Phase 6 changes it.

Run the same workload three times with a 1 ms pace so that records arrive one at a time rather than in one burst:

For each run, record request-total, batch-size-avg, and records-per-request-avg in observations/phase-2.txt.

What to compare: Increasing linger.ms from 0 to 50 gives records more time to share a batch. Look for fewer requests and larger average batches, and record the measured change. Scheduling and request overhead affect these metrics, so they need not match exact record counts or change by a fixed amount. At this pace, batches are likely to stay below 16 KiB, so raising batch.size to 64 KiB may have little effect. Explain your observed results rather than treating that expectation as a required outcome.

Add OrderPlaced.java and OrderPlacedSerializer.java exactly as written in the serializers chapter: a six-field Java record, and a Serializer<OrderPlaced> whose serialize() returns null for a null event, otherwise ObjectMapper.writeValueAsBytes(event), wrapping any JsonProcessingException in a SerializationException.

Change the value serializer to OrderPlacedSerializer.class and the producer type to KafkaProducer<String, OrderPlaced>. Replace the loop's String value = json(...) with:

The ProducerRecord infers the new value type. Delete json(). Create a fresh three-partition topic named orders.placed.serialized using the setup command, run the producer against it, and inspect one record:

Compare the JSON fields with the original fixture. The fresh topic keeps this check independent of earlier runs.

To see what a consumer receives when the bytes are not JSON, write WireFormatProbe.java: a main method that builds a KafkaProducer<String, Long> with StringSerializer for keys and LongSerializer for values, sends key ord-1042 and value 129900L to orders.placed, calls get() on the future, and prints partition() and offset() from the RecordMetadata.

Read that one record with the console consumer, substituting the printed partition and offset:

Run it twice more: once adding --formatter-property value.deserializer=org.apache.kafka.common.serialization.LongDeserializer, and once with that same deserializer but pointing at offset 0 of the same partition, which holds a JSON record.

What you should see: the default string decoding prints the eight bytes of the long as unreadable characters, because StringDeserializer treats any bytes as UTF-8 text. The LongDeserializer run prints 129900. The third run fails with a SerializationException, because a JSON document is not eight bytes long. Save all three outputs in observations/phase-3.txt.

Use a short request timeout and delivery budget, and a 20 ms pace so the run lasts about 20 seconds. Leave retries at its default; the delivery timeout bounds recovery.

About five seconds into the run, stop the broker from a second terminal, wait at least 20 seconds, then start it again:

The producer may continue accepting sends using cached metadata while buffer space remains. If buffers fill or usable metadata becomes unavailable, send() can block or fail. Batches the producer creates during the outage age in the accumulator. Any batch older than delivery.timeout.ms fails, and the callback receives org.apache.kafka.common.errors.TimeoutException. The producer can retry batches whose delivery budget has not expired when the broker returns; success still depends on recovery conditions.

Repeat the run, but start the broker immediately after stopping it, so the outage is only the restart time:

What you should see: after a completed loop, confirmed plus the counts in errors and immediateErrors equals 1,000. Failed sends make Maven report an unsuccessful execution after printing the summary. Producer error/retry metrics describe client activity and need not equal the total application error count. A shorter outage may let all attempts succeed, but measure that outcome instead of assuming it. If records still expire, repeat with a larger delivery budget and record the settings and result. Use a fresh topic for each comparison when inspecting stored record counts. Save both summaries in observations/phase-4.txt.

A paused broker differs from a stopped one: the kernel still accepts bytes into the socket buffer, but the process cannot respond. A request timeout can cause a retry. If the broker accepts the original attempt and the retry, it may append duplicates without idempotence. Pausing alone does not guarantee that both attempts reach the append stage. With idempotence it recognises the producer sequence on the retry and appends once.

Run with idempotence off and five in-flight requests:

While it runs, pause the broker for about five seconds, twice:

Repeat against orders.placed.dup-on with enable.idempotence=true and the same pauses. Dump each topic to a file:

Count event ids that appear more than once:

Check submission order per key. With fixed routing, each key belongs to one partition and the output follows log order. Internal retries without idempotence can make event numbers decrease; the following command reports any such case:

Run both checks on dup-on.txt as well.

What you should see: record confirmed sends, both error maps, retry metrics, record count, duplicate event IDs, and any decreases in event number. Without idempotence, a retry may create duplicates or change submission order; a particular pause can also produce neither. With idempotence, successful internal retries preserve per-partition submission order and avoid adding a second copy of the same send. A fresh topic with 1,000 confirmed sends and no other writes should contain 1,000 records. Failed or uncertain sends need separate interpretation. For another attempt, create fresh topics such as orders.placed.dup-off-2 and orders.placed.dup-on-2 with the setup command and use those names in the producer and dump commands. Never mix a second application run into the first run's counts. Save the results and pause timings in observations/phase-5.txt. A run with zero duplicates while idempotence is off is a valid observation, not a reason to manufacture duplicates.

Save HotKeyPartitioner.java. The name illustrates a possible use for custom routing; this fixture gives every key the same 20 events. It reserves the last partition for ord-1042 and hashes every other key over the remaining partitions with the same Murmur2 calculation the built-in routing uses:

The partitioner requires at least two partitions and rejects null keys. Select it through partitioner.class:

Verify with the console consumer on partition 2 only, reading from offset 0 with --max-messages 20, print.partition=true, and print.key=true.

What you should see: the summary shows ord-1042=[2] and splitKeys=0. All 20 records on partition 2 carry key ord-1042, and the lastOffsets entry for partition 2 is 19 on a fresh topic. Compare with the partition ord-1042 used in Phase 1: the key bytes did not change, only the routing rule. Save the output as observations/phase-6.txt.

Stop the broker and remove the container and its data:

Verify that nothing is left:

The output should contain only the header line. Keep the project directory and observations for your report.

You produced the same 1,000 keyed events under different producer settings and let the evidence show the differences. Callback metadata proved that key hashing keeps an order on one partition. The metrics showed that request count depends on arrival rate, routing, batch capacity, and waiting time. Your comparisons show which settings affected this workload. Swapping the serializer changed the value bytes, but the broker still treated them as bytes, so a mismatched deserializer failed only on the consumer side.

The failure phases let you distinguish expired sends, successful retries, and possible duplicate appends when retry protection is off. The observed outcome depends on when requests reach the broker. Idempotence gives the broker sequence information to recognize supported internal retries; it does not remove earlier records. With one replica, acks=all confirms only that single copy. A second send() of the same event remains a new operation that no producer setting deduplicates.