Producers: Beginner

A producer is a library object inside your own program, built from a handful of settings and kept open for the life of the process. On startup it connects to one of the brokers listed in bootstrap.servers, asks it for metadata, and learns from the reply every partition of every topic it writes to and which broker currently leads each one. From then on it talks to leaders directly: a record bound for partition 2 of trip-pings travels to the broker that leads partition 2, with nothing in between. If a leader moves, the next metadata refresh tells the producer where.

Only bootstrap.servers and the 2 serializer lines are required. The broker list is a starting point, so any live broker from it is enough to learn the whole cluster. The serializers name how the key and value are turned into bytes, because a broker stores bytes and nothing else; StringSerializer writes the UTF-8 bytes of a string, and a later lesson covers the other formats. client.id is a label the brokers attach to this producer's requests in their logs and metrics, so a busy producer can be found by name.

The command-line producer shipped with Kafka 4.4 wraps the same library: kafka-console-producer.sh --bootstrap-server broker-0:9092 --topic trip-pings reads a line from the terminal and calls send() for each one, so every step below happens for a line you type.

The first 3 steps run on the thread that called send(), and they take microseconds. Nothing has crossed the network when send() returns: the record sits in a memory buffer inside the producer, and the call hands back a future that will hold the result. Step 4 belongs to a second thread the producer started on its own, the sender. It drains the buffer, groups the records headed for each partition, and writes one produce request per leader. The leader appends each record at the end of its partition's log, stamps it with the next offset, and the followers copy it as they do any other write. The reply travels back along the same connection, and the sender completes the future for every record the request carried.

The reply is a RecordMetadata: the topic, the partition the producer chose, and the offset the leader assigned. The partition was known before the request left, because the producer picked it. The offset was known to nobody until the append, because only the leader knows how long its log is at that moment, and 2 producers writing to the same partition get interleaved offsets in whatever order their requests arrived. Your code reaches the metadata in 1 of 2 ways: call get() on the future, which blocks the calling thread until the reply, or pass a callback to send(), which the sender thread invokes when the reply comes. The callback form keeps the application thread free to keep sending while earlier records are still in flight.

Step 2 of the path, picking the partition, is the producer's call, and the default rule is arithmetic. The producer takes the key's bytes, runs them through a hash function named murmur2, forces the result positive, and takes the remainder when dividing by the topic's partition count. For trip-pings with 6 partitions the formula is hash(key) % 6, and the answer is a number from 0 to 5. The same bytes always hash to the same number, so a key always lands in the same partition for as long as the topic keeps 6 partitions. The decision runs entirely inside the producer; it could be computed with the cluster unreachable.

The hash column holds stand-in values to keep the division readable; the real murmur2 result is a 32-bit number, and the remainder step is the same.

Every record with key trip-8841 lands in partition 5, and the leader appends them in the order the producer's requests arrive. A single producer sending ping 1, ping 2 and ping 3 for that trip therefore produces offsets that increase in that order, and anyone reading partition 5 sees the trip's pings in the sequence the phone sent them. That is the whole value of a key: it names a stream whose order matters, and the hash guarantees the stream never splits across partitions. Records with different keys get no ordering guarantee, so a reader of partitions 1 and 5 may see trip 8842's ping before or after trip 8841's.

A key also lets the record be found later. Because trip-8841 can only be in partition 5, a tool looking for that trip's pings reads 1 partition out of 6 and skips the other 5.

The hash can be bypassed in 2 ways. A record can carry an explicit partition number, set when the record is built, and the producer honors it without hashing; the key still travels with the record and is stored, but plays no part in routing. Or partitioner.class can name your own class, which receives the topic, the key bytes, the value bytes and the live partition count and returns a partition number. Section 4 uses that hook. Either way, the rule that readers rely on still holds: whatever puts a key's records in one partition must keep doing so, or the order you promised downstream breaks.

Given --property parse.key=true --property key.separator=:, the console producer splits each typed line at the first colon, so trip-8841:lat=37.61,lon=-122.38 sends key trip-8841 and lands in partition 5 every time.

Changing the partition count changes the divisor, and a key that lived in partition 5 under 6 partitions may land elsewhere under 8. A later lesson works through that move; for now, the count is part of the contract.

Step 6 of the path, the reply, happens when the leader decides the record is safe enough, and the producer sets that threshold with 1 setting, acks. It takes 3 values. With acks=0 the producer asks for no reply at all, and treats the send as done the instant the request is written to the socket; the leader may append it, or may never receive it, and the producer will not know. With acks=1 the leader replies as soon as the record is in its own log, before any follower has fetched it. With acks=all the leader waits until every replica currently in the in-sync set has copied the record, then replies. Kafka 4.4 defaults to acks=all.

The word all means the replicas in the in-sync set at the moment of the write, the set from the replication lesson. A follower that has dropped out of the ISR is skipped: with 3 replicas and 1 of them lagging, acks=all returns after the leader and the remaining follower hold the record. The acknowledgement travels with the replication: when the leader replies, the record already sits on every broker that could be elected leader next, so the leader's death a millisecond later changes nothing about that record. With acks=1 the same death loses the record if the followers had not yet fetched it, and the producer holds an offset for a record that no surviving log contains.

A later replication lesson adds a floor under how small the in-sync set may shrink before the leader refuses writes; without it, acks=all with an ISR of 1 is the leader alone.

The animation plays 6 payments through a 3-broker cluster with the followers deliberately slow. Switch between acks=1 and acks=all and watch when each record is marked acknowledged relative to the followers' copies; then strike the leader down after a couple of sends and compare what the surviving brokers hold against what the producer believed was done.

