Sending and Receiving Kafka Messages in Spring Boot

1. Overview

Apache Kafka is a popular event streaming platform for services to communicate asynchronously. Spring Boot supports Kafka through the Spring Kafka library, so you don’t have to manage KafkaProducer and the poll() loop yourself.

In this tutorial, you’ll build a small example: an order service sends an OrderCreatedEvent, and a consumer receives it to reserve inventory. After reading, you’ll know how to configure, send and receive JSON messages, handle errors with Dead Letter Topic, and test with Embedded Kafka.

2. Setup

2.1. Maven Dependencies

You need spring-kafka. Spring Boot already manages the version, so you don’t need to specify it:

<dependency>
    <groupId>org.springframework.kafka</groupId>
    <artifactId>spring-kafka</artifactId>
</dependency>

<!-- For testing -->
<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-test</artifactId>
    <scope>test</scope>
</dependency>
<dependency>
    <groupId>org.springframework.kafka</groupId>
    <artifactId>spring-kafka-test</artifactId>
    <scope>test</scope>
</dependency>
<dependency>
    <groupId>org.awaitility</groupId>
    <artifactId>awaitility</artifactId>
    <scope>test</scope>
</dependency>

spring-kafka-test provides an in-memory Kafka broker for testing. Awaitility helps wait for asynchronous processing without using Thread.sleep().

2.2. Running Kafka Locally

The quickest way is to use Apache Kafka’s official image in KRaft mode, which doesn’t require ZooKeeper:

docker run -d --name kafka -p 9092:9092 apache/kafka:latest

The broker will listen at localhost:9092.

3. Data Model and Configuration

The event to be sent is a simple record:

package com.nonla.shop.order;

import java.math.BigDecimal;

public record OrderCreatedEvent(String orderId, String customerId, BigDecimal amount) {
}

Next is the configuration in application.yml:

spring:
  kafka:
    bootstrap-servers: localhost:9092
    producer:
      key-serializer: org.apache.kafka.common.serialization.StringSerializer
      value-serializer: org.springframework.kafka.support.serializer.JsonSerializer
      acks: all
    consumer:
      group-id: inventory-service
      auto-offset-reset: earliest
      key-deserializer: org.apache.kafka.common.serialization.StringDeserializer
      value-deserializer: org.springframework.kafka.support.serializer.JsonDeserializer
      properties:
        spring.json.trusted.packages: "com.nonla.shop.order"

Meaning of each part:

• bootstrap-servers is the broker address.
• Producer uses StringSerializer for key and JsonSerializer for value. JsonSerializer uses Jackson to convert objects to JSON.
• acks: all requires the broker to acknowledge only after all replicas have written. This is safer but a bit slower.
• group-id identifies the consumer group. Instances in the same group will share partitions.
• auto-offset-reset: earliest tells the consumer to read from the beginning of the topic when the group has no committed offset.
• spring.json.trusted.packages declares which packages JsonDeserializer is allowed to instantiate. Without this line, you’ll get an error “The class … is not in the trusted packages”.

3.1. Auto-create Topics

Spring Boot automatically configures a KafkaAdmin. This bean will create any NewTopic declared in the context:

package com.nonla.shop.config;

import org.apache.kafka.clients.admin.NewTopic;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.kafka.config.TopicBuilder;

@Configuration
public class KafkaTopicConfig {

    public static final String ORDERS_TOPIC = "orders";

    @Bean
    public NewTopic ordersTopic() {
        return TopicBuilder.name(ORDERS_TOPIC)
                .partitions(3)
                .replicas(1)
                .build();
    }
}

When the application starts, the orders topic with 3 partitions will be created if it doesn’t exist. If the topic already exists, KafkaAdmin will skip it and won’t override the current configuration.

4. Sending Messages with KafkaTemplate

Spring Boot automatically creates a KafkaTemplate bean based on producer configuration. You just need to inject it:

package com.nonla.shop.order;

import com.nonla.shop.config.KafkaTopicConfig;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.support.SendResult;
import org.springframework.stereotype.Service;

import java.util.concurrent.CompletableFuture;

@Service
public class OrderEventPublisher {

    private static final Logger log = LoggerFactory.getLogger(OrderEventPublisher.class);

    private final KafkaTemplate<String, OrderCreatedEvent> kafkaTemplate;

    public OrderEventPublisher(KafkaTemplate<String, OrderCreatedEvent> kafkaTemplate) {
        this.kafkaTemplate = kafkaTemplate;
    }

    public CompletableFuture<SendResult<String, OrderCreatedEvent>> publish(OrderCreatedEvent event) {
        return kafkaTemplate.send(KafkaTopicConfig.ORDERS_TOPIC, event.orderId(), event)
                .whenComplete((result, ex) -> {
                    if (ex != null) {
                        log.error("Failed to send order {}", event.orderId(), ex);
                        return;
                    }
                    var metadata = result.getRecordMetadata();
                    log.info("Sent order {} to partition {} offset {}",
                            event.orderId(), metadata.partition(), metadata.offset());
                });
    }
}

