Blog

Spring Boot Kafka Consumer JSON Example

Key Takeaways

  • Configure a fixed JSON target type so Kafka record values become User objects without requiring producer-side Java type headers.
  • ErrorHandlingDeserializer exposes deserialization failures to the listener container; it does not define a recovery policy by itself.
  • Combine DefaultErrorHandler with DeadLetterPublishingRecoverer to give failed records an explicit destination.
  • Test malformed JSON, dead-letter publication, delivery of the following record, and committed offsets—not just successful deserialization.

A Spring Boot Kafka consumer JSON configuration needs two separate decisions: how bytes become Java objects, and what happens when that conversion fails. This example consumes JSON into a User record, sends malformed values to a dead-letter topic, and continues only after successful recovery.

If you need the basic consumer lifecycle first, see our Spring Boot Kafka consumer guide.

Spring Boot Kafka Consumer JSON Example

Versions and project layout

This example pins Spring Boot to 4.1.1 and compiles with Java 21. Spring Boot 4.1.1 manages Spring Kafka 4.1.1 and Kafka clients 4.2.1; Boot itself requires at least Java 17. Use Maven 3.6.3 or later and a local Kafka 4.2 broker available at localhost:9092.

pom.xml
src/main/java/com/example/demo/DemoApplication.java
src/main/java/com/example/demo/model/User.java
src/main/java/com/example/demo/config/KafkaConfig.java
src/main/java/com/example/demo/messaging/UserConsumer.java
src/main/resources/application.properties
src/test/java/com/example/demo/UserConsumerTest.java

pom.xml

The Boot parent manages dependency versions. Include Jackson explicitly for JSON mapping; the Kafka test starter brings in Spring Kafka's test utilities and Boot's general test starter.

<project>
  <modelVersion>4.0.0</modelVersion>
  <parent>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-parent</artifactId>
    <version>4.1.1</version>
    <relativePath/>
  </parent>
  <groupId>com.example</groupId>
  <artifactId>kafka-json-demo</artifactId>
  <version>1.0.0</version>
  <properties>
    <java.version>21</java.version>
  </properties>
  <dependencies>
    <dependency>
      <groupId>org.springframework.boot</groupId>
      <artifactId>spring-boot-starter-kafka</artifactId>
    </dependency>
    <dependency>
      <groupId>org.springframework.boot</groupId>
      <artifactId>spring-boot-starter-jackson</artifactId>
    </dependency>
    <dependency>
      <groupId>org.springframework.boot</groupId>
      <artifactId>spring-boot-starter-kafka-test</artifactId>
      <scope>test</scope>
    </dependency>
  </dependencies>
  <build>
    <plugins>
      <plugin>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-maven-plugin</artifactId>
      </plugin>
    </plugins>
  </build>
</project>

application.properties

spring.kafka.bootstrap-servers=${KAFKA_BOOTSTRAP_SERVERS:localhost:9092}
spring.kafka.consumer.group-id=myGroup
spring.kafka.consumer.auto-offset-reset=earliest
spring.kafka.consumer.enable-auto-commit=false
spring.kafka.consumer.key-deserializer=org.apache.kafka.common.serialization.StringDeserializer
spring.kafka.consumer.value-deserializer=org.springframework.kafka.support.serializer.ErrorHandlingDeserializer
spring.kafka.consumer.properties.spring.deserializer.value.delegate.class=org.springframework.kafka.support.serializer.JacksonJsonDeserializer
spring.kafka.consumer.properties.spring.json.value.default.type=com.example.demo.model.User
spring.kafka.consumer.properties.spring.json.use.type.headers=false
spring.kafka.consumer.properties.spring.json.trusted.packages=com.example.demo.model
spring.kafka.listener.ack-mode=record
spring.kafka.listener.concurrency=1
spring.kafka.producer.acks=all

The value deserializer is the error-handling wrapper; its delegate performs JSON mapping. The fixed target type is com.example.demo.model.User, and spring.json.use.type.headers=false prevents producer type headers from selecting a different class. These are consumer properties, not settings inferred from the listener parameter.

This policy works with producers written in other languages or using different Java packages: they send the agreed JSON shape, not a consumer-side class name. Keep trusted packages narrow rather than using * as a workaround; with type headers disabled, the configured target type determines mapping. Use either property-based deserializer configuration or programmatic setters, not a mixture.

Application entry point and User model

Create DemoApplication.java:

package com.example.demo;

import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;

@SpringBootApplication
public class DemoApplication {
    public static void main(String[] args) {
        SpringApplication.run(DemoApplication.class, args);
    }
}

