---

name: spring-kafka description: | Spring for Apache Kafka integration with KafkaTemplate and @KafkaListener. Covers producers, consumers, retry topics, DLT, and transactions.

USE WHEN: user mentions "spring kafka", "KafkaTemplate", "@KafkaListener", "kafka producer Spring", "kafka consumer Spring", "retry topic"

DO NOT USE FOR: raw Kafka - use kafka skill, RabbitMQ - use spring-amqp instead allowed-tools: Read, Grep, Glob, Write, Edit

Spring Kafka - Quick Reference

Deep Knowledge: Use mcp__documentation__fetch_docs with technology: kafka for comprehensive documentation.

Dependencies

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

Configuration

application.yml

spring:
  kafka:
    bootstrap-servers: localhost:9092
    consumer:
      group-id: my-group
      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.example.dto"
    producer:
      key-serializer: org.apache.kafka.common.serialization.StringSerializer
      value-serializer: org.springframework.kafka.support.serializer.JsonSerializer
      acks: all
      properties:
        enable.idempotence: true
    listener:
      ack-mode: manual
      concurrency: 3

Producer Pattern

KafkaTemplate

@Service
@RequiredArgsConstructor
public class OrderProducer {

    private final KafkaTemplate<String, OrderEvent> kafkaTemplate;

    public void sendOrder(OrderEvent event) {
        kafkaTemplate.send("orders", event.getOrderId(), event)
            .whenComplete((result, ex) -> {
                if (ex != null) {
                    log.error("Failed to send order: {}", event.getOrderId(), ex);
                } else {
                    log.info("Order sent: {} to partition {}",
                        event.getOrderId(),
                        result.getRecordMetadata().partition());
                }
            });
    }

    // With headers
    public void sendWithHeaders(OrderEvent event, String correlationId) {
        ProducerRecord<String, OrderEvent> record = new ProducerRecord<>(
            "orders", event.getOrderId(), event);
        record.headers()
            .add("correlation-id", correlationId.getBytes())
            .add("source", "order-service".getBytes());

        kafkaTemplate.send(record);
    }
}

Transactional Producer

@Configuration
public class KafkaConfig {

    @Bean
    public ProducerFactory<String, Object> producerFactory() {
        Map<String, Object> config = new HashMap<>();
        config.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
        config.put(ProducerConfig.TRANSACTIONAL_ID_CONFIG, "tx-");
        config.put(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG, true);
        return new DefaultKafkaProducerFactory<>(config);
    }

    @Bean
    public KafkaTemplate<String, Object> kafkaTemplate() {
        return new KafkaTemplate<>(producerFactory());
    }

    @Bean
    public KafkaTransactionManager<String, Object> kafkaTransactionManager() {
        return new KafkaTransactionManager<>(producerFactory());
    }
}

@Service
@Transactional("kafkaTransactionManager")
public class TransactionalProducer {

    public void sendMultiple(List<OrderEvent> events) {
        events.forEach(e -> kafkaTemplate.send("orders", e.getId(), e));
    }
}

Consumer Patterns

Basic @KafkaListener

@Service
@RequiredArgsConstructor
public class OrderConsumer {

    @KafkaListener(topics = "orders", groupId = "order-processor")
    public void consume(
            @Payload OrderEvent event,
            @Header(KafkaHeaders.RECEIVED_KEY) String key,
            @Header(KafkaHeaders.RECEIVED_PARTITION) int partition,
            @Header(KafkaHeaders.OFFSET) long offset,
            Acknowledgment ack) {

        log.info("Received order: {} from partition {} offset {}",
            event.getOrderId(), partition, offset);

        try {
            processOrder(event);
            ack.acknowledge();
        } catch (Exception e) {
            log.error("Failed to process order: {}", event.getOrderId(), e);
            throw e; // Will trigger retry
        }
    }
}

Batch Consumer

@KafkaListener(
    topics = "orders",
    groupId = "batch-processor",
    containerFactory = "batchKafkaListenerContainerFactory"
)
public void consumeBatch(
        List<OrderEvent> events,
        @Header(KafkaHeaders.RECEIVED_PARTITION) List<Integer> partitions,
        Acknowledgment ack) {

    log.info("Received batch of {} orders", events.size());
    processBatch(events);
    ack.acknowledge();
}

