Messaging with Kafka · Commitwise

An order is placed. The warehouse must reserve stock, the customer gets an email, analytics counts the sale, and loyalty points are added. If the checkout service called all four directly, a slow email provider would slow down checkout, and a new team would need checkout to change its code to hear about orders. Instead, checkout publishes one event, OrderPlaced, and every interested service reads it at its own pace.

Apache Kafka is the most common way to do that between backend services. This tutorial builds a producer and consumers with Spring Boot, and then covers what the message queues lesson warns about: duplicates, ordering, poison messages and lost events.

New to some of these? Open what you need, or skip ahead.

Basics:What is a message queue?

A message queue sits between two programs. The producer puts a message in and moves on; a consumer takes it out and does the work later. If the consumer is slow or down, messages wait instead of getting lost, and the producer never notices.

Basics:What is Spring Boot?

Spring Boot is the most popular framework for Java and Kotlin backends. You write classes and mark them with annotations; Spring creates the objects (called beans), connects them to each other (dependency injection: a controller gets the service it needs through its constructor) and runs a web server that sends HTTP requests to your methods.

Starters (spring-boot-starter-web, spring-kafka…) add features by adding a dependency, with sensible defaults you can change in application.yml.

Part Not started

Topics, partitions and consumer groups

Picture a newspaper archive rather than a mailbox. A mailbox empties when you take a letter out; an archive keeps every issue on the shelf, and each reader keeps a bookmark of how far they’ve read. Kafka is the archive.

A topic is one shelf, for one kind of event: orders, payments.
A topic is split into partitions, numbered rows on the shelf. Each new event is appended to the end of one partition and gets a number, its offset.
Every event has a key, such as the order id. Events with the same key always go to the same partition, so they’re read in the order they were written.
A consumer group is one team of readers, for example “warehouse”. Every group sees every event. Within a group, each partition is read by exactly one member, so adding members (up to the number of partitions) reads faster.
Each group remembers its bookmark: the committed offset. Events aren’t deleted when read; they stay for the retention period (say 7 days) so other groups, or a replay, can read them again.

A topic is a set of append-only partition logs, replicated across brokers (replication factor 3, min.insync.replicas=2 is the usual durable setup). The producer’s partitioner hashes the key to choose a partition, so per-key ordering holds as long as the partition count doesn’t change. Consumers in a group split partitions between them; the group coordinator rebalances on joins, leaves and missed heartbeats or poll timeouts. Progress is the committed offset per partition, which makes replay (reset offsets) and independent consumers cheap. Parallelism is capped by partition count; ordering is only within a partition, never across a topic.

One topic, two consumer groups

Tap a box to see what it does.

Predict the output

The orders topic has 6 partitions. The warehouse service scales to 10 instances in the same consumer group. How many instances receive messages?

Topic = replicated partition logs (RF 3, min.insync.replicas=2). Key → partition hash: per-key ordering, none across partitions.
Consumer group: each partition to one member; parallelism ≤ partitions; rebalances on membership changes and poll timeouts.
Committed offsets per group per partition; retention independent of consumption, so replay and new consumers are cheap.
Adding partitions remaps keys: plan the count up front.

Check this part

Why use the order id as the message key?
Ordering is guaranteed only within a partition.
Does reading an event remove it from the topic?
That's how many groups and replays can read the same events.
What limits how many consumers in one group can work in parallel?
Each partition is assigned to one consumer in a group.

Kafka stores events in topics, split into partitions. Messages with the same key go to the same partition and stay in order. Each consumer group gets every message; within a group, each partition is read by one consumer, so partitions are the unit of parallelism.

Part Not started

Producing and consuming with Spring

With the spring-kafka dependency, sending is one call on a KafkaTemplate, and receiving is a method marked @KafkaListener. Spring turns your objects into JSON and back, keeps track of offsets, and runs the listener in its own threads.

Spring Kafka wraps the clients: KafkaTemplate for producing (async, returns a CompletableFuture<SendResult>), listener containers behind @KafkaListener with AckMode.RECORD/BATCH committing after the listener returns, and DefaultErrorHandler for retries and recovery. Producers default to acks=all and idempotence on recent clients, which removes duplicates caused by producer retries but says nothing about duplicates on the consumer side. Prefer explicit event schemas (JSON with versioning, or Avro/Protobuf with a schema registry) over serializing domain entities.

