60-Day Kafka 4 Learning Plan · Week 1 — Day 5 of 60

---

60-Day Kafka 4 Learning Plan · Week 1 — Foundations
Sources: Kafka: The Definitive Guide Ch.3 · kafka.apache.org/43 — Producer API

Goal

Master how Kafka producers work end-to-end — from app code to broker acknowledgment, including delivery guarantees, failure handling, and production monitoring.

1. Producer internal flow

Your App         Serializer      Partitioner     Record             Sender Thread
kafkaTemplate  → key→bytes    → hash(key)     → Accumulator      → broker
.send(t,k,v)    value→bytes     mod N           batches by part.

                                                 ←←←← batch.size + linger.ms ←←←←

                                          ←── ack / error callback ────────────────

Pipeline steps:

1. App calls kafkaTemplate.send(topic, key, value)
2. Serializer converts key and value to byte arrays
3. Partitioner decides which partition (hash(key) mod numPartitions)
4. Record Accumulator buffers records into batches per partition
5. Sender Thread (background) drains batches → sends to broker leader
6. Broker responds with ack or error → callback fires

2. acks — acknowledgment levels

Production rule: acks=all + min.insync.replicas=2 + replication.factor=3 = zero data loss guarantee

3. Retries & Idempotent Producer

The retry duplicate problem

Producer → broker  (broker writes msg, then crashes before sending ack)
         ← network error
Producer retries → broker writes AGAIN → DUPLICATE message in partition!

Solution — enable.idempotence=true (Kafka 4 default)

Each message is tagged with a PID (Producer ID) + sequence number. The broker tracks sequence numbers per producer and silently drops duplicates on retry.

Producer  →  msg {PID=42, seq=7, value="order#123"}  →  Broker
                                ← ack lost
Producer  →  msg {PID=42, seq=7, value="order#123"}  →  Broker
                                                         "seq=7 already seen → drop"

Note: idempotence guarantees no duplicates per partition, per producer session. It does not span multiple partitions or multiple topics — that’s what transactions are for (next section).

4. Transactional producers — exactly-once across partitions/topics

Idempotence solves duplicates on a single partition. When a producer writes to multiple partitions or topics atomically (e.g. “write order event + write payment event, both or neither”), you need transactions.

@Bean
public ProducerFactory<String, String> transactionalProducerFactory() {
    Map<String, Object> props = new HashMap<>();
    props.put(ProducerConfig.TRANSACTIONAL_ID_CONFIG, "boottech-producer-tx-1"); // must be unique per instance
    props.put(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG, true); // required for transactions
    DefaultKafkaProducerFactory<String, String> factory = new DefaultKafkaProducerFactory<>(props);
    factory.setTransactionIdPrefix("boottech-tx-");
    return factory;
}

kafkaTemplate.executeInTransaction(operations -> {
    operations.send("orders-topic", orderId, orderPayload);
    operations.send("payments-topic", orderId, paymentPayload);
    return true; // commit — if an exception is thrown instead, Kafka aborts both
});

How it works under the hood:

• transactional.id binds the producer to a Transaction Coordinator broker
• beginTransaction() → writes go to partitions but are marked READ_UNCOMMITTED
• commitTransaction() → coordinator writes a commit marker; consumers with isolation.level=read_committed only see committed records
• If the producer crashes mid-transaction, the coordinator fences the old transactional.id (via an incremented epoch) so a zombie instance can’t commit stale data

Use case fit: Kafka Streams “consume-transform-produce” exactly-once pipelines, or Spring’s @Transactional + KafkaTransactionManager when a single message triggers writes to several topics that must succeed or fail together.

5. Partitioning strategies

Sticky Partitioner (Kafka 2.4+ default for null keys): batches to the same partition until full, then switches — better throughput than pure round-robin.

Custom partitioner example

public class VipAwarePartitioner implements Partitioner {

    @Override
    public int partition(String topic, Object key, byte[] keyBytes,
                          Object value, byte[] valueBytes, Cluster cluster) {
        int numPartitions = cluster.partitionsForTopic(topic).size();

        if (key != null && key.toString().startsWith("VIP-")) {
            return 0; // VIP traffic always lands on partition 0 — dedicated consumer capacity
        }
        return Math.abs(Utils.murmur2(keyBytes)) % (numPartitions - 1) + 1; // rest spread across remaining partitions
    }

    @Override public void close() {}
    @Override public void configure(Map<String, ?> configs) {}
}

spring.kafka.producer:
  properties:
    partitioner.class: com.boottech.kafka.VipAwarePartitioner

6. Error handling — retriable vs fatal exceptions

Not all send failures are equal. The producer’s retries config only helps with retriable errors — fatal ones need application-level handling.

kafka.send("boottech-topic", "key", "payload")
    .whenComplete((result, ex) -> {
        if (ex instanceof RetriableException) {
            log.warn("Transient error, Kafka client will retry internally", ex);
        } else if (ex != null) {
            log.error("Fatal send error — routing to dead-letter", ex);
            deadLetterPublisher.publish("boottech-topic", "key", "payload", ex);
        }
    });

