Building the Consumer Microservice
The producer's events are only useful once something reads them. This module wires up @KafkaListener and works through the offset-management decisions that actually matter in production.
Learning objectives
- Beginner: A single-instance consumer with default auto-commit, fine for a low-stakes dev environment where occasional reprocessing is harmless.
- Intermediate: Manual acknowledgment tied to a successful database write, plus concurrency matched to partition count for real throughput.
- Advanced: Combine manual acknowledgment with idempotent upserts in the service layer (Module 10) and DLT-based recovery (Module 14) so at-least-once delivery can never corrupt state, even under repeated rebalances.
@Component public class LibraryEventsConsumer { private final LibraryEventsService libraryEventsService; @KafkaListener(topics = "library-events", groupId = "library-events-listener-group") public void onMessage(ConsumerRecord<String, String> record) throws JsonProcessingException { log.info("received record: partition={} offset={} key={}", record.partition(), record.offset(), record.key()); libraryEventsService.processLibraryEvent(record); } }
@KafkaListener hides the manual poll loop from §03.2 entirely — Spring manages the container thread that repeatedly calls poll(), dispatches each record to your method, and (by default) commits the offset once your method returns without throwing.
💻 Code example
@Component public class LibraryEventsConsumer { private final LibraryEventsService libraryEventsService; @KafkaListener(topics = "library-events", groupId = "library-events-listener-group") public void onMessage(ConsumerRecord<String, String> record) throws JsonProcessingException { log.info("received record: partition={} offset={} key={}", record.partition(), record.offset(), record.key()); libraryEventsService.processLibraryEvent(record); } }
Since the producer (Module 06) serialized with Jackson to a plain String, the consumer mirrors that with StringDeserializer and deserializes into a DTO explicitly inside the service layer — consistent with §06.5's choice to keep serialization explicit rather than relying on JsonDeserializer 's automatic type-header behavior.
public void processLibraryEvent(ConsumerRecord<String, String> record) throws JsonProcessingException { LibraryEvent libraryEvent = objectMapper.readValue(record.value(), LibraryEvent.class); switch (libraryEvent.libraryEventType()) { case NEW -> save(libraryEvent); case UPDATE -> { validate(libraryEvent); save(libraryEvent); } } }
💻 Code example
public void processLibraryEvent(ConsumerRecord<String, String> record) throws JsonProcessingException { LibraryEvent libraryEvent = objectMapper.readValue(record.value(), LibraryEvent.class); switch (libraryEvent.libraryEventType()) { case NEW -> save(libraryEvent); case UPDATE -> { validate(libraryEvent); save(libraryEvent); } } }
spring: kafka: consumer: bootstrap-servers: localhost:9092 group-id: library-events-listener-group auto-offset-reset: earliest key-deserializer: org.apache.kafka.common.serialization.StringDeserializer value-deserializer: org.apache.kafka.common.serialization.StringDeserializer
auto-offset-reset: earliest tells the consumer what to do the very first time this group-id connects (or if its committed offset has expired past retention) — start from the oldest retained record rather than only new ones (latest). This is a one-time decision per new group; it has no effect once the group has a committed offset.
💻 Code example
spring: kafka: consumer: bootstrap-servers: localhost:9092 group-id: library-events-listener-group auto-offset-reset: earliest key-deserializer: org.apache.kafka.common.serialization.StringDeserializer value-deserializer: org.apache.kafka.common.serialization.StringDeserializer
Start two instances of the consumer service with the same group.id against the 3-partition library-events topic, and Kafka assigns each instance roughly half the partitions. Kill one instance and watch the logs of the survivor — you'll see a rebalance log line and then the survivor picking up the partitions the dead instance used to own, exactly as described conceptually in Module 03 §4.
| Default (auto-commit / after-listener) | Manual acknowledgment | |
|---|---|---|
| When offset commits | Automatically, after your listener method returns successfully | Only when you explicitly call acknowledgment.acknowledge() |
| Control | Low — you can't defer a commit past "method returned" | High — commit only after e.g. a DB write actually succeeds |
| Risk if misused | Committing before slow downstream work finishes (if you spin off async work) can lose track of failures | Forgetting to call acknowledge() at all stalls the partition — no offset ever advances |
@KafkaListener(topics = "library-events") public void onMessage(ConsumerRecord<String, String> record, Acknowledgment acknowledgment) { libraryEventsService.processLibraryEvent(record); // throws on failure acknowledgment.acknowledge(); // only commit if processing truly succeeded }
spring: kafka: listener: ack-mode: manual
◆ Under the hood
Whichever mode you use, this is only ever at-least-once delivery, not exactly-once — a crash between "DB write succeeded" and "offset commit sent" replays the record on restart, and your processing logic must be idempotent (or you need transactions, Module 16) to avoid duplicate effects.
💻 Code example
@KafkaListener(topics = "library-events") public void onMessage(ConsumerRecord<String, String> record, Acknowledgment acknowledgment) { libraryEventsService.processLibraryEvent(record); // throws on failure acknowledgment.acknowledge(); // only commit if processing truly succeeded }
The concurrency property spins up multiple listener container threads within a single application instance, each behaving like a separate consumer within the same group — directly exploiting the partition-parallelism model from Module 03 §4 without needing multiple deployed instances.
spring: kafka: listener: concurrency: 3 # matches the topic's 3 partitions — one thread per partition
▲ Pitfall
Setting concurrency higher than the topic's partition count doesn't increase throughput — the extra threads simply have no partition to be assigned and sit idle, identical to the "consumers > partitions" row in Module 03's table.
✓ Quick recap
When does auto-offset-reset take effect? Only the first time a consumer group connects with no prior committed offset (or after that offset has expired past retention) — it has no effect afterward. What's the actual delivery guarantee with manual acknowledgment alone? Still at-least-once — a crash between processing and commit can replay a record. Why doesn't concurrency=6 help on a 3-partition topic? A partition can only be assigned to one consumer thread within a group at a time — the extra 3 threads have nothing to consume.
💻 Code example
spring: kafka: listener: concurrency: 3 # matches the topic's 3 partitions — one thread per partition
Want a visual for this concept?
Generate a diagram tailored to “Building the Consumer Microservice” — the AI picks whichever visual (flowchart, comparison, sequence, etc.) best fits.
Sign in to generate a visual →