Key points:

• send(topic, key, value) uses orderId as the key. Kafka hashes the key to choose a partition, so all messages for the same order go to the same partition. This preserves order between them.
• Since Spring Kafka 3.0, send() returns CompletableFuture<SendResult> instead of ListenableFuture which was deprecated.
• send() is asynchronous. The call returns immediately, and the callback in whenComplete runs when the broker responds.

On successful send, the log will look like:

INFO  c.n.s.order.OrderEventPublisher : Sent order ORD-1001 to partition 2 offset 0

If you need to block until the send completes, for example in a batch job, you can call publish(event).get(5, TimeUnit.SECONDS). Don’t do this in high-load request handling threads, because each call will block a thread waiting for the broker.

Note: Writing to the database and then sending to Kafka in the same method is not an atomic transaction. If the database commits successfully but Kafka send fails, data will diverge. For critical data, consider the Transactional Outbox pattern.

5. Receiving Messages with @KafkaListener

On the consumer side, you just need a method with the @KafkaListener annotation:

package com.nonla.shop.inventory;

import com.nonla.shop.config.KafkaTopicConfig;
import com.nonla.shop.order.OrderCreatedEvent;
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;

@Component
public class OrderCreatedListener {

    private static final Logger log = LoggerFactory.getLogger(OrderCreatedListener.class);

    private final InventoryService inventoryService;

    public OrderCreatedListener(InventoryService inventoryService) {
        this.inventoryService = inventoryService;
    }

    @KafkaListener(topics = KafkaTopicConfig.ORDERS_TOPIC, concurrency = "3")
    public void onOrderCreated(@Payload OrderCreatedEvent event,
                               @Header(KafkaHeaders.RECEIVED_PARTITION) int partition,
                               @Header(KafkaHeaders.OFFSET) long offset) {
        log.info("Received order {} from partition {} offset {}", event.orderId(), partition, offset);
        inventoryService.reserve(event);
    }
}

InventoryService is just a regular service:

package com.nonla.shop.inventory;

import com.nonla.shop.order.OrderCreatedEvent;
import org.springframework.stereotype.Service;

@Service
public class InventoryService {

    public void reserve(OrderCreatedEvent event) {
        if (event.amount().signum() <= 0) {
            throw new IllegalArgumentException("Invalid amount: " + event.amount());
        }
        // Reserve the item in inventory...
    }
}

Explanation:

• topics specifies the topic to listen to. groupId is taken from spring.kafka.consumer.group-id, or you can override it directly in the annotation.
• concurrency = "3" creates 3 consumer threads, each handling one partition. A value higher than the number of partitions is pointless, as extra threads won’t receive anything.
• @Payload receives the deserialized object. @Header gets metadata like partition and offset.
• By default, the offset is committed after the method completes without throwing an exception (AckMode.BATCH). So if the application dies in between, the message will be processed again. This is at-least-once semantics, so your logic must be idempotent: processing the same orderId twice shouldn’t reserve inventory twice.

If you need the key, timestamp, or custom headers, you can receive the whole ConsumerRecord<String, OrderCreatedEvent> as a parameter.

6. Error Handling and Dead Letter Topic

When a listener throws an exception, the default DefaultErrorHandler will retry 9 times in a row, then log and skip the message. In practice, you usually want to move failed messages to a separate topic for inspection later:

package com.nonla.shop.config;

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 KafkaErrorConfig {

    @Bean
    public DefaultErrorHandler errorHandler(KafkaTemplate<Object, Object> kafkaTemplate) {
        var recoverer = new DeadLetterPublishingRecoverer(kafkaTemplate);
        var handler = new DefaultErrorHandler(recoverer, new FixedBackOff(1000L, 2));
        handler.addNotRetryableExceptions(IllegalArgumentException.class);
        return handler;
    }
}

Spring Boot automatically attaches this CommonErrorHandler bean to the default container factory. The configuration above works as follows:

• FixedBackOff(1000L, 2) retries at most 2 times, 1 second apart.
• When retries are exhausted, DeadLetterPublishingRecoverer sends the message to topic orders.DLT. This name is the original topic name plus the .DLT suffix, and the message keeps its partition. So the DLT topic needs at least as many partitions as the original topic.
• IllegalArgumentException is a data error, so retrying won’t help. The handler will send the message directly to the DLT.

A message with amount = -5 will produce a log like:

ERROR o.s.k.l.DefaultErrorHandler : Backoff none exhausted for orders-1@4
INFO  c.n.s.inventory.DltListener  : Error message in DLT: ORD-1002