Create model/User.java. A Java record supplies accessors, value-based equality, and a useful toString() without boilerplate.

package com.example.demo.model;

public record User(String firstName, String lastName) {}

KafkaConfig.java: configure dead-letter recovery

Let Boot create the consumer factory and the default kafkaListenerContainerFactory from spring.kafka.*. Boot automatically attaches a CommonErrorHandler bean to that factory, so a separate @EnableKafka or hand-built consumer factory isn't needed here.

package com.example.demo.config;

import java.util.Map;

import com.example.demo.model.User;
import org.apache.kafka.clients.admin.NewTopic;
import org.apache.kafka.common.TopicPartition;
import org.apache.kafka.common.serialization.ByteArraySerializer;
import org.apache.kafka.common.serialization.StringSerializer;
import org.springframework.boot.kafka.autoconfigure.KafkaProperties;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.kafka.config.TopicBuilder;
import org.springframework.kafka.core.DefaultKafkaProducerFactory;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.core.ProducerFactory;
import org.springframework.kafka.listener.DeadLetterPublishingRecoverer;
import org.springframework.kafka.listener.DefaultErrorHandler;
import org.springframework.kafka.support.serializer.DelegatingByTypeSerializer;
import org.springframework.kafka.support.serializer.JacksonJsonSerializer;
import org.springframework.util.backoff.FixedBackOff;

@Configuration(proxyBeanMethods = false)
public class KafkaConfig {
    @Bean
    NewTopic inputTopic() {
        return TopicBuilder.name("test").partitions(1).replicas(1).build();
    }

    @Bean
    NewTopic deadLetterTopic() {
        return TopicBuilder.name("test-dlt").partitions(1).replicas(1).build();
    }

    @Bean
    ProducerFactory<String, Object> producerFactory(KafkaProperties properties) {
        var serializer = new DelegatingByTypeSerializer(Map.of(
                byte[].class, new ByteArraySerializer(),
                User.class, new JacksonJsonSerializer<User>()));
        return new DefaultKafkaProducerFactory<>(
                properties.buildProducerProperties(),
                new StringSerializer(), serializer);
    }

    @Bean
    KafkaTemplate<String, Object> kafkaTemplate(
            ProducerFactory<String, Object> producerFactory) {
        return new KafkaTemplate<>(producerFactory);
    }

    @Bean
    DefaultErrorHandler errorHandler(KafkaTemplate<String, Object> template) {
        var recoverer = new DeadLetterPublishingRecoverer(template,
                (record, exception) -> new TopicPartition(
                        record.topic() + "-dlt", record.partition()));
        recoverer.setFailIfSendResultIsError(true);

        var handler = new DefaultErrorHandler(
                recoverer, new FixedBackOff(1000L, 2L));
        handler.setAckAfterHandle(true);
        return handler;
    }
}

The custom producer factory starts with KafkaProperties.buildProducerProperties(), preserving the configured broker and producer settings. Its serializer handles both User values and raw byte[]: deserialization failures are republished with their original bytes, while listener-processing failures still have a deserialized object. Sending those bytes through a JSON serializer would change their representation.

setFailIfSendResultIsError(true) makes recovery wait for the publish result and throw if publication fails. That prevents a failed dead-letter send from being treated as successful recovery.

UserConsumer.java: receive typed records

package com.example.demo.messaging;

import com.example.demo.model.User;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Component;

@Component
public class UserConsumer {
    private static final Logger log = LoggerFactory.getLogger(UserConsumer.class);

    @KafkaListener(topics = "test",
            groupId = "${spring.kafka.consumer.group-id}",
            containerFactory = "kafkaListenerContainerFactory")
    public void processMessage(User user) {
        log.info("Received {}", user);
    }
}

The listener explicitly selects Boot's default factory and the configured group. Its parameter receives an already-deserialized User; malformed values take the container error-handling path instead. For annotation options, see our Spring Boot KafkaListener example.

Publishing JSON messages from a file

Start your local broker first, then run mvn spring-boot:run from the project directory. The NewTopic beans provision the demo topics; existing topics are left in place.

Run the following from the Kafka distribution directory. The console producer sends one record per input line, so keep each JSON object on one line—not pretty-printed across several lines.

printf '%s\n' '{"firstName":"Ada","lastName":"Lovelace"}' > testUser.json

bin/kafka-console-producer.sh \
  --bootstrap-server localhost:9092 \
  --topic test < testUser.json

The application log includes:

