Using Kafka in Spring Boot — Producer & Consumer Implementation Guide

This post is a follow-up to Kafka Infrastructure Setup & Topic Design.
Please refer to that article first for Docker Compose configuration and topic design principles.

In this post, we’ll go through how to implement Kafka Producer and Consumer in a Spring Boot service.
We’ll publish events using KafkaTemplate, consume them with @KafkaListener, and isolate failed messages using a Dead Letter Topic (DLT).

1. Add Dependency (pom.xml)

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

2. application.yml Configuration

spring:
  kafka:
    bootstrap-servers: ${SPRING_KAFKA_BOOTSTRAP_SERVERS:<KAFKA_HOST_IP>:9092}
    properties:
      security.protocol: ${SPRING_KAFKA_SECURITY_PROTOCOL:PLAINTEXT}

    # Producer configuration (service that publishes events)
    producer:
      key-serializer: org.apache.kafka.common.serialization.StringSerializer
      value-serializer: org.apache.kafka.common.serialization.StringSerializer

    # Consumer configuration (service that consumes events)
    consumer:
      group-id: my-service-cg          # unique group-id per service
      auto-offset-reset: earliest
      enable-auto-commit: false       # Manual ACK is required
      key-deserializer: org.apache.kafka.common.serialization.StringDeserializer
      value-deserializer: org.apache.kafka.common.serialization.StringDeserializer

    listener:
      ack-mode: manual_immediate      # Manual ACK mode

Environment Variables:

• SPRING_KAFKA_BOOTSTRAP_SERVERS: default <KAFKA_HOST_IP>:9092 (local development)
• SPRING_KAFKA_SECURITY_PROTOCOL: default PLAINTEXT (local), use SASL_SSL in production

3. Define MessagingConstants

Always manage topic names and group IDs as constants.
Avoid using raw strings directly in your code.

// src/main/java/com/example/myservice/constants/MessagingConstants.java
public final class MessagingConstants {

    private MessagingConstants() {}

    // Kafka Topics
    public static final String TOPIC_MY_EVENT_CREATED = "myapp.my.event.created";
    public static final String TOPIC_MY_EVENT_DELETED = "myapp.my.event.deleted";
    public static final String TOPIC_MY_EVENT_DLT     = "myapp.my.event.dlt";

    // Consumer Group IDs
    public static final String GROUP_MY_SERVICE     = "my-service-cg";
    public static final String GROUP_MY_SERVICE_DLT = "my-service-dlt-cg";
}

4. Kafka Config Bean (for Consumer services)

If your service consumes messages, you need to register a kafkaListenerContainerFactory bean.
This is also where DLT (Dead Letter Topic) routing is configured.

// src/main/java/com/example/myservice/config/KafkaConfig.java
@Configuration
@EnableKafka
@RequiredArgsConstructor
public class KafkaConfig {

    private final KafkaTemplate<String, String> kafkaTemplate;

    @Bean
    public ConcurrentKafkaListenerContainerFactory<String, String> kafkaListenerContainerFactory(
            ConsumerFactory<String, String> consumerFactory) {

        ConcurrentKafkaListenerContainerFactory<String, String> factory =
                new ConcurrentKafkaListenerContainerFactory<>();
        factory.setConsumerFactory(consumerFactory);
        factory.getContainerProperties().setAckMode(ContainerProperties.AckMode.MANUAL_IMMEDIATE);

        // Retry 3 times at 1-second intervals, then send to DLT
        DefaultErrorHandler errorHandler = new DefaultErrorHandler(
                new DeadLetterPublishingRecoverer(kafkaTemplate),
                new FixedBackOff(1000L, 3L)
        );
        factory.setCommonErrorHandler(errorHandler);
        return factory;
    }
}

5. Producer Implementation (Publishing Events)

// src/main/java/com/example/myservice/service/kafka/KafkaMyEventPublisher.java
@Service
@Slf4j
@RequiredArgsConstructor
public class KafkaMyEventPublisher {

    private final KafkaTemplate<String, String> kafkaTemplate;
    private final ObjectMapper objectMapper;

    private static final Map<String, String> TOPIC_MAP = Map.of(
        "MY_EVENT_CREATED", MessagingConstants.TOPIC_MY_EVENT_CREATED,
        "MY_EVENT_DELETED", MessagingConstants.TOPIC_MY_EVENT_DELETED
    );

