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

rust 
use serde::Deserialize;
 
#[derive(Debug, Deserialize)]
#[serde(untagged)]

Parsing flexible JSON shapes with serde

deserialization enums json
Intermediate 6 steps
ruby 
require "shellwords"
require "open3"
 
module Backup

Building safe shell commands in Ruby

shell-out subprocess command-injection
Intermediate 7 steps
typescript 
import { Module } from '@nestjs/common';
import { ConfigModule } from '@nestjs/config';
import * as Joi from 'joi';
 

Validating env config at boot in NestJS

configuration schema-validation environment-variables
Intermediate 8 steps
java 
@Component
@Converter
public class EncryptedStringConverter implements AttributeConverter<String, String> {
 

Transparent column encryption in Spring & JPA

encryption aes-gcm jpa-converter
Advanced 10 steps
java 
package com.acme.billing.config;
 
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.boot.context.properties.ConfigurationProperties;

Feature-flagged beans with Spring @ConditionalOnProperty

feature-flags conditional-beans strategy-pattern
Intermediate 5 steps
rust 
use chrono::{Duration, NaiveDate};
 
#[derive(Debug)]
pub struct DateRange {

Parsing and iterating date ranges in Rust

error-handling iterators parsing
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