Publishing and consuming OrderPlaced

data class OrderPlaced(
    val eventId: String, val orderId: String, val customerId: String,
    val lines: List<Line>, val placedAt: Instant,
) {
    data class Line(val sku: String, val quantity: Int)
}

@Service
class OrderEvents(private val kafka: KafkaTemplate<String, OrderPlaced>) {
    fun placed(order: Order): CompletableFuture<SendResult<String, OrderPlaced>> {
        val event = OrderPlaced(UUID.randomUUID().toString(), order.id, order.customerId,
            order.lines, Instant.now())
        return kafka.send("orders", order.id, event)          // topic, key, value
    }
}

@Component
class ReserveStockListener(private val warehouse: Warehouse) {
    @KafkaListener(topics = ["orders"], groupId = "warehouse")
    fun on(event: OrderPlaced) {
        warehouse.reserve(event.orderId, event.lines)
        // returning normally = success; Spring commits the offset after this
    }
}

public record OrderPlaced(String eventId, String orderId, String customerId,
                          List<Line> lines, Instant placedAt) {
    public record Line(String sku, int quantity) {}
}

@Service
public class OrderEvents {
    private final KafkaTemplate<String, OrderPlaced> kafka;
    public OrderEvents(KafkaTemplate<String, OrderPlaced> kafka) { this.kafka = kafka; }

    public CompletableFuture<SendResult<String, OrderPlaced>> placed(Order order) {
        var event = new OrderPlaced(UUID.randomUUID().toString(), order.id(), order.customerId(),
                order.lines(), Instant.now());
        return kafka.send("orders", order.id(), event);        // topic, key, value
    }
}

@Component
public class ReserveStockListener {
    private final Warehouse warehouse;
    public ReserveStockListener(Warehouse warehouse) { this.warehouse = warehouse; }

    @KafkaListener(topics = "orders", groupId = "warehouse")
    public void on(OrderPlaced event) {
        warehouse.reserve(event.orderId(), event.lines());
        // returning normally = success; Spring commits the offset after this
    }
}

spring:
  kafka:
    bootstrap-servers: ${KAFKA_BOOTSTRAP_SERVERS}
    producer:
      acks: all                       # wait until replicas have it
      value-serializer: org.springframework.kafka.support.serializer.JsonSerializer
      properties:
        enable.idempotence: true      # no duplicates from producer retries
    consumer:
      auto-offset-reset: earliest     # a brand-new group starts from the beginning
      value-deserializer: org.springframework.kafka.support.serializer.JsonDeserializer
      properties:
        spring.json.trusted.packages: com.shop.events

The key (order.id()) keeps all events for one order in order.
eventId gives every event a unique id, which consumers use to recognize duplicates (next part).
acks: all means send succeeds only after Kafka has safely copied the event; with acks: 1, a broker crash can lose it.
Offsets after processing: Spring commits the offset after your method returns. If the app crashes in the middle, the event is delivered again after restart, never skipped.
Event classes are a contract between teams: add fields freely, but don’t rename or remove them, because old consumers still read new events and new consumers read old ones.

Spot the bugAn order with no event

Tap the line with the bug.

@Transactional
fun place(cart: Cart): Order {
    val order = orders.save(Order.from(cart))
    kafka.send("orders", order.id, OrderPlaced.of(order))
    payments.authorize(order)       // may throw
    return order
}

@Transactional
public Order place(Cart cart) {
    Order order = orders.save(Order.from(cart));
    kafka.send("orders", order.id(), OrderPlaced.of(order));
    payments.authorize(order);      // may throw
    return order;
}

The database and Kafka aren’t in the same transaction, so they can disagree in both directions:

Event without an order: the event is sent, then authorize throws and the database rolls back. The warehouse reserves stock and the customer gets an email for an order that doesn’t exist.
Order without an event: the order commits, but the send fails (Kafka unavailable, app killed after commit). send is asynchronous, and nobody checks its result. The order exists and nothing downstream ever hears about it.

