Kafka with Spring for Apache Kafka
Produce keyed JSON records with KafkaTemplate, consume them with @KafkaListener in a consumer group, and send failures to a dead-letter topic.
@Bean NewTopic t() { return TopicBuilder.name("orders").partitions(6).replicas(replicas).build(); }
kafkaTemplate.send(topic, key, value);
@KafkaListener(topics = "orders", groupId = "billing") void on(Event e) { }Add spring-boot-starter-kafka and Spring Boot creates a KafkaTemplate, a KafkaAdmin that creates topics from NewTopic beans, and a listener container factory for @KafkaListener. Point it at the cluster with spring.kafka.bootstrap-servers.
Topics, partitions and keys
A topic is split into partitions, and each partition is an ordered log. The producer picks the partition from the record key, so all records with the same key, say an order id, land in the same partition and are read in order. Records without a key are spread across partitions and have no ordering between them. Partitions are also the unit of parallelism: a topic with six partitions keeps at most six consumers in one group busy; a seventh sits idle.
The replication factor must fit the cluster. A topic with three replicas cannot be created on the single broker that a local Compose file runs, so the example reads it from app.kafka.topic-replicas: 1 in application.properties for local development and 3 in application-prod.properties, where a broker failure must not lose acknowledged records.
Producing
KafkaTemplate.send(topic, key, value) returns a CompletableFuture<SendResult>. The call is asynchronous: it returns before the broker has the record. Attach a callback to log failures, or call join() when the caller must know the write succeeded. Configure the value serialiser as JacksonJsonSerializer, the Jackson 3 serialiser in Spring for Apache Kafka.
kafka.send("orders", String.valueOf(orderId), event)
.whenComplete((result, ex) -> { if (ex != null) log.error("send failed", ex); });Consuming
Every listener with the same groupId shares the partitions of a topic; a different group gets its own copy of every record. By default the container commits offsets once all the records returned by a poll have been processed, so a crash before the commit means those records are delivered again. Set auto-offset-reset=earliest for a new group that should read existing records instead of only new ones.
Errors and poison records
When a listener throws, the default error handler retries the record nine more times with no delay and then logs and skips it. That loses the record silently. Declare a DefaultErrorHandler bean with a DeadLetterPublishingRecoverer and Spring Boot wires it into the container factory, so a record that keeps failing is published to a dead-letter topic with the exception in its headers. A record that cannot even be deserialised needs ErrorHandlingDeserializer around your real deserialiser, otherwise the consumer cannot move past it.
Trust only your own packages when deserialising JSON with type headers, and keep listener work short: if processing a batch takes longer than max.poll.interval.ms, the broker assumes the consumer died and rebalances the group, which causes duplicates.
Example
import java.math.BigDecimal;
import java.util.UUID;
import org.apache.kafka.clients.admin.NewTopic;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.config.TopicBuilder;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.stereotype.Component;
import org.springframework.stereotype.Service;
record OrderPlaced(UUID eventId, long orderId, BigDecimal total) {
}
public class OrderTopics {
NewTopic ordersTopic(("${app.kafka.topic-replicas}") int replicas) {
return TopicBuilder.name("orders").partitions(6).replicas(replicas).build();
}
}
class OrderEvents {
private static final Logger log = LoggerFactory.getLogger(OrderEvents.class);
private final KafkaTemplate<String, OrderPlaced> kafka;
OrderEvents(KafkaTemplate<String, OrderPlaced> kafka) {
this.kafka = kafka;
}
void placed(long orderId, BigDecimal total) {
OrderPlaced event = new OrderPlaced(UUID.randomUUID(), orderId, total);
kafka.send("orders", String.valueOf(orderId), event)
.whenComplete((result, ex) -> {
if (ex != null) {
log.error("Could not publish order {}", orderId, ex);
}
});
}
}
class BillingConsumer {
(topics = "orders", groupId = "billing")
void onOrderPlaced(OrderPlaced event) {
// idempotent work keyed by event.eventId()
}
}
When to use it
- Publish every order state change keyed by order id so downstream services see each order's history in sequence.
- Add a fraud-scoring service as a new consumer group that reads the last seven days of payments from the start.
- Send records that fail validation to a dead-letter topic and replay them after a consumer fix is deployed.
More examples
Serialisation, consumer and topic settings
<code>acks=all</code> waits until every in-sync replica has the record, so an acknowledged send survives a broker failure; recent Kafka clients already default to it, and setting it explicitly protects you from an override elsewhere. <code>ErrorHandlingDeserializer</code> wraps the JSON deserialiser so a malformed record becomes an error the error handler can dead-letter, rather than an exception that stops the consumer on the same offset. The trusted packages list stops type headers from naming arbitrary classes. <code>app.kafka.topic-replicas</code> is 1 for the single local broker and 3 in the production profile; with three replicas and <code>acks=all</code>, an acknowledged record is on every in-sync broker.
spring.kafka.bootstrap-servers=${KAFKA_BOOTSTRAP_SERVERS:localhost:9092}
spring.kafka.producer.key-serializer=org.apache.kafka.common.serialization.StringSerializer
spring.kafka.producer.value-serializer=org.springframework.kafka.support.serializer.JacksonJsonSerializer
spring.kafka.producer.acks=all
spring.kafka.consumer.group-id=billing
spring.kafka.consumer.auto-offset-reset=earliest
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.trusted.packages=com.example.orders.events
app.kafka.topic-replicas=1
# application-prod.properties
# app.kafka.topic-replicas=3
Retries, then a dead-letter topic
Spring Boot passes a single <code>CommonErrorHandler</code> bean to the listener container factory. Each failing record is retried three times one second apart, then published to <code>orders.dlt</code> on the same partition number, so the dead-letter topic needs at least as many partitions as the source. Exceptions marked not retryable go to the dead-letter topic at once. A record that failed deserialisation is republished with its original bytes, so if the template's value serialiser is the JSON one, those bytes arrive in the dead-letter topic as a base64 string; give the recoverer a template that writes <code>byte[]</code> unchanged when you need them byte for byte.
import org.apache.kafka.common.TopicPartition;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.listener.DeadLetterPublishingRecoverer;
import org.springframework.kafka.listener.DefaultErrorHandler;
import org.springframework.util.backoff.FixedBackOff;
@Configuration
public class KafkaErrorHandlingConfig {
@Bean
DefaultErrorHandler kafkaErrorHandler(KafkaTemplate<Object, Object> template) {
DeadLetterPublishingRecoverer recoverer = new DeadLetterPublishingRecoverer(template,
(record, ex) -> new TopicPartition(record.topic() + ".dlt", record.partition()));
DefaultErrorHandler handler = new DefaultErrorHandler(recoverer, new FixedBackOff(1000L, 3L));
handler.addNotRetryableExceptions(IllegalArgumentException.class);
return handler;
}
}
Reading the key, partition and offset
Partition and offset identify a record exactly, which makes them the most useful values to log when tracing a duplicate or a gap. <code>concurrency = "3"</code> starts three consumers in this instance; with six partitions and two instances, each consumer gets one partition.
import java.util.UUID;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.support.KafkaHeaders;
import org.springframework.messaging.handler.annotation.Header;
import org.springframework.messaging.handler.annotation.Payload;
import org.springframework.stereotype.Component;
record OrderCancelled(UUID eventId, long orderId, String reason) {
}
@Component
public class ShippingConsumer {
private static final Logger log = LoggerFactory.getLogger(ShippingConsumer.class);
@KafkaListener(topics = "orders-cancelled", groupId = "shipping", concurrency = "3")
void onCancelled(@Payload OrderCancelled event,
@Header(KafkaHeaders.RECEIVED_KEY) String key,
@Header(KafkaHeaders.RECEIVED_PARTITION) int partition,
@Header(KafkaHeaders.OFFSET) long offset) {
log.info("Cancelling shipment for order {} (key {}, partition {}, offset {})",
event.orderId(), key, partition, offset);
}
}
Discussion