java 44 lines · 8 steps

Wiring a resilient Kafka consumer in Spring

A Spring @Configuration class builds a fault-tolerant Kafka consumer for order events with JSON deserialization and safe error handling.

Explained by highlit
Spring Intermediate kafka deserialization error-handling messaging configuration
1@Configuration
2public class OrderConsumerConfig {
3 
4 @Bean
5 public ConsumerFactory<String, OrderEvent> orderConsumerFactory(KafkaProperties properties) {
6 Map<String, Object> props = new HashMap<>(properties.buildConsumerProperties());
7 props.put(ConsumerConfig.GROUP_ID_CONFIG, "order-processing");
8 props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
9 props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
10 props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, ErrorHandlingDeserializer.class);
11 props.put(ErrorHandlingDeserializer.VALUE_DESERIALIZER_CLASS, JsonDeserializer.class);
12 
13 JsonDeserializer<OrderEvent> valueDeserializer = new JsonDeserializer<>(OrderEvent.class);
14 valueDeserializer.setRemoveTypeHeaders(false);
15 valueDeserializer.addTrustedPackages("com.acme.orders.events");
16 
17 return new DefaultKafkaConsumerFactory<>(
18 props,
19 new StringDeserializer(),
20 new ErrorHandlingDeserializer<>(valueDeserializer));
21 }
22 
23 @Bean
24 public ConcurrentKafkaListenerContainerFactory<String, OrderEvent> orderKafkaListenerContainerFactory(
25 ConsumerFactory<String, OrderEvent> orderConsumerFactory) {
26 ConcurrentKafkaListenerContainerFactory<String, OrderEvent> factory =
27 new ConcurrentKafkaListenerContainerFactory<>();
28 factory.setConsumerFactory(orderConsumerFactory);
29 factory.setConcurrency(3);
30 return factory;
31 }
32 
33 @KafkaListener(
34 topics = "orders.placed",
35 groupId = "order-processing",
36 containerFactory = "orderKafkaListenerContainerFactory")
37 public void onOrderPlaced(
38 @Payload OrderEvent event,
39 @Header(KafkaHeaders.RECEIVED_PARTITION) int partition,
40 @Header(KafkaHeaders.RECEIVED_KEY) String key) {
41 log.info("Received order {} from partition {}", event.orderId(), partition);
42 orderService.process(event);
43 }
44}
01 / 01
STEP 01

Walkthrough

Space play ←→ step click any line
Three takeaways
  1. 1Wrapping a deserializer in ErrorHandlingDeserializer stops one poison message from crashing the whole consumer.
  2. 2Trusted packages must be declared explicitly so the JSON deserializer only instantiates types you control.
  3. 3Container concurrency lets a single listener method process multiple partitions in parallel.

Related explainers

go 
package email
 
import (
	"errors"

Normalizing and deduping email addresses in Go

validation normalization deduplication
Intermediate 8 steps
java 
@Configuration
@EnableBatchProcessing
public class CustomerImportJobConfig {
 

How a chunk-based CSV import job works in Spring

batch-processing etl csv-parsing
Intermediate 9 steps
javascript 
const express = require('express');
const multer = require('multer');
const path = require('path');
const crypto = require('crypto');

Safe image uploads with Multer in Express

file-upload multer validation
Intermediate 7 steps
java 
@Configuration
@EnableWebSecurity
public class ResourceServerConfig {
 

Configuring a JWT resource server in Spring

oauth2 jwt authorization
Intermediate 8 steps
rust 
use std::collections::VecDeque;
 
#[derive(Debug)]
pub struct Hunk {

Applying a diff hunk in Rust

enums error-handling pattern-matching
Intermediate 8 steps
go 
package config
 
import (
	"fmt"

Parsing timeout config in Go

configuration validation error-wrapping
Intermediate 7 steps

Share this explainer

Here's the card — post it anywhere.

Wiring a resilient Kafka consumer in Spring — share card
Made with highlit — turn any snippet into a walkthrough like this in about a minute.
Explain your code