Use the transactional outbox: save the event in an outbox table in the same database transaction as the order. A separate relay publishes outbox rows to Kafka and marks them sent. Either both the order and its event are committed, or neither is; the relay may publish a row twice after a crash, which idempotent consumers handle.

@Transactional
fun place(cart: Cart): Order {
    val order = orders.save(Order.from(cart))
    payments.authorize(order)
    outbox.save(OutboxEvent.of("orders", order.id, OrderPlaced.of(order)))     // same transaction
    return order
}
// A relay (a scheduled poller, or Debezium reading the database log) sends outbox rows to Kafka.

@Transactional
public Order place(Cart cart) {
    Order order = orders.save(Order.from(cart));
    payments.authorize(order);
    outbox.save(OutboxEvent.of("orders", order.id(), OrderPlaced.of(order)));   // same transaction
    return order;
}
// A relay (a scheduled poller, or Debezium reading the database log) sends outbox rows to Kafka.

Exerciseeasy

A loyalty points consumer

Write a listener in group loyalty that adds one point per euro to the customer’s account when an OrderPaid event (eventId, orderId, customerId, amountCents) arrives on the payments topic. Which key should the payments producer use, and why?

›

Points are per customer. Which events must be applied in order for a given customer, and which can run in parallel?

SolutionTry it yourself first

@Component
class LoyaltyListener(private val loyalty: LoyaltyService) {
    @KafkaListener(topics = ["payments"], groupId = "loyalty")
    fun on(event: OrderPaid) {
        val points = event.amountCents / 100
        loyalty.addPoints(event.customerId, points, event.eventId)
    }
}

@Component
public class LoyaltyListener {
    private final LoyaltyService loyalty;
    public LoyaltyListener(LoyaltyService loyalty) { this.loyalty = loyalty; }

    @KafkaListener(topics = "payments", groupId = "loyalty")
    public void on(OrderPaid event) {
        long points = event.amountCents() / 100;
        loyalty.addPoints(event.customerId(), points, event.eventId());
    }
}

Key the events by customerId: all of one customer’s payments then go to the same partition and are applied in order by one consumer, while different customers are processed in parallel. Passing eventId lets addPoints ignore an event it has already applied, because the same event can arrive twice (next part).

Producer and consumer settings that matter

send is async: handle the future (log and alert, or fail the request) or use the outbox; a fire-and-forget send loses events silently.
Batching: linger.ms (5–20) and batch.size trade a few milliseconds for much higher throughput; compression.type=zstd or lz4 for JSON.
Poll loop health: max.poll.records × processing time must stay under max.poll.interval.ms, or the consumer is kicked out and its partitions rebalanced, which redelivers the batch and can loop forever. Slow work: smaller batches, or hand off with care for offsets.
Rebalances: use the cooperative sticky assignor (default in recent clients) and static membership (group.instance.id) for rolling deploys, so restarts don’t stop the whole group.
Concurrency: @KafkaListener(concurrency = "3") runs three consumers in one instance; total consumers across instances still can’t exceed partitions.

KafkaTemplate.send is async: check the future or use an outbox. acks=all + idempotence on producers.
DB write + send isn’t atomic: outbox table in the same transaction, relay (poller or Debezium CDC).
Listener commits after return (at-least-once). Watch max.poll.interval.ms vs batch processing time; cooperative rebalancing, static membership.
Event schemas are contracts: additive changes, versioning, schema registry for Avro/Protobuf.

Check this part

When does Spring commit the offset for a @KafkaListener by default?
A crash before that means the message is delivered again, not lost.
Why save the event in an outbox table instead of sending it inside the transaction?
A relay publishes it afterwards.
A consumer's batch takes longer than max.poll.interval.ms. What happens?
Without a fix this can loop: smaller batches or faster processing.

Spring Kafka's KafkaTemplate sends events with a key; @KafkaListener methods consume them. Configure the producer for durability (acks=all, idempotence) and the consumer to commit offsets only after processing, so a crash causes a redelivery, never a loss.

Part Not started

Duplicates, retries and poison messages

