Producer Internals & Configuration

flowchart LR
  PR[ProducerRecord]
  SER[Serializer\nkey + value]
  PAR[Partitioner]
  ACC[RecordAccumulator\nper-partition batches]
  SND[Sender thread\nnetwork I/O]
  BRK[Broker leader]
  PR --> SER --> PAR --> ACC --> SND --> BRK

ProducerRecord

The unit you create: topic, optional partition, key, value, headers, timestamp (CREATE_TIME or LOG_APPEND_TIME). If partition is set explicitly, the partitioner is skipped. Headers carry metadata (trace IDs, content-type) without polluting the value payload.

ProducerRecord<String, OrderEvent> record = new ProducerRecord<>(
    "orders",                    // topic
    order.getCustomerId(),       // key → partition affinity
    order                        // value
);
record.headers().add("correlation-id", uuid.getBytes(StandardCharsets.UTF_8));
producer.send(record, (metadata, ex) -> {
    if (ex != null) log.error("Failed offset={}", metadata != null ? metadata.offset() : "n/a", ex);
});

Serializer

Serializer<K> and Serializer<V> convert objects to bytes before batching. Configure via key.serializer / value.serializer (class name or Spring property). With Schema Registry: KafkaAvroSerializer prepends magic byte + schema ID.

Partitioner

Chooses partition when not explicitly set on the record. Default: sticky partitioner (Kafka 2.4+) or murmur2 hash of key. Runs on the calling thread before append to accumulator.

RecordAccumulator

In-memory buffer organized as batch deque per topic-partition (actually per TopicPartition → leader broker node). Each batch holds compressed record sets waiting for batch.size or linger.ms. Total memory capped by buffer.memory—when full, send() blocks up to max.block.ms.

RecordAccumulator (buffer.memory = 64 MB)
├── TopicPartition orders-0 → Broker 1  [Batch: 12 KB] [Batch: 4 KB filling…]
├── TopicPartition orders-1 → Broker 2  [Batch: 48 KB ready → Sender drains]
├── TopicPartition orders-2 → Broker 3  [Batch: empty, linger timer running]
└── …

Sender thread

Background thread polls the accumulator for ready batches (full size, linger expired, or flush/close). Groups batches by broker node, sends Produce requests, handles responses, retries retriable errors, and invokes callbacks. One Sender per KafkaProducer instance—sharing a producer across threads is supported; sharing one producer per thread is simpler.

In-flight requests

max.in.flight.requests.per.connection (default 5) allows multiple unacknowledged produce batches per broker connection. Higher values improve throughput but—with retries > 0 and acks > 0—can reorder records within a partition if batch 2 succeeds before batch 1 is retried.

Default partitioner: sticky (Kafka 2.4+)

Legacy behavior (pre-2.4): round-robin across partitions when key is null—tiny batches spread everywhere. Sticky partitioner (current default): null-key records stick to one partition until a batch is ready to send, then rotate to the next—improving batch fill ratio without sacrificing long-term fairness.

Key-based partitioning

When key is non-null: partition = (murmur2(keyBytes) & 0x7fffffff) % numPartitions. Same key → same partition → per-key ordering. Key bytes come from the serialized key, not hashCode() on the Java object unless your serializer preserves that.

Null key behavior

No ordering guarantee across records. Sticky partitioner batches null-key records to one partition at a time, then switches—good for throughput on audit/log firehose topics where order is irrelevant.

Custom partitioner

Implement org.apache.kafka.clients.producer.Partitioner for domain routing—e.g. route EU customers to partitions 0–3, US to 4–7.

public class RegionPartitioner implements Partitioner {
  @Override
  public int partition(String topic, Object key, byte[] keyBytes,
                       Object value, byte[] valueBytes, Cluster cluster) {
    String region = key != null ? key.toString() : "default";
    int numPartitions = cluster.partitionCountForTopic(topic);
    return switch (region) {
      case "EU" -> Math.floorMod(region.hashCode(), numPartitions / 2);
      case "US" -> (numPartitions / 2) + Math.floorMod(region.hashCode(), numPartitions / 2);
      default -> Math.floorMod(region.hashCode(), numPartitions);
    };
  }
  @Override public void close() {}
  @Override public void configure(Map<String, ?> configs) {}
}

spring.kafka.producer.properties:
  partitioner.class: com.example.kafka.RegionPartitioner

Partition skew scenarios

At-most-once

acks=0 — producer does not wait for broker acknowledgment. Fire-and-forget. Records can be lost if the broker fails after accepting socket bytes but before durable write, or if the batch never arrives. Acceptable for: metrics, sampling, non-critical telemetry where loss is statistically tolerable.

At-least-once (default production pattern)

acks=1 or acks=all with retries > 0 (default Integer.MAX_VALUE with idempotence). If the producer does not receive an ack (timeout, broker restart), it retries—the same record may be written twice. Consumers must deduplicate (idempotent handler, natural key in DB) unless you add idempotence or transactions.

Exactly-once: idempotent producer

enable.idempotence=true (default when using acks=all in modern clients). Broker assigns a Producer ID (PID); each partition gets a monotonic sequence number per batch. Broker deduplicates retries within the same producer session—no duplicate batches land in the log.

• Automatically sets acks=all, retries=MAX, max.in.flight.requests.per.connection=5 (safe with sequences)
• Limitation: PID changes on producer restart—duplicates possible across sessions without transactional.id

Exactly-once: transactional API

For atomic write across multiple partitions/topics and consume-transform-produce:

1. initTransactions() — register with transaction coordinator, fence zombie producers with same transactional.id
2. beginTransaction()
3. send() — records tagged with PID, epoch, sequence
4. sendOffsetsToTransaction() — atomic offset commit with output records (stream processing)
5. commitTransaction() or abortTransaction()

Transaction state lives in internal topic __transaction_state (50 partitions default). Consumers with isolation.level=read_committed skip aborted transactions and uncommitted data.

sequenceDiagram
  participant P as Transactional Producer
  participant TC as Transaction Coordinator
  participant B as Partition Leaders
  participant C as read_committed Consumer
  P->>TC: initTransactions / beginTransaction
  P->>B: produce batches (txn marker)
  P->>TC: commitTransaction
  TC->>B: write COMMIT marker
  C->>B: fetch (sees only committed)

Properties props = new Properties();
props.put(ProducerConfig.TRANSACTIONAL_ID_CONFIG, "order-processor-1");
props.put(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG, "true");

KafkaProducer<String, String> producer = new KafkaProducer<>(props);
producer.initTransactions();

try {
  producer.beginTransaction();
  producer.send(new ProducerRecord<>("orders", "42", "placed"));
  producer.send(new ProducerRecord<>("inventory", "sku-9", "reserved"));
  producer.commitTransaction();
} catch (Exception e) {
  producer.abortTransaction();
  throw e;
}

acks: 0, 1, all — when to use each

• acks=0 — metrics, lossy telemetry, max throughput, no durability promise
• acks=1 — leader wrote to local log; follower loss before replication means data loss on leader failure
• acks=all (-1) — all ISR replicas acked; requires min.insync.replicas alignment on broker/topic

linger.ms and batch.size interaction

A batch sends when either condition is met: compressed size ≥ batch.size OR linger.ms elapsed since first record added. High linger.ms + large batch.size = maximum throughput; both low = minimum latency, poor batch fill.

compression.type trade-offs

bootstrap.servers=broker1:9092,broker2:9092,broker3:9092
key.serializer=org.apache.kafka.common.serialization.StringSerializer
value.serializer=org.apache.kafka.common.serialization.StringSerializer
acks=all
enable.idempotence=true
retries=2147483647
max.in.flight.requests.per.connection=5
compression.type=lz4
linger.ms=10
batch.size=131072
buffer.memory=67108864
delivery.timeout.ms=120000

spring:
  kafka:
    bootstrap-servers: broker1:9092,broker2:9092,broker3:9092
    producer:
      key-serializer: org.apache.kafka.common.serialization.StringSerializer
      value-serializer: org.springframework.kafka.support.serializer.JsonSerializer
      acks: all
      properties:
        enable.idempotence: true
        compression.type: lz4
        linger.ms: 10
        batch.size: 131072
        delivery.timeout.ms: 120000
        max.in.flight.requests.per.connection: 5

Retriable vs non-retriable errors

Callback API

send(ProducerRecord, Callback) fires on the Sender thread when the batch containing your record is acknowledged or failed. onCompletion(RecordMetadata metadata, Exception exception) — if exception != null, the record did not land (after retries exhausted).

producer.send(record, (metadata, ex) -> {
  if (ex != null) {
    metrics.increment("producer.errors");
    deadLetterPublisher.publish(record, ex);
    return;
  }
  log.debug("topic={} partition={} offset={}",
      metadata.topic(), metadata.partition(), metadata.offset());
});

Dead letter topic pattern

Records that cannot be delivered (serialization failure caught pre-send, or non-retriable post-send) go to {original-topic}.DLT with headers: kafka_dlt-exception-message, original topic/partition/offset, timestamp. Enables replay after fix without blocking the main pipeline.

Baseline benchmark

kafka-producer-perf-test.sh \
  --topic perf-test \
  --num-records 5000000 \
  --record-size 1024 \
  --throughput -1 \
  --producer-props bootstrap.servers=broker:9092 \
    acks=1 compression.type=lz4 linger.ms=20 batch.size=131072

Recommended throughput profile

Application patterns

• Async send — fire send() with callback; never get() per record in loop
• Flush on shutdown — producer.flush() then close() with timeout in @PreDestroy
• Partition count — more partitions = more parallel broker writes (ceiling: broker metadata/file handles)
• Binary payloads — Avro/Protobuf/JSON compressed beats verbose JSON strings

spring.kafka.producer:
  acks: 1
  properties:
    linger.ms: 20
    batch.size: 131072
    compression.type: lz4
    buffer.memory: 134217728
    enable.idempotence: true

Low-latency profile

acks=1
linger.ms=0
batch.size=16384
compression.type=none
max.in.flight.requests.per.connection=1
enable.idempotence=true
delivery.timeout.ms=30000

spring.kafka.producer:
  acks: 1
  properties:
    linger.ms: 0
    batch.size: 16384
    compression.type: none
    max.in.flight.requests.per.connection: 1
    enable.idempotence: true

What to avoid on the latency path

• Large linger.ms — directly adds artificial delay
• Compression — CPU time per batch; use only if network is the proven bottleneck
• acks=all — waits for slowest ISR replica; necessary for durability, costly for latency
• Synchronous send().get() — blocks app thread on broker RTT
• TLS without hardware acceleration on CPU-bound hosts — profile SSL handshake and encrypt cost