Definition
This coding question combines several Spring Kafka features into one cohesive banking service.
- JSON producer keyed by userId – KafkaTemplate sends events as JSON with userId as the key for per-user partition affinity.
- MANUAL acknowledgment – the container does not auto-commit; you call
Acknowledgment.acknowledge()after successful processing. - Bounded retries + DLT – DefaultErrorHandler retries with backoff and routes exhausted records to a dead letter topic.
- Batch-mode listener – the listener receives a
Listof records from one poll for bulk processing. - Pause/resume – KafkaListenerEndpointRegistry pauses the container when the database is down and resumes on recovery.
At National Bank of India this is the full mailroom wired together. The producer addresses each letter by the customer's ID so all of Rajesh's letters go to one sorter. The consumer only moves the bookmark after filing the letter. The batch sorter grabs a whole tray for speed. And when the filing room (database) is closed, the supervisor tells the sorter to pause so letters wait safely.
Understanding the Concepts
How do you write a JSON producer keyed by userId?
The producer uses a JSON serializer and passes userId as the key so Kafka hashes it to a consistent partition. All of one user's events stay ordered.
package com.bank.kafka;
import com.bank.model.UserEvent;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.support.SendResult;
import org.springframework.stereotype.Service;
import java.math.BigDecimal;
import java.util.concurrent.CompletableFuture;
@Service
public class UserEventProducer {
private final KafkaTemplate<String, UserEvent> kafkaTemplate;
public UserEventProducer(KafkaTemplate<String, UserEvent> kafkaTemplate) {
this.kafkaTemplate = kafkaTemplate;
}
public void publishTransfer(String userId, BigDecimal amount) {
UserEvent event = new UserEvent(userId, "TRANSFER", amount);
CompletableFuture<SendResult<String, UserEvent>> future =
kafkaTemplate.send("user.events", userId, event);
future.whenComplete((result, ex) -> {
if (ex == null) {
System.out.println("Sent to partition "
+ result.getRecordMetadata().partition()
+ " offset " + result.getRecordMetadata().offset());
} else {
System.err.println("Send failed: " + ex.getMessage());
}
});
}
}
Sent to partition 3 offset 42
How do you write a manual-ack consumer with bounded retries and DLT?
The consumer uses MANUAL acknowledgment. The DefaultErrorHandler retries transient errors and routes exhausted records to the DLT. The listener acknowledges only on success.
package com.bank.kafka;
import com.bank.model.UserEvent;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.stereotype.Component;
@Component
public class ManualAckConsumer {
@KafkaListener(id = "settlement", topics = "user.events",
groupId = "settlement-group", containerFactory = "manualAckFactory")
public void onUserEvent(UserEvent event, Acknowledgment ack) {
System.out.println("Processing: " + event.getUserId()
+ " amount=" + event.getAmount());
// Business logic -- may throw transient or permanent exceptions
settlementService.process(event);
// Acknowledge ONLY after processing succeeds
ack.acknowledge();
System.out.println("Acknowledged: " + event.getUserId());
}
}
Processing: SAV2001 amount=50000
Acknowledged: SAV2001
How do you configure the manual-ack factory with error handler?
package com.bank.kafka;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.kafka.config.ConcurrentKafkaListenerContainerFactory;
import org.springframework.kafka.core.ConsumerFactory;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.listener.ContainerProperties;
import org.springframework.kafka.listener.DefaultErrorHandler;
import org.springframework.kafka.listener.DeadLetterPublishingRecoverer;
import org.springframework.kafka.support.serializer.DeserializationException;
import org.springframework.util.backoff.FixedBackOff;
@Configuration
public class ManualAckFactoryConfig {
@Bean
public ConcurrentKafkaListenerContainerFactory<String, Object> manualAckFactory(
ConsumerFactory<String, Object> consumerFactory,
KafkaTemplate<String, Object> template) {
DefaultErrorHandler errorHandler = new DefaultErrorHandler(
new DeadLetterPublishingRecoverer(template),
new FixedBackOff(1000L, 3L) // retry 3 times, 1s apart
);
errorHandler.addNotRetryableExceptions(DeserializationException.class);
ConcurrentKafkaListenerContainerFactory<String, Object> factory =
new ConcurrentKafkaListenerContainerFactory<>();
factory.setConsumerFactory(consumerFactory);
factory.getContainerProperties().setAckMode(ContainerProperties.AckMode.MANUAL);
factory.setCommonErrorHandler(errorHandler);
factory.setConcurrency(3);
return factory;
}
}
# Transient DB error: retry 1s -> 1s -> 1s -> DLT
# Deserialization error: skip retries -> DLT immediately
# Success: ack.acknowledge() -> offset advances
How do you write a batch-mode listener?
Batch mode hands the whole list of records from one poll to your method. You process in bulk and acknowledge once.
package com.bank.kafka;
import com.bank.model.UserEvent;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.stereotype.Component;
import java.util.List;
@Component
public class BatchSettlementListener {
@KafkaListener(id = "batch-settlement", topics = "user.events",
groupId = "batch-settlement-group", containerFactory = "batchFactory")
public void onBatch(List<UserEvent> events, Acknowledgment ack) {
System.out.println("Batch received: " + events.size() + " events");
// Bulk insert for efficiency
settlementRepository.saveAll(events);
ack.acknowledge(); // one commit for entire batch
System.out.println("Batch acknowledged");
}
}
Batch received: 50 events
Batch acknowledged
How do you configure the batch factory?
package com.bank.kafka;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.kafka.config.ConcurrentKafkaListenerContainerFactory;
import org.springframework.kafka.core.ConsumerFactory;
import org.springframework.kafka.listener.ContainerProperties;
@Configuration
public class BatchFactoryConfig {
@Bean
public ConcurrentKafkaListenerContainerFactory<String, Object> batchFactory(
ConsumerFactory<String, Object> consumerFactory) {
ConcurrentKafkaListenerContainerFactory<String, Object> factory =
new ConcurrentKafkaListenerContainerFactory<>();
factory.setConsumerFactory(consumerFactory);
factory.setBatchListener(true); // enable batch mode
factory.getContainerProperties().setAckMode(ContainerProperties.AckMode.MANUAL);
factory.setConcurrency(3);
return factory;
}
}
# Batch mode: listener receives List<UserEvent> from one poll()
# One commit for the whole batch: efficient for bulk writes
How do you pause and resume on downstream failure?
Inject KafkaListenerEndpointRegistry, check DB health on a schedule, and call pause/resume.
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 DownstreamHealthMonitor {
private final KafkaListenerEndpointRegistry registry;
private final DatabaseHealthChecker dbHealth;
public DownstreamHealthMonitor(KafkaListenerEndpointRegistry registry,
DatabaseHealthChecker dbHealth) {
this.registry = registry;
this.dbHealth = dbHealth;
}
@Scheduled(fixedDelay = 5000)
public void monitor() {
controlListener("settlement");
controlListener("batch-settlement");
}
private void controlListener(String id) {
MessageListenerContainer container = registry.getListenerContainer(id);
if (container == null) return;
if (!dbHealth.isHealthy() && !container.isContainerPaused()) {
container.pause();
System.out.println("Paused " + id + " -- DB down");
} else if (dbHealth.isHealthy() && container.isContainerPaused()) {
container.resume();
System.out.println("Resumed " + id + " -- DB recovered");
}
}
}
Paused settlement -- DB down
Paused batch-settlement -- DB down
Resumed settlement -- DB recovered
Resumed batch-settlement -- DB recovered
What is the UserEvent model?
package com.bank.model;
import java.math.BigDecimal;
public class UserEvent {
private String userId;
private String type;
private BigDecimal amount;
public UserEvent() {}
public UserEvent(String userId, String type, BigDecimal amount) {
this.userId = userId;
this.type = type;
this.amount = amount;
}
public String getUserId() { return userId; }
public void setUserId(String userId) { this.userId = userId; }
public String getType() { return type; }
public void setType(String type) { this.type = type; }
public BigDecimal getAmount() { return amount; }
public void setAmount(BigDecimal amount) { this.amount = amount; }
}
UserEvent{userId=SAV2001, type=TRANSFER, amount=50000}
What should a senior engineer know?
Keying by userId concentrates a hot user on one partition; mitigate with salting if needed. Manual-ack must always acknowledge on both the success and the DLT paths or the partition stalls. Batch mode needs partial-failure handling: filter bad records to the DLT and process the rest. Pausing is healthier than letting a listener fail repeatedly against a dead dependency.
# Senior checklist:
# 1. Key by userId -> consistent partition -> ordered per user
# 2. MANUAL ack -> offset advances only after success
# 3. Bounded retries + DLT -> poison pills handled
# 4. Batch mode -> bulk DB write -> one commit
# 5. Pause/resume -> downstream-aware flow control
# 6. Idempotent processing -> at-least-once is safe
Visual Overview
Full Spring Kafka Service Architecture
Producer:
UserEventProducer ──send(userId, event)──→ user.events topic
Key = userId → consistent partition → per-user ordering
Single-Record Consumer (manual ack + DLT):
user.events ──→ ManualAckConsumer
│
├── success → ack.acknowledge() → offset advances
└── failure → DefaultErrorHandler
│
├── retryable → retry 3x with backoff
│ │
│ exhausted → DLT
└── non-retryable → DLT immediately
Batch Consumer:
user.events ──→ BatchSettlementListener
│
└── List<UserEvent> → bulk DB write → ack.acknowledge()
Pause/Resume:
┌──────────────┐ DB down? ┌──────────────────┐
│ Health Check │── yes ──────→│ container.pause() │
│ (scheduled) │── no ───────→│ container.resume()│
└──────────────┘ └──────────────────┘
Key Takeaway
This coding question wires the whole Spring Kafka toolkit into one coherent banking service. The producer publishes JSON events keyed by userId so Kafka routes every event for one user to the same partition, preserving per-user ordering. The single-record consumer uses MANUAL acknowledgment to commit the offset only after processing succeeds, with bounded retries and a DLT for records that exhaust attempts.
The batch-mode listener receives the whole poll batch as a list for efficient bulk database writes with one commit. Pause and resume provide downstream-aware flow control: when the health checker detects the database is down, it pauses both listeners, freezing offsets and holding the consumers in their groups without rebalance, then resumes when the database recovers.
At National Bank of India, this exact design handled a transient database blip on Rajesh's transfer with a quick retry, parked a corrupt event in the DLT without stalling, bulk-inserted a batch efficiently, and sailed through a database maintenance window by pausing and resuming with nothing lost.
Remember one sentence: key by userId for ordering, MANUAL ack for safety, bounded retries for resilience, batch mode for throughput, and pause/resume for downstream-aware flow control.
---