@Bean
public ConcurrentKafkaListenerContainerFactory<String, OrderEvent> batchKafkaListenerContainerFactory() {
    ConcurrentKafkaListenerContainerFactory<String, OrderEvent> factory =
        new ConcurrentKafkaListenerContainerFactory<>();
    factory.setConsumerFactory(consumerFactory());
    factory.setBatchListener(true);
    factory.getContainerProperties().setAckMode(AckMode.MANUAL);
    return factory;
}

Class-Level Listener

@KafkaListener(topics = "orders", groupId = "order-handler")
@Service
public class OrderHandler {

    @KafkaHandler
    public void handleCreated(OrderCreatedEvent event) {
        // Handle order created
    }

    @KafkaHandler
    public void handleUpdated(OrderUpdatedEvent event) {
        // Handle order updated
    }

    @KafkaHandler(isDefault = true)
    public void handleDefault(Object event) {
        log.warn("Unknown event type: {}", event.getClass());
    }
}

Retry Topics (Spring Kafka 3.x)

@RetryableTopic

@RetryableTopic(
    attempts = "3",
    backoff = @Backoff(delay = 1000, multiplier = 2.0, maxDelay = 10000),
    dltStrategy = DltStrategy.FAIL_ON_ERROR,
    autoCreateTopics = "true",
    topicSuffixingStrategy = TopicSuffixingStrategy.SUFFIX_WITH_INDEX_VALUE
)
@KafkaListener(topics = "orders", groupId = "retry-consumer")
public void consumeWithRetry(OrderEvent event, Acknowledgment ack) {
    processOrder(event);
    ack.acknowledge();
}

@DltHandler
public void handleDlt(OrderEvent event,
        @Header(KafkaHeaders.RECEIVED_TOPIC) String topic,
        @Header(KafkaHeaders.EXCEPTION_MESSAGE) String errorMessage) {

    log.error("DLT received: {} from {} - error: {}",
        event.getOrderId(), topic, errorMessage);
    // Store in database for manual review
    failedOrderRepository.save(new FailedOrder(event, errorMessage));
}

Manual Retry Configuration

@Configuration
@EnableKafka
public class KafkaRetryConfig {

    @Bean
    public RetryTopicConfiguration retryTopicConfiguration(KafkaTemplate<String, Object> template) {
        return RetryTopicConfigurationBuilder
            .newInstance()
            .maxAttempts(4)
            .fixedBackOff(3000)
            .includeTopic("orders")
            .doNotAutoCreateRetryTopics()
            .create(template);
    }
}

Error Handling

Custom Error Handler

@Bean
public DefaultErrorHandler errorHandler(KafkaTemplate<String, Object> template) {
    // Send to DLT after 3 retries
    DeadLetterPublishingRecoverer recoverer = new DeadLetterPublishingRecoverer(template,
        (record, ex) -> new TopicPartition(record.topic() + ".DLT", record.partition()));

    DefaultErrorHandler handler = new DefaultErrorHandler(recoverer,
        new FixedBackOff(1000L, 3L));

    // Don't retry for these exceptions
    handler.addNotRetryableExceptions(
        ValidationException.class,
        DeserializationException.class
    );

    return handler;
}

@Bean
public ConcurrentKafkaListenerContainerFactory<String, Object> kafkaListenerContainerFactory() {
    ConcurrentKafkaListenerContainerFactory<String, Object> factory =
        new ConcurrentKafkaListenerContainerFactory<>();
    factory.setConsumerFactory(consumerFactory());
    factory.setCommonErrorHandler(errorHandler(kafkaTemplate()));
    return factory;
}

Testing

@EmbeddedKafka

@SpringBootTest
@EmbeddedKafka(partitions = 1, topics = {"orders"})
class OrderProducerTest {

    @Autowired
    private EmbeddedKafkaBroker embeddedKafka;

    @Autowired
    private OrderProducer orderProducer;

