Definition
Consuming JSON and controlling listener lifecycle are two essential Spring Kafka skills.
- JsonDeserializer – Spring Kafka's Jackson-based deserializer that reconstructs JSON bytes into a Java object.
- Trusted packages – a security allowlist of packages the deserializer is permitted to instantiate; prevents a malicious type header from creating dangerous objects.
- ErrorHandlingDeserializer – a wrapper that catches deserialization exceptions and converts them into recoverable errors instead of blocking the partition.
- KafkaListenerEndpointRegistry – Spring's registry of all listener containers, used to pause and resume individual listeners at runtime.
- Pause/resume – pausing stops record delivery but keeps the consumer alive, holding partitions and heartbeating to avoid rebalance.
A Kafka message arrives at National Bank of India as a sealed envelope of bytes. The JsonDeserializer is the mail clerk who opens the envelope and reconstructs the actual transfer form inside. Trusted packages is the rule that the clerk only accepts forms from approved departments, never a forged form from an unknown source. Pause and resume is the manager telling the clerk to stop opening mail while the vault computer is down and to start again once it recovers, so no form is lost and nothing piles up half-processed.
Understanding the Concepts
What is the default deserializer in Kafka?
Kafka itself is type-agnostic and moves bytes. The raw client's default is ByteArrayDeserializer, and a typical Spring Boot setup defaults to StringDeserializer, giving you JSON as a plain string that you parse yourself.
package com.bank.kafka;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Component;
@Component
public class StringDeserializerListener {
// With StringDeserializer, the value is a raw JSON string
@KafkaListener(topics = "settlements", groupId = "settlement-group")
public void onSettlement(ConsumerRecord<String, String> record) {
String jsonString = record.value();
System.out.println("Raw JSON: " + jsonString);
// You must parse this yourself with ObjectMapper
}
}
Raw JSON: {"txnId":"TXN-1","accountId":"SAV2001","amount":50000}
How do you configure JsonDeserializer for typed consumption?
To get a strongly typed object directly, set the value deserializer to Spring Kafka's JsonDeserializer. It uses Jackson under the hood to reconstruct the payload into your target type.
# application.properties
spring.kafka.consumer.value-deserializer=org.springframework.kafka.support.serializer.JsonDeserializer
spring.kafka.consumer.properties.spring.json.value.default.type=com.bank.model.TxnEvent
spring.kafka.consumer.properties.spring.json.trusted.packages=com.bank.model
package com.bank.model;
import java.math.BigDecimal;
public class TxnEvent {
private String txnId;
private String accountId;
private BigDecimal amount;
// getters, setters, toString
public String getTxnId() { return txnId; }
public void setTxnId(String txnId) { this.txnId = txnId; }
public String getAccountId() { return accountId; }
public void setAccountId(String accountId) { this.accountId = accountId; }
public BigDecimal getAmount() { return amount; }
public void setAmount(BigDecimal amount) { this.amount = amount; }
@Override
public String toString() {
return "TxnEvent{txnId=" + txnId + ", account=" + accountId + ", amount=" + amount + "}";
}
}
package com.bank.kafka;
import com.bank.model.TxnEvent;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Component;
@Component
public class TypedJsonListener {
@KafkaListener(topics = "settlements", groupId = "settlement-group")
public void onSettlement(TxnEvent event) {
// Jackson already deserialized the JSON into TxnEvent
System.out.println("Typed event: " + event);
System.out.println("Account: " + event.getAccountId()
+ " Amount: " + event.getAmount());
}
}
Typed event: TxnEvent{txnId=TXN-1, account=SAV2001, amount=50000}
Account: SAV2001 Amount: 50000
Why are trusted packages important for security?
By default JsonDeserializer refuses to instantiate arbitrary classes named in a type header. A malicious producer could stamp a dangerous type header and trick the deserializer into building an unexpected class. Trusted packages is an allowlist. Setting it to * trusts everything and is a security risk in production.
# CORRECT: only trust your own model package
spring.kafka.consumer.properties.spring.json.trusted.packages=com.bank.model
# WRONG: trusts everything -- security risk in production
# spring.kafka.consumer.properties.spring.json.trusted.packages=*
How do you wrap JsonDeserializer with ErrorHandlingDeserializer?
If a JSON payload is malformed, the raw JsonDeserializer throws on the poll thread, blocking the partition in an infinite retry loop (the poison pill). ErrorHandlingDeserializer catches the exception and converts it into a recoverable error that flows to the error handler.
# application.properties -- wrap with ErrorHandlingDeserializer
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.JsonDeserializer
spring.kafka.consumer.properties.spring.json.value.default.type=com.bank.model.TxnEvent
spring.kafka.consumer.properties.spring.json.trusted.packages=com.bank.model
# Malformed JSON payload arrives:
# {"txnId":"TXN-1","accountId":"SAV2001","amount":INVALID}
# Without ErrorHandlingDeserializer: partition stuck forever
# With ErrorHandlingDeserializer: error handler routes to DLT, partition continues
How do you pause a listener at runtime?
Every @KafkaListener can have an id. You inject KafkaListenerEndpointRegistry, look up the container by id, and call pause(). The consumer stays alive, holds its partitions and heartbeats (no rebalance), but stops delivering records.
package com.bank.kafka;
import org.springframework.kafka.config.KafkaListenerEndpointRegistry;
import org.springframework.kafka.listener.MessageListenerContainer;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Component;
@Component
public class ListenerFlowControl {
private final KafkaListenerEndpointRegistry registry;
private final DatabaseHealthChecker dbHealth;
public ListenerFlowControl(KafkaListenerEndpointRegistry registry,
DatabaseHealthChecker dbHealth) {
this.registry = registry;
this.dbHealth = dbHealth;
}
@Scheduled(fixedDelay = 5000)
public void checkAndControl() {
MessageListenerContainer container = registry.getListenerContainer("settlement");
if (container == null) return;
if (!dbHealth.isHealthy() && container.isRunning() && !container.isContainerPaused()) {
container.pause();
System.out.println("Paused settlement listener -- DB is down");
} else if (dbHealth.isHealthy() && container.isContainerPaused()) {
container.resume();
System.out.println("Resumed settlement listener -- DB recovered");
}
}
}
Paused settlement listener -- DB is down
Resumed settlement listener -- DB recovered
Why is pausing better than letting the listener keep failing?
Continued failures would burn through retries, flood the dead letter topic with transient errors, or risk the consumer being kicked out of the group for exceeding max.poll.interval.ms. Pausing cleanly holds position without any of these costs.
# Without pause (DB down):
# Record 1 -> process -> DB error -> retry 1, 2, 3 -> DLT (transient error in DLT!)
# Record 2 -> process -> DB error -> retry 1, 2, 3 -> DLT
# DLT fills with records that would have succeeded if DB was up
# Consumer may exceed max.poll.interval.ms and get kicked
# With pause (DB down):
# Pause listener -> consumer stays in group, no records delivered
# DB recovers -> resume -> pick up from last committed offset
# Zero records lost, zero DLT pollution
How does the listener method receive a typed object with a listener id?
Combining JSON deserialization with a listener id for pause/resume control:
package com.bank.kafka;
import com.bank.model.TxnEvent;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.stereotype.Component;
@Component
public class SettlementJsonListener {
@KafkaListener(id = "settlement", topics = "settlements", groupId = "settlement-group")
public void onSettlement(TxnEvent event, Acknowledgment ack) {
System.out.println("Settlement: " + event.getAccountId()
+ " amount=" + event.getAmount());
settlementService.process(event);
ack.acknowledge();
}
}
Settlement: SAV2001 amount=50000
What should a senior engineer know?
Pausing takes effect after the current poll's records are processed. Drive pause/resume from a health indicator or circuit breaker, not a manual flag. The ErrorHandlingDeserializer is non-negotiable in production: without it, one corrupt record can stall an entire partition. And never use trusted.packages=* in production -- always name your own packages explicitly.
# Production-grade deserializer chain
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.JsonDeserializer
spring.kafka.consumer.properties.spring.json.value.default.type=com.bank.model.TxnEvent
spring.kafka.consumer.properties.spring.json.trusted.packages=com.bank.model
Visual Overview
JSON Deserialization Chain
Kafka bytes ──→ ErrorHandlingDeserializer
│
├── valid JSON ──→ JsonDeserializer ──→ TxnEvent object
│ │
│ ▼
│ @KafkaListener method
│
└── malformed JSON ──→ Error handler ──→ Dead Letter Topic
(partition NOT blocked)
Listener Pause/Resume
┌───────────────┐ DB healthy? ┌──────────────────────┐
│ Health Checker │───── yes ──────→│ container.resume() │
│ (scheduled) │ │ Records delivered │
│ │───── no ───────→│ container.pause() │
│ │ │ Consumer stays alive │
└───────────────┘ │ No rebalance │
│ Offsets frozen │
└──────────────────────┘
Key Takeaway
Kafka stores values as raw bytes, so a deserializer turns those bytes back into an object. For JSON, Spring Kafka's JsonDeserializer uses Jackson to reconstruct the payload directly into a typed object like TxnEvent, configured with the target type and trusted packages as a security allowlist. Always wrap it in ErrorHandlingDeserializer so one malformed payload becomes a recoverable error routed to a dead letter topic instead of blocking the entire partition.
For runtime flow control, pause and resume give you clean downstream-aware consumption. Give the listener an id, inject the endpoint registry, and call pause when a health check detects the database is down. The consumer stays in the group, holds its partitions, keeps heartbeating (no rebalance), and offsets freeze. When the database recovers, resume picks up exactly where you stopped.
In the settlement service at National Bank of India, a health indicator watched the account database. When it went down, it paused the settlement listener so Rajesh's and Priya's events waited safely. When the database came back, it resumed and settled every held transfer in order with nothing lost and zero DLT pollution.
Remember one sentence: JsonDeserializer with trusted packages gives you safe typed consumption, ErrorHandlingDeserializer prevents poison pills, and pause/resume gives you clean downstream-aware flow control.
---