Note: Errors that occur during deserialization, for example corrupt JSON, won’t reach the listener. If not handled, the consumer can get stuck on that message, called a “poison pill”. Wrap your deserializer with ErrorHandlingDeserializer:

spring:
  kafka:
    consumer:
      value-deserializer: org.springframework.kafka.support.serializer.ErrorHandlingDeserializer
      properties:
        spring.deserializer.value.delegate.class: org.springframework.kafka.support.serializer.JsonDeserializer

With this configuration, deserialization errors are passed to DefaultErrorHandler and the failed message also goes to the DLT.

7. Testing with Embedded Kafka

@EmbeddedKafka starts a broker in memory, so you don’t need Docker when running tests. The example below tests the full round trip: send via OrderEventPublisher, then verify that InventoryService receives the event correctly.

package com.nonla.shop.order;

import com.nonla.shop.inventory.InventoryService;
import org.junit.jupiter.api.Test;
import org.mockito.ArgumentCaptor;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.test.context.SpringBootTest;
import org.springframework.kafka.test.context.EmbeddedKafka;
import org.springframework.test.context.bean.override.mockito.MockitoBean;

import java.math.BigDecimal;
import java.time.Duration;

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

@SpringBootTest(properties = "spring.kafka.bootstrap-servers=${spring.embedded.kafka.brokers}")
@EmbeddedKafka(partitions = 3, topics = "orders")
class OrderKafkaIntegrationTest {

    @Autowired
    private OrderEventPublisher publisher;

    @MockitoBean
    private InventoryService inventoryService;

    @Test
    void whenOrderPublished_thenListenerReservesInventory() throws Exception {
        var event = new OrderCreatedEvent("ORD-1001", "CUS-42", new BigDecimal("250000"));

        var result = publisher.publish(event).get();
        assertThat(result.getRecordMetadata().topic()).isEqualTo("orders");

        var captor = ArgumentCaptor.forClass(OrderCreatedEvent.class);
        await().atMost(Duration.ofSeconds(10))
                .untilAsserted(() -> verify(inventoryService).reserve(captor.capture()));

        assertThat(captor.getValue()).isEqualTo(event);
    }
}

What each part does:

• ${spring.embedded.kafka.brokers} is a property provided by @EmbeddedKafka. It contains the embedded broker address, which we use to override bootstrap-servers.
• @MockitoBean (available since Spring Boot 3.4) replaces the real InventoryService with a mock for easier verification. With earlier versions, use @MockBean.
• publish(event).get() waits for broker acknowledgment, so the test is sure the message was written.
• The listener runs on a different thread, so await().untilAsserted(...) will retry the verify call for up to 10 seconds.
• Since OrderCreatedEvent is a record, equals() compares by value. assertThat(...).isEqualTo(event) verifies that JSON was serialized and deserialized correctly.

The auto-offset-reset: earliest configuration in the consumer settings helps make tests stable. If the consumer joins the group after the message was already sent, it will still read it.

8. Common Errors and When Not to Use Kafka

Some errors you might encounter:

• Missing trusted.packages: JsonDeserializer refuses to create unknown classes. Declare the correct package. Use only "*" if you fully control all producers.
• Producer and consumer in different packages: JsonSerializer by default writes the full class name in the __TypeId__ header. If the consumer service has a class with the same name but in a different package, deserialization will fail. In that case, set spring.json.use.type.headers: false and spring.json.value.default.type: com.nonla.inventory.OrderCreatedEvent on the consumer side.
• Listener takes too long: if a single poll() takes longer than max.poll.interval.ms (default 5 minutes), the broker considers the consumer dead and rebalances. This causes messages to be processed again. Reduce max.poll.records or optimize your logic.
• concurrency larger than partition count: extra threads will sit idle. To increase throughput, you must increase the partition count first.
• Sending null key: messages are distributed evenly across partitions, so you lose ordering guarantees per entity.

Kafka also isn’t the right choice for every scenario:

• You need synchronous request/response and want immediate results. Use REST or gRPC instead.
• Your system is small, traffic is low, and you don’t need to retain event history. Running a Kafka cluster is then excessive overhead, and RabbitMQ or a queue table in the database may be enough.
• You need complex message routing per message, or need special latency or priority per message. This is where traditional message brokers excel.

9. Conclusion

In this tutorial, you configured Spring Kafka with JSON serializers and created a topic with NewTopic. You sent messages via KafkaTemplate with a key to preserve order, and received messages with @KafkaListener with concurrency. You also handled errors with DefaultErrorHandler combined with Dead Letter Topic, then wrote an integration test with @EmbeddedKafka and Awaitility.

When deploying to production, remember three things: listeners must be idempotent, handle deserialization errors with ErrorHandlingDeserializer, and event publishing must be consistent with database data.