Received User[firstName=Ada, lastName=Lovelace]

Now send malformed JSON followed by a valid record:

printf '%s\n' \
  '{"firstName":' \
  '{"firstName":"Grace","lastName":"Hopper"}' |
  bin/kafka-console-producer.sh \
    --bootstrap-server localhost:9092 \
    --topic test

Why malformed JSON can block a Kafka offset

A poison pill is a record that repeatedly fails consumption or processing. Without the wrapper, deserialization can fail before poll() returns, before your listener can catch anything. An error from the older configuration might look like this; exact wording varies by client version.

Caused by: org.apache.kafka.common.errors.SerializationException: Error deserializing key/value for partition test-0 at offset 87. If needed, please seek past the record to continue consumption.

ErrorHandlingDeserializer captures the failure, returns a null value, and adds a deserialization-exception header containing the cause and raw data. For this record listener, the container detects that header and invokes its error handler instead of your listener.

Kafka record
  |
  v
ErrorHandlingDeserializer -> JacksonJsonDeserializer<User>
  | success                         | failure header
  v                                 v
UserConsumer                 DefaultErrorHandler
  |                          | fatal / retries exhausted
  |                          v
  |                       test-dlt
  |                          | publish succeeds
  +--------------------------+
                 |
                 v
       commit progress; continue

DLT publication fails -> recovery throws -> record remains eligible for redelivery

Deserialization failures versus listener failures

FailureDoes the listener run?Policy in this example
Malformed JSON or incompatible JSON mappingNoThe resulting DeserializationException is classified as fatal by DefaultErrorHandler and goes directly to recovery.
Retryable business-processing exceptionYesTwo retries with a one-second backoff: three listener attempts total, then recovery.

The backoff does not override the handler's fatal-exception classifications. If the recoverer throws, the failed record is included in the error handler's seeks and can be delivered again; installing the wrapper alone is therefore not a guarantee of forward progress.

With automatic commits disabled, ack-mode=record commits progress after successful listener processing. For recovered records, setAckAfterHandle(true) allows the container to commit progress when the handler returns successfully. This is container-managed acknowledgment, not a requirement to call acknowledge() yourself.

Offsets are partition-specific; our Kafka architecture explanation covers how partitions and consumer groups fit together.

Verify dead-letter publication and committed offsets

In another terminal, inspect the dead-letter topic:

bin/kafka-console-consumer.sh \
  --bootstrap-server localhost:9092 \
  --topic test-dlt --from-beginning

You should see the malformed value, while the application logs Grace's valid record. Check the source group's committed progress separately:

bin/kafka-consumer-groups.sh \
  --bootstrap-server localhost:9092 \
  --describe --group myGroup

On fresh demo topics with no other producers, the sequence writes three input records. After processing completes, CURRENT-OFFSET should reach 3 and LAG should reach 0. The committed offset identifies where consumption resumes, not the last processed record's offset.

Test valid JSON and poison-record recovery

This integration test uses an embedded broker, the application's actual deserializers and recovery beans, and a spy around the listener. It checks both valid deliveries, the exact failed bytes on the DLT, and committed progress beyond the final record. The Kafka test starter supplies the embedded-broker utilities.

package com.example.demo;

import java.nio.charset.StandardCharsets;
import java.time.Duration;
import java.util.concurrent.TimeUnit;

import com.example.demo.messaging.UserConsumer;
import com.example.demo.model.User;
import org.apache.kafka.common.serialization.ByteArrayDeserializer;
import org.apache.kafka.common.serialization.StringDeserializer;
import org.junit.jupiter.api.Test;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.test.context.SpringBootTest;
import org.springframework.kafka.core.DefaultKafkaConsumerFactory;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.test.EmbeddedKafkaBroker;
import org.springframework.kafka.test.context.EmbeddedKafka;
import org.springframework.kafka.test.utils.KafkaTestUtils;
import org.springframework.test.annotation.DirtiesContext;
import org.springframework.test.context.bean.override.mockito.MockitoSpyBean;

import static org.assertj.core.api.Assertions.assertThat;
import static org.awaitility.Awaitility.await;
import static org.mockito.Mockito.timeout;
import static org.mockito.Mockito.verify;

@SpringBootTest
@DirtiesContext
@EmbeddedKafka(partitions = 1, topics = {"test", "test-dlt"},
        bootstrapServersProperty = "spring.kafka.bootstrap-servers")
class UserConsumerTest {
    @Autowired EmbeddedKafkaBroker broker;
    @Autowired KafkaTemplate<String, Object> template;
    @MockitoSpyBean UserConsumer listener;