Kafka promises that every event is delivered at least once. Not exactly once: if the consumer reserves stock and crashes a millisecond before its offset is committed, it gets the same event again after the restart. So the same event must be safe to process twice. That property is called idempotency.

Some events fail. If the warehouse database is briefly down, trying again in a second will work. But an event with a missing SKU will fail every time, forever. If the consumer keeps retrying it, every event behind it in that partition waits. Such a message is a poison message, and it belongs on a dead-letter topic where a person can look at it.

At-least-once delivery means consumers must be idempotent: dedupe by event id in the same transaction as the effect, or make the effect naturally idempotent (upserts, conditional updates, “set” rather than “add”). Kafka transactions give exactly-once for read-process-write within Kafka; the moment the effect is a database write or an HTTP call, idempotency is still yours. Errors split into transient (retry with backoff, bounded) and permanent (deserialization failures, validation errors: dead-letter immediately). A blocking retry holds the partition; non-blocking retry topics keep it moving at the cost of ordering for that key.

An idempotent consumer with retries and a dead-letter topic

@Component
class ReserveStockListener(
    private val processed: ProcessedEvents,
    private val warehouse: Warehouse,
) {
    @KafkaListener(topics = ["orders"], groupId = "warehouse")
    @Transactional
    fun on(event: OrderPlaced) {
        if (!processed.markIfNew(event.eventId)) return       // INSERT ... ON CONFLICT DO NOTHING
        warehouse.reserve(event.orderId, event.lines)         // same DB transaction as the mark
    }
}

@Configuration
class KafkaErrorConfig {
    @Bean
    fun errorHandler(template: KafkaTemplate<Any, Any>): DefaultErrorHandler {
        val recoverer = DeadLetterPublishingRecoverer(template)           // sends to orders.DLT
        val backoff = ExponentialBackOff(500, 2.0).apply { maxElapsedTime = 30_000 }
        return DefaultErrorHandler(recoverer, backoff).apply {
            addNotRetryableExceptions(UnknownSkuException::class.java, ValidationException::class.java)
        }
    }
}

@Component
public class ReserveStockListener {
    private final ProcessedEvents processed;
    private final Warehouse warehouse;
    // constructor omitted

    @KafkaListener(topics = "orders", groupId = "warehouse")
    @Transactional
    public void on(OrderPlaced event) {
        if (!processed.markIfNew(event.eventId())) return;   // INSERT ... ON CONFLICT DO NOTHING
        warehouse.reserve(event.orderId(), event.lines());    // same DB transaction as the mark
    }
}

@Configuration
public class KafkaErrorConfig {
    @Bean
    DefaultErrorHandler errorHandler(KafkaTemplate<Object, Object> template) {
        var recoverer = new DeadLetterPublishingRecoverer(template);       // sends to orders.DLT
        var backoff = new ExponentialBackOff(500, 2.0);
        backoff.setMaxElapsedTime(30_000);                                 // then give up
        var handler = new DefaultErrorHandler(recoverer, backoff);
        handler.addNotRetryableExceptions(UnknownSkuException.class, ValidationException.class);
        return handler;
    }
}

markIfNew inserts the event id into a processed_events table and returns false if it’s already there. Because it’s in the same transaction as the reservation, either both happen or neither does.
Retries: a failing event is retried after 0.5 s, 1 s, 2 s… for up to 30 seconds, which covers short outages.
Not retryable: an unknown SKU will never succeed, so it goes straight to the dead-letter topic orders.DLT with the error in its headers. Someone fixes the data and replays it.
Consumer lag (how far a group is behind the newest events) is the number to watch: growing lag means consumers can’t keep up or are stuck.

Spot the bugPoints added twice

Tap the line with the bug.

@KafkaListener(topics = ["payments"], groupId = "loyalty")
fun on(event: OrderPaid) {
    val points = event.amountCents / 100
    val account = accounts.findByCustomerId(event.customerId)
    account.points += points
    accounts.save(account)
    mail.send(event.customerId, "You earned $points points")
}