    @Test
    void shouldSendOrder() throws Exception {
        OrderEvent event = new OrderEvent("123", "CREATED");

        Map<String, Object> consumerProps = KafkaTestUtils.consumerProps(
            "test-group", "true", embeddedKafka);
        ConsumerFactory<String, OrderEvent> cf = new DefaultKafkaConsumerFactory<>(consumerProps);
        Consumer<String, OrderEvent> consumer = cf.createConsumer();
        embeddedKafka.consumeFromAnEmbeddedTopic(consumer, "orders");

        orderProducer.sendOrder(event);

        ConsumerRecord<String, OrderEvent> record = KafkaTestUtils.getSingleRecord(consumer, "orders");
        assertThat(record.value().getOrderId()).isEqualTo("123");
    }
}

Consumer Test with CountDownLatch

@SpringBootTest
@EmbeddedKafka(partitions = 1, topics = {"orders"})
class OrderConsumerTest {

    @SpyBean
    private OrderConsumer orderConsumer;

    @Autowired
    private EmbeddedKafkaBroker embeddedKafka;

    @Test
    void shouldConsumeAndProcessOrder() throws Exception {
        CountDownLatch latch = new CountDownLatch(1);
        doAnswer(inv -> { inv.callRealMethod(); latch.countDown(); return null; })
            .when(orderConsumer).consume(any(), any());

        Map<String, Object> props = KafkaTestUtils.producerProps(embeddedKafka);
        KafkaTemplate<String, String> template = new KafkaTemplate<>(
            new DefaultKafkaProducerFactory<>(props));
        template.send("orders", "{\"orderId\":\"456\",\"status\":\"CREATED\"}");

        assertThat(latch.await(10, TimeUnit.SECONDS)).isTrue();
        verify(orderConsumer).consume(any(), any());
    }
}

Testcontainers

@SpringBootTest
@Testcontainers
class OrderIntegrationTest {

    @Container
    @ServiceConnection
    static KafkaContainer kafka = new KafkaContainer(
        DockerImageName.parse("apache/kafka-native:3.8.0"));

    @Autowired
    private KafkaTemplate<String, OrderEvent> kafkaTemplate;

    @Autowired
    private OrderRepository orderRepository;

    @Test
    void shouldProcessOrderEndToEnd() throws Exception {
        kafkaTemplate.send("orders", "key-1",
            new OrderEvent("789", "CREATED")).get(10, TimeUnit.SECONDS);

        await().atMost(Duration.ofSeconds(10))
            .untilAsserted(() -> {
                Optional<Order> order = orderRepository.findById("789");
                assertThat(order).isPresent();
                assertThat(order.get().getStatus()).isEqualTo("CREATED");
            });
    }
}

Deep dive: For MockConsumer/MockProducer, @EmbeddedKafka advanced patterns, and Node.js/Python Kafka testing, see the messaging-testing-kafka skill.

Best Practices

Do	Don't
Use acks=all for durability	Use acks=0 in production
Enable idempotence	Ignore duplicate messages
Configure DLT for failures	Silently drop failed messages
Use manual acknowledgment	Auto-commit without processing
Set proper deserializer trust	Trust all packages

Production Checklist

• acks=all configured
• Idempotence enabled
• Consumer group ID set
• Manual acknowledgment mode
• Retry topics configured
• DLT handler implemented
• Error handler configured
• Proper serializers set
• Trusted packages configured
• Monitoring metrics exposed

When NOT to Use This Skill

• Raw Kafka - Use kafka skill for broker config
• RabbitMQ - Use spring-amqp instead
• Simple messaging - Consider Spring Events
• Kafka Streams - May need additional skill

Anti-Patterns

Anti-Pattern	Problem	Solution
Auto commit	Message loss	Use manual ack
No error handler	Silent failures	Configure error handler
No DLT	Lost failed messages	Add dead letter topic
Blocking in listener	Consumer lag	Use async processing
Wrong deserializer	Errors on consume	Match producer serializer
No idempotency	Duplicate processing	Implement idempotent consumer

Quick Troubleshooting

Problem	Diagnostic	Fix
Consumer not receiving	Check group.id	Verify consumer group
Serialization error	Check value type	Configure correct deserializer
Rebalancing often	Check session.timeout	Increase timeout
Consumer lag	Check processing time	Optimize or scale consumers
Messages in DLT	Check error logs	Fix processing error

Reference Documentation

• Spring Kafka Reference
• Spring Boot Kafka Docs