    @Test
    void recoversMalformedJsonAndContinues() throws Exception {
        var properties = KafkaTestUtils.consumerProps("dlt-check", "false", broker);
        var factory = new DefaultKafkaConsumerFactory<String, byte[]>(
                properties, new StringDeserializer(), new ByteArrayDeserializer());
        byte[] bad = "{\"firstName\":".getBytes(StandardCharsets.UTF_8);
        var ada = new User("Ada", "Lovelace");
        var grace = new User("Grace", "Hopper");

        try (var dltConsumer = factory.createConsumer()) {
            broker.consumeFromAnEmbeddedTopic(dltConsumer, "test-dlt");

            template.send("test", 0, "before", ada).get(10, TimeUnit.SECONDS);
            template.send("test", 0, "bad", bad).get(10, TimeUnit.SECONDS);
            long lastOffset = template.send("test", 0, "after", grace)
                    .get(10, TimeUnit.SECONDS).getRecordMetadata().offset();

            verify(listener, timeout(10000)).processMessage(ada);
            verify(listener, timeout(10000)).processMessage(grace);

            var dead = KafkaTestUtils.getSingleRecord(
                    dltConsumer, "test-dlt", Duration.ofSeconds(10));
            assertThat(dead.value()).containsExactly(bad);

            await().atMost(Duration.ofSeconds(10)).untilAsserted(() -> {
                var committed = KafkaTestUtils.getCurrentOffset(
                        broker.getBrokersAsString(), "myGroup", "test", 0);
                assertThat(committed).isNotNull();
                assertThat(committed.offset()).isEqualTo(lastOffset + 1);
            });
        }
    }
}

Run mvn test. Extend this suite with retryable listener failures and failed DLT publication before relying on the policy in production. For more patterns, see our Spring Kafka testing examples.

Production Checklist

  • Define the JSON contract. Mapping JSON into a record is not JSON Schema validation. Specify required fields, null handling, and compatibility rules; this minimal model does not enforce non-null names.
  • Provision the DLT deliberately. This resolver preserves the original partition number, so provision at least as many DLT partitions as source partitions. The single replica shown here is for local development; choose replication, retention, and permissions for your production durability requirements.
  • Make recovery observable. Count deserialization failures, retry attempts, successful recoveries, and recovery failures. Use error-handler RetryListener callbacks where appropriate; alert on DLT growth and stalled committed offsets, not just process uptime. Avoid logging sensitive payloads.
  • Plan inspection and replay. Dead-letter records include exception and original-record metadata headers. Inspect the raw value with byte- or string-based tooling, fix the producer or mapping issue, and replay through a controlled workflow rather than automatically cycling failures back into the input topic.
  • Expect duplicates. This nontransactional example publishes to the DLT and commits source offsets separately. A crash between those operations can duplicate a DLT record; make replay and business side effects idempotent. Kafka transaction-based recovery requires a different transaction-aware configuration.
  • Budget blocking work. Keep listener processing, retries, and recovery waits within your consumer's poll-time budget. Long gaps between polls can cause the consumer to leave its group.

FAQ

How do I consume JSON as a Java object with Spring Kafka?

Configure JacksonJsonDeserializer with the expected Java target type and use that type as your KafkaListener parameter. Wrap the delegate in ErrorHandlingDeserializer when you want deserialization failures routed to the container's error handler.

Does ErrorHandlingDeserializer alone skip a bad Kafka message?

No. It captures the deserialization failure and exposes it to the listener container; the container's error handler determines retry and recovery behavior. Configure an explicit recovery action rather than assuming the wrapper guarantees progress.

How do I send deserialization failures to a dead-letter topic?

Configure DefaultErrorHandler with a DeadLetterPublishingRecoverer and a producer serializer that can publish raw byte arrays. Deserialization failures are fatal by default for this handler, so they go directly to recovery rather than consuming its retry budget.

How should I configure trusted packages and type headers?

For a topic containing one known payload type, configure a fixed target class and disable type-header lookup. If you intentionally use producer type headers, keep trusted packages narrow and use explicit type mappings when producer and consumer Java class names differ.

How do I handle multiple consumer groups?

Create separate listener endpoints with distinct groupId values when independent subscribers need their own consumption progress. They can share a container factory if their deserialization and recovery policies match; separate ConsumerFactory beans are not required solely to change the group.

Further Reading

Written by Sam Erickson

Software engineer. I write every article on StackChief myself. Author of Mastering the Tech Interview. More about me

Related reading