Producers

batch.size caps how many bytes a producer collects per partition before sending; linger.ms tells it to wait up to that many milliseconds for more records to fill a batch. Bigger batches and more lingering raise throughput but add latency.

Retriable errors (like a leader change or timeout) are transient, so the producer automatically resends. Fatal errors (like an unknown topic, message too large, or auth failure) cannot be fixed by resending, so the send fails immediately.

An idempotent producer guarantees that producer retries don't create duplicate messages on a partition. You enable it by setting enable.idempotence=true (the default since Kafka 3.0).

For a keyed record, Kafka hashes the key and maps it to a partition. The same key always lands on the same partition (as long as the partition count is unchanged), which keeps records with that key in order.

send() is asynchronous. The producer serializes the key/value, picks a partition, and appends the record to an in-memory buffer (RecordAccumulator). A background Sender thread later batches and transmits it. send() returns a Future immediately, before the broker has the record.

A Serializer turns your key/value objects into the byte arrays Kafka stores. You set them with key.serializer and value.serializer producer configs, e.g. StringSerializer for text.

replication.factor sets how many copies of each partition exist. min.insync.replicas sets how many in-sync copies must acknowledge a write under acks=all. Only acks=all enforces min.insync.replicas; together they define how many failures the system tolerates without losing acknowledged data.

compression.type tells the producer to compress each batch as a whole using a codec: gzip (best ratio, slow), snappy (fast, modest ratio), lz4 (fast), or zstd (great ratio and good speed). Compression is per-batch, so bigger batches compress better.

Each producer gets a producer ID (PID). Each batch carries a per-partition sequence number that increases by one. The broker remembers the last sequence written per (PID, partition) and discards any batch it has already seen.

With a null key there is nothing to hash, so Kafka picks a partition without a key. The old DefaultPartitioner round-robined per record; KIP-480's sticky partitioner instead fills one partition's batch, then switches, producing fuller batches and lower latency.

The RecordAccumulator is the producer's in-memory buffer. It keeps one deque of ProducerBatches per topic-partition. A record is appended to the partition's current batch. A batch becomes sendable when it fills to batch.size or when linger.ms elapses, whichever comes first.

max.request.size limits how big a single producer request/record can be (default ~1 MB). message.max.bytes is the broker's per-batch ceiling, and max.message.bytes is its per-topic version. If the producer allows bigger messages than the broker accepts, the broker rejects them; replicas also need replica.fetch.max.bytes large enough or replication stalls.

delivery.timeout.ms is the single upper bound on the total time from send() to success or failure, covering batching, all retries, and inflight requests. It must be >= linger.ms + request.timeout.ms. When it expires, the record fails with TimeoutException regardless of remaining retries.

If more than one request is in flight per connection and an earlier batch is retried while a later one already succeeded, the retried batch lands after it, reordering messages within a partition. Enabling idempotence (enable.idempotence=true) prevents reordering even with up to 5 in-flight requests.

Idempotence guarantees exactly-once writes to a single partition within one producer session. It does NOT give cross-partition atomicity, cross-session deduplication, or atomic consume-process-produce. Those require transactions (transactional.id, initTransactions, begin/commit).

buffer.memory caps the total bytes the accumulator can hold. If the app produces faster than the Sender can drain, the buffer fills. Then send() blocks waiting for free space for up to max.block.ms; if it can't get memory in time, send() throws a TimeoutException.

KafkaAvroSerializer registers the record's Avro schema in Schema Registry, gets a schema ID, and writes a magic byte + 4-byte schema ID + Avro-encoded payload. The registry enforces compatibility per subject before allowing new schema versions.

request.timeout.ms is how long the producer waits for a single request's ack before giving up on that attempt. retry.backoff.ms is the pause before retrying a failed attempt, so the producer doesn't hammer a struggling broker. Both repeat until delivery.timeout.ms expires.

Idempotence requires acks=all, retries>0, and max.in.flight.requests.per.connection<=5. If you explicitly set a conflicting value (e.g., acks=1), the producer fails fast with a ConfigException.

Both use sticky batching for null keys. DefaultPartitioner hashes non-null keys with murmur2 to preserve per-key ordering; UniformStickyPartitioner ignores the key entirely and sticks for every record, so it never gives key-based ordering.

send() returns a Future<RecordMetadata>; calling future.get() blocks until the broker responds, turning async into sync. Alternatively pass a Callback to send(record, callback) that the producer invokes asynchronously on completion. Use Future.get() for sync confirmation, Callback for non-blocking handling.

Order inside send(): interceptor.onSend (sees objects) -> key/value serializers (produce bytes) -> partitioner (uses serialized key) -> accumulator/batching -> network. onAcknowledgement fires later on ack/failure.

StringSerializer encodes text to UTF-8 bytes; ByteArraySerializer passes byte[] through unchanged. Use String for human-readable text/JSON-as-string; use ByteArray when you've already encoded bytes yourself.

A record is 'written at the leader' once the leader appends it to its log. It becomes 'committed' only once all in-sync replicas have replicated it — at which point the high watermark advances. Consumers only read up to the high watermark, so they never see uncommitted records.

Both occur with acks=all when the in-sync replica count is below min.insync.replicas. NotEnoughReplicas is raised before the leader appends the record; NotEnoughReplicasAfterAppend is raised after appending but before full replication. Both are retriable, so the producer retries until enough replicas rejoin the ISR.

Normally a broker stores the producer's compressed batch untouched (zero-copy friendly). But if the topic's compression.type forces a different codec, or older message-format conversion / timestamp validation / offset assignment requires it, the broker must decompress and recompress, costing CPU and breaking the zero-copy path.

acks=all means a write is acknowledged only after all in-sync replicas store it, so an acked record can't be silently lost on leader failover. Idempotence dedups retries, but only acks=all makes the underlying write durable enough for the guarantee to mean exactly-once.

Implement org.apache.kafka.clients.producer.Partitioner (the partition() method returns an int partition), then set partitioner.class to your class. A common reason is routing hot or special keys to dedicated partitions to control skew or isolate VIP traffic.

Skew comes from uneven key distribution (hot keys), not the hash. Options: salt or compound the key, route hot keys with a custom partitioner, or increase partitions. The tension: spreading a key for balance destroys the per-key ordering the key was meant to guarantee.