The price of acks is time per send. acks=0 costs 1 one-way trip, and the producer never waits. acks=1 costs a round trip to the leader, which inside 1 datacenter is typically 1 to 3 ms. acks=all adds the followers' fetch on top: the leader cannot answer until each in-sync follower has asked for the record and received it, which adds around 1 more round trip, so a send that took 2 ms at acks=1 takes 3 to 5 ms at acks=all. Because the producer keeps many records in flight at once, the longer wait rarely reduces how many records per second a producer can push; it lengthens how long each individual record waits for its callback.

The command-line producer takes the same setting on its command line: kafka-console-producer.sh --bootstrap-server broker-0:9092 --topic payments --producer-property acks=1 sends every typed line at the weaker level, which is handy for watching the difference and for nothing else.

Under acks=1 the producer's callbacks reported 5 landed records, and the new leader's log holds 3; the application has already written done for pay-4 and pay-5 somewhere else. Under acks=all the callbacks for pay-4 and pay-5 had not fired, so the producer still holds both records, and once a new leader is elected it sends them there, which section 3 walks through.

A produce request can fail for reasons that will pass and reasons that will not. The leader of partition 2 died and a new one is being elected: wait a moment and the same request to the new leader succeeds. The connection to broker 1 dropped: reconnect and try again. The request got no reply within request.timeout.ms: send it again. The producer classifies these as retriable and handles them itself, inside the sender thread, without waking the application. A record larger than the broker accepts, a key that cannot be serialized, a topic name with illegal characters, or a client that lacks permission to write to the topic will fail the same way on the hundredth attempt, so the producer stops at once and hands the error to your code.

Retries are bounded by time rather than count. delivery.timeout.ms starts when send() is called and, when it runs out, the record fails to the application with a TimeoutException no matter how many attempts were made; the default gives the producer 2 minutes. Inside that window, request.timeout.ms governs 1 attempt: 30 seconds with no reply counts as a failed attempt. Between attempts the producer waits a backoff that starts small and doubles each time up to a ceiling, so a 400 ms leader election costs a few backoffs and metadata refreshes, then 1 retry that lands. The retries setting still exists and still caps the count, but its default is so large that in practice the delivery timeout is what ends the wait.

The application wrote send() once and received 1 callback with an offset, 338 ms later than usual. Had the election outlasted the delivery timeout, that single callback would instead carry a TimeoutException, and your code would decide what to do with pay-7. The leader change itself is the replication lesson's story, playing out exactly as described there; the producer's part is to keep the record in its buffer, refresh metadata, and aim the next attempt at whoever leads now.

A retry can reach a leader that did append the previous attempt before the reply was lost. Kafka 4.4's default producer settings let the broker recognize such a resend and keep 1 copy; the intermediate producer lesson explains how.

The callback hears every outcome after the record entered the buffer, the same one get() on the future returns; a serializer error never reaches it, because send() throws it. The first branch carries the metadata from section 0. The second fires for a retriable error that outlasted the delivery timeout: decide whether the record is still worth sending and, if so, call send() again. The third fires for an error the producer refused to retry, and resending without changing the record or the configuration fails the same way.

A key buys order by pinning its records to 1 partition, and the pin holds no matter how much traffic the key carries. Suppose trip-pings were keyed by pickup zone instead of trip, so that every ping from a driver waiting at the airport carries key zone-SFO. The airport is where drivers queue, so half of all pings in the city may carry that 1 key, and all of them hash to partition 3. Partition 3's leader appends, replicates and serves half the topic's bytes while the other 5 partitions share the rest; a reader of partition 3 falls behind while readers of the others sit idle. That imbalance is skew, and a key that causes it is a hot key.

kafka-get-offsets.sh --bootstrap-server broker-0:9092 --topic trip-pings prints the end offset of every partition, which equals its record count while the log still starts at offset 0: partition 3 above holds 54% of 11.8 million records, and --time earliest shows where each log starts. kafka-log-dirs.sh --describe --bootstrap-server broker-0:9092 --topic-list trip-pings reports the bytes each partition occupies on each broker, which is the number the disks feel.

The reflex is to give the topic more partitions, and for a hot key it changes nothing: zone-SFO hashes to exactly 1 partition out of 12 just as it did out of 6, and that partition still carries half the topic. Partition count sets how many keys can be spread; it sets no limit on how much 1 key can concentrate. The fix has to change what the key names.

Ask what order the readers actually depend on. A rider's map needs 1 driver's pings in sequence; it never needs driver A's ping ordered against driver B's because both sat at the same airport. The unit whose order matters is the driver, so the key should name the driver: zone-SFO:driver-2201. Thousands of drivers mean thousands of keys, the hash spreads them over all 6 partitions, and each driver's pings still sit in 1 partition in send order. The order given up, across drivers inside a zone, was never promised to anyone. This is the general rule for a hot key: shrink the key to the finest unit that still needs ordering, and the heat spreads with it.

Sometimes readers key their own state on the original value and cannot take a longer key. Then the spreading moves into the producer's partitioner. A custom partitioner.class keeps the record's key as zone-SFO but routes it by a finer field taken from the value, say the driver id, hashed into a fixed set of partitions reserved for that zone. Readers still see key zone-SFO on every record; each driver still lands in 1 partition; the zone's traffic covers 2 or 3 partitions instead of 1. The partitioner must be deterministic over something stable in the record, because a partitioner that picks at random spreads the load and shreds every driver's order at the same time.

The function keeps the hot zone's drivers on partitions 1, 3 and 5, each driver on 1 of them for good, and sends every other key through an ordinary hash. The real class implements Kafka's Partitioner interface and hashes with murmur2 so that cold keys land where the default partitioner would put them; the Python shows the decision, which runs on the app thread at step 2 of the path for every record.