@KafkaListener(topics = "payments", groupId = "loyalty")
public void on(OrderPaid event) {
    long points = event.amountCents() / 100;
    Account account = accounts.findByCustomerId(event.customerId());
    account.setPoints(account.getPoints() + points);
    accounts.save(account);
    mail.send(event.customerId(), "You earned " + points + " points");
}

The listener isn’t idempotent. The points are saved, then mail.send throws, so the error handler retries the whole event: the points are added again (and again, on every retry), and the email is sent once more each time it gets through. A rebalance or a crash after saving does the same thing without any error at all.

Record that the event was applied, in the same transaction as the points, and skip it if it was already applied. Keep side effects that can’t be undone (emails, calls to other services) out of the retried part: publish a PointsEarned event through the outbox and let a separate consumer send the email.

@KafkaListener(topics = ["payments"], groupId = "loyalty")
@Transactional
fun on(event: OrderPaid) {
    if (!processed.markIfNew(event.eventId)) return
    val points = event.amountCents / 100
    accounts.addPoints(event.customerId, points)      // UPDATE ... SET points = points + :p
    outbox.save(OutboxEvent.of("loyalty", event.customerId, PointsEarned(event.customerId, points)))
}

@KafkaListener(topics = "payments", groupId = "loyalty")
@Transactional
public void on(OrderPaid event) {
    if (!processed.markIfNew(event.eventId())) return;
    long points = event.amountCents() / 100;
    accounts.addPoints(event.customerId(), points);   // UPDATE ... SET points = points + :p
    outbox.save(OutboxEvent.of("loyalty", event.customerId(), new PointsEarned(event.customerId(), points)));
}

Production puzzles

Predict the output

A single event with a malformed JSON payload lands in partition 4 of orders. The consumer has the default DefaultErrorHandler but no ErrorHandlingDeserializer. What happens to partition 4?

Choosing a retry strategy

Blocking retries (DefaultErrorHandler + backoff) keep per-key ordering but stall the whole partition while retrying. Keep the total short (seconds), or one slow dependency backs up every key on that partition.
Non-blocking retries (@RetryableTopic: orders-retry-1s, orders-retry-30s, orders.DLT) move failing events aside so the partition keeps flowing, but later events for the same key can overtake the failed one. Fine for emails; wrong for a state machine where OrderCancelled must follow OrderPlaced.
DLT handling is a process, not a topic: alert on DLT growth, include the exception and original offsets in headers, and have a tool to replay after a fix.
Lag-based alerting and autoscaling: consumer lag (by group and partition) and the age of the oldest unprocessed event are the SLIs; scale on lag, up to the partition count.

At-least-once: idempotent consumers (dedupe table in the same transaction, or naturally idempotent writes). Kafka EOS doesn’t cover DB or HTTP side effects.
DefaultErrorHandler: bounded exponential backoff, not-retryable exceptions straight to DeadLetterPublishingRecoverer.
ErrorHandlingDeserializer, or a bad payload blocks the partition in poll().
Blocking retries keep order but stall partitions; @RetryableTopic keeps flow but reorders per key.
Alert on lag, oldest-event age and DLT growth; have a replay tool.

Check this part

Why must a Kafka consumer be idempotent?
Delivery is at least once, not exactly once.
An event fails because its SKU doesn't exist. What should happen?
It will never succeed; a person needs to look at it.
What do non-blocking retry topics give up?
Later events for the same key can be processed before the retried one.

Every consumer must be idempotent, because Kafka delivers at least once. Transient errors are retried with backoff; messages that can never succeed go to a dead-letter topic instead of blocking their partition. Monitor consumer lag.

Key takeaways

Topics are split into partitions; the same key always goes to the same partition, in order.
Each consumer group gets every event; within a group, partitions are shared between instances.
Publish events through an outbox so an order and its event are saved together.
Consumers must handle the same event twice; failing events are retried, then sent to a dead-letter topic.

Partition count caps parallelism and changing it remaps keys; ordering is per partition only.
Async sends and dual writes lose events: outbox plus relay, acks=all, idempotent producers.
Idempotent consumers in the same transaction as the effect; side effects that can’t be undone go through their own events.
Bounded blocking retries vs retry topics is a trade-off between ordering and flow; watch lag and the DLT.