    public boolean publishEvent(String eventType, String userId, Object payload) {
        String topic = TOPIC_MAP.get(eventType);
        if (topic == null) {
            log.error("Unknown event type: {}", eventType);
            return false;
        }

        try {
            String jsonPayload = objectMapper.writeValueAsString(payload);
            kafkaTemplate.send(topic, userId, jsonPayload).get();  // userId = partition key
            log.info("Kafka published: topic={}, userId={}", topic, userId);
            return true;
        } catch (Exception e) {
            log.error("Kafka publish failed: topic={}, userId={}, error={}", topic, userId, e.getMessage(), e);
            return false;
        }
    }
}

Message Payload Structure (JSON):

{
  "eventType": "MY_EVENT_CREATED",
  "userId": "user-id-123",
  "idempotencyKey": "uuid-v4",
  "data": { ... }
}

Including an idempotencyKey (UUID) helps prevent duplicate processing on the consumer side.

6. Consumer Implementation (Consuming Events)

// src/main/java/com/example/myservice/service/kafka/MyEventKafkaConsumer.java
@Service
@Slf4j
@RequiredArgsConstructor
public class MyEventKafkaConsumer {

    private final MyBusinessService myBusinessService;
    private final ObjectMapper objectMapper;

    @KafkaListener(
        topics = MessagingConstants.TOPIC_MY_EVENT_CREATED,
        groupId = MessagingConstants.GROUP_MY_SERVICE,
        containerFactory = "kafkaListenerContainerFactory"
    )
    public void handleMyEventCreated(
        @Payload String payload,
        @Header(KafkaHeaders.RECEIVED_KEY) String userId,
        Acknowledgment ack
    ) {
        try {
            JsonNode messageData = objectMapper.readTree(payload);
            log.info("Processing event: userId={}", userId);

            // Always delegate business logic to a service layer
            myBusinessService.processCreatedEvent(userId, messageData);

            ack.acknowledge();  // ACK only on success
            log.info("Event processed: userId={}", userId);

        } catch (RuntimeException e) {
            log.error("Event processing failed: userId={}", userId, e);
            throw e;  // retry → DLT routing
        } catch (Exception e) {
            log.error("Event processing failed: userId={}", userId, e);
            throw new RuntimeException(e);
        }
    }

    // Dead Letter Topic handler — receives final failed messages
    @KafkaListener(
        topics = MessagingConstants.TOPIC_MY_EVENT_DLT,
        groupId = MessagingConstants.GROUP_MY_SERVICE_DLT
    )
    public void handleDlt(
        @Payload String payload,
        @Header(value = KafkaHeaders.EXCEPTION_MESSAGE, required = false) String errorMessage
    ) {
        log.error("DLT received: error={}, payload={}", errorMessage, payload);
        // Add alerting, monitoring, or manual retry logic here
    }
}

7. Key Consumer Rules

• Call ack.acknowledge() only on success
• Throw an exception on failure → Spring Kafka retries → then sends to DLT
• If you ACK on failure, the message is lost (no retry, no DLT)

This is why enable-auto-commit: false is set in the config. Auto-commit doesn’t tie the committed offset to successful business processing — it commits on the poll cycle, so an offset can get ahead of the work when processing outlives a poll or is handed to another thread, and a crash then loses the record. Disabling it hands offset control to the listener container’s AckMode instead: RECORD commits after each listener call returns, BATCH after the poll’s batch, and MANUAL_IMMEDIATE (used here) only when you call ack.acknowledge(). So you acknowledge after the work succeeds, and on failure you throw and let the configured error handler retry or route to the DLT.

8. Checklist for Adding Kafka to a New Service

• [ ] Add spring-kafka dependency to pom.xml
• [ ] Configure spring.kafka in application.yml
• [ ] Create MessagingConstants.java (no hardcoding)
• [ ] Producer: use KafkaTemplate<String, String>
• [ ] Consumer: register kafkaListenerContainerFactory + enable Manual ACK
• [ ] Consumer: use constants in @KafkaListener
• [ ] Consumer: ACK only on success, throw exception on failure
• [ ] Implement DLT handler (*_DLT topic)

Related Posts

• Kafka Core Concepts — Broker, Topic, Partition, Consumer Group, ACK, DLT
• Kafka Infrastructure Setup & Topic Design — Docker Compose, topic naming, common rules
• Using Kafka in FastAPI — aiokafka + pydantic-settings + async Producer