delivery.timeout.ms (default 120000) is the umbrella: it bounds the entire send attempt — batching + retries + network — after which the callback fails even if retries hasn’t been exhausted.

7. Key producer configs

8. Spring Boot producer — production config

# application.yml
spring.kafka.producer:
  bootstrap-servers: localhost:9092
  key-serializer: org.apache.kafka.common.serialization.StringSerializer
  value-serializer: org.springframework.kafka.support.serializer.JsonSerializer
  acks: all                      # strongest durability
  retries: 10                    # retry on transient errors
  enable-idempotence: true       # no duplicates on retry
  batch-size: 32768              # 32 KB batch
  linger-ms: 10                  # wait 10ms to fill batch
  compression-type: snappy       # compress batches
  properties:
    delivery.timeout.ms: 120000
    max.block.ms: 60000

9. send modes — when to use each

@Autowired KafkaTemplate<String, String> kafka;

// 1. Fire and forget — no callback, errors silently dropped
//    use: logs, analytics, non-critical events
kafka.send("boottech-topic", "userId-123", "payload");

// 2. Async + callback — non-blocking, recommended for most apps
//    use: business events where you need error visibility
kafka.send("boottech-topic", "key", "payload")
    .whenComplete((result, ex) -> {
        if (ex != null) {
            log.error("Send failed", ex);
        } else {
            log.info("Sent to partition={}, offset={}",
                result.getRecordMetadata().partition(),
                result.getRecordMetadata().offset());
        }
    });

// 3. Synchronous — blocks thread until broker acks
//    use: critical single messages where you must confirm before proceeding
//    WARNING: kills throughput — avoid in hot paths
RecordMetadata meta = kafka.send("boottech-topic", "key", "payload")
    .get(); // blocks

10. Monitoring producer health

Key JMX metrics to wire into Micrometer/Grafana dashboards:

@Bean
public MeterRegistryCustomizer<MeterRegistry> kafkaProducerMetrics(KafkaTemplate<String, String> template) {
    return registry -> new KafkaClientMetrics(template.getProducerFactory().createProducer()).bindTo(registry);
}

11. Testing producers

Unit test — MockProducer (no broker needed):

MockProducer<String, String> mockProducer =
    new MockProducer<>(true, new StringSerializer(), new StringSerializer());

KafkaTemplate<String, String> template = new KafkaTemplate<>(new MockProducerFactory<>(mockProducer));
template.send("boottech-topic", "key1", "payload1");

assertThat(mockProducer.history()).hasSize(1);
assertThat(mockProducer.history().get(0).key()).isEqualTo("key1");

Integration test — @EmbeddedKafka (real in-memory broker):

@SpringBootTest
@EmbeddedKafka(partitions = 3, topics = "boottech-topic")
class ProducerIntegrationTest {

    @Autowired KafkaTemplate<String, String> kafkaTemplate;

    @Test
    void sendsMessageSuccessfully() throws Exception {
        SendResult<String, String> result =
            kafkaTemplate.send("boottech-topic", "key1", "payload1").get(5, TimeUnit.SECONDS);

        assertThat(result.getRecordMetadata().partition()).isGreaterThanOrEqualTo(0);
    }
}

Use MockProducer for fast, isolated unit tests of your service logic; reserve @EmbeddedKafka for integration tests that need real partition/offset behavior.

12. Common pitfalls

• Not setting max.block.ms — under broker outage, .send() can hang indefinitely waiting for metadata
• retries without enable.idempotence — retries can cause message reordering within a partition; always pair with idempotence
• Reusing the same transactional.id across instances — causes fencing errors; must be unique per producer instance (e.g. suffix with pod name)
• Ignoring RecordTooLargeException — silently dropped in fire-and-forget mode; always monitor record-error-rate
• Blocking on .get() in a hot path — a single synchronous send can throttle an entire request thread pool under load

TL;DR

A producer serializes → partitions → batches → sends messages to the broker leader. acks=all + enable.idempotence=true (Kafka 4 default) gives the strongest durability with no duplicate risk per partition. For atomicity across partitions/topics, use transactions (transactional.id + read_committed consumers). Key-based partitioning ensures ordering per entity. In Spring Boot, use async + callback sends — fire-and-forget silently loses errors, synchronous sends block throughput. Always separate retriable errors (handled internally) from fatal ones (need dead-lettering), and monitor record-error-rate / record-retry-rate in production.

---

Support me through GitHub Sponsors.

Next

➡️ Day 6 — Consumers: @KafkaListener, consumer groups, offset commit modes, rebalancing

Resources

• 📘 Kafka: The Definitive Guide — Chapter 3 (Kafka Producers)
• 🌐 kafka.apache.org/43/documentation/#producerconfigs
• 🌐 Spring Kafka — KafkaTemplate
• 🌐 Kafka Transactions design (KIP-98)

👉 Link to Medium blog