Spring Kafka: Producing Messages with KafkaTemplate

KafkaTemplate is the central Spring Kafka class for sending messages. It wraps the Kafka producer and integrates with Spring's exception translation and transaction support.

Share

Contents

Dependency & Configuration

Add spring-kafka to your project. Spring Boot auto-configures a KafkaTemplate bean using producer properties from application.properties.

XML
<dependency>
    <groupId>org.springframework.kafka</groupId>
    <artifactId>spring-kafka</artifactId>
</dependency>
Properties
spring.kafka.bootstrap-servers=localhost:9092
spring.kafka.producer.key-serializer=org.apache.kafka.common.serialization.StringSerializer
spring.kafka.producer.value-serializer=org.apache.kafka.common.serialization.StringSerializer
# Optional: set client id for easier monitoring
spring.kafka.producer.client-id=my-producer
Spring Boot creates a KafkaTemplate<String, String> bean automatically when spring-kafka is on the classpath and producer properties are set. Inject it directly into any Spring bean.

Sending a String Message

Inject KafkaTemplate and call send(). You can send to a topic with just a value, with a key, or to a specific partition.

Java
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.stereotype.Service;

@Service
public class OrderProducer {

    private final KafkaTemplate<String, String> kafkaTemplate;

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

    // Send value only — Kafka picks the partition via round-robin or sticky
    public void sendOrder(String message) {
        kafkaTemplate.send("orders", message);
    }

    // Send with a key — same key always goes to the same partition
    public void sendOrderWithKey(String orderId, String message) {
        kafkaTemplate.send("orders", orderId, message);
    }

    // Send to a specific partition
    public void sendToPartition(String message) {
        kafkaTemplate.send("orders", 0, null, message); // partition 0
    }
}

Sending an Object

To send POJOs, switch the value serializer to JsonSerializer. You can also use ProducerRecord directly to attach custom headers.

Properties
spring.kafka.producer.value-serializer=org.springframework.kafka.support.serializer.JsonSerializer
Java
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.common.header.internals.RecordHeader;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.stereotype.Service;

@Service
public class OrderProducer {

    private final KafkaTemplate<String, Order> kafkaTemplate;

    public OrderProducer(KafkaTemplate<String, Order> kafkaTemplate) {
        this.kafkaTemplate = kafkaTemplate;
    }

    // Send a POJO — serialized to JSON automatically
    public void sendOrder(Order order) {
        kafkaTemplate.send("orders", order.getId(), order);
    }

    // Send with custom headers using ProducerRecord
    public void sendWithHeaders(Order order) {
        ProducerRecord<String, Order> record = new ProducerRecord<>("orders", order.getId(), order);
        record.headers().add(new RecordHeader("source", "order-service".getBytes()));
        record.headers().add(new RecordHeader("version", "1".getBytes()));
        kafkaTemplate.send(record);
    }
}
When using JsonSerializer, Spring Kafka adds a type header to the Kafka record by default. Configure the consumer's JsonDeserializer with spring.kafka.consumer.properties.spring.json.trusted.packages=* to deserialize correctly.

Send Callbacks & ListenableFuture

kafkaTemplate.send() returns a CompletableFuture (Spring Kafka 3.x) that resolves to a SendResult. Use whenComplete to log success or failure without blocking.

Java
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 OrderProducer {

    private final KafkaTemplate<String, String> kafkaTemplate;

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

    public void sendWithCallback(String message) {
        CompletableFuture<SendResult<String, String>> future =
            kafkaTemplate.send("orders", message);

        future.whenComplete((result, ex) -> {
            if (ex == null) {
                System.out.printf("Sent message [%s] to partition %d offset %d%n",
                    message,
                    result.getRecordMetadata().partition(),
                    result.getRecordMetadata().offset());
            } else {
                System.err.println("Failed to send message: " + ex.getMessage());
            }
        });
    }
}
If you need guaranteed delivery confirmation before returning to the caller, use future.get() to block, but this significantly reduces throughput. Prefer async callbacks for high-volume producers.

Transactional Sending

Enable Kafka transactions by setting a transaction-id-prefix. Annotate service methods with @Transactional and Spring Kafka will wrap all sends in a single Kafka transaction.

Properties
spring.kafka.producer.transaction-id-prefix=tx-orders-
Java
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;

@Service
public class OrderService {

    private final KafkaTemplate<String, String> kafkaTemplate;

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

    // All sends inside this method are committed atomically
    @Transactional
    public void processOrder(String orderId, String orderData) {
        kafkaTemplate.send("orders", orderId, orderData);
        kafkaTemplate.send("audit-log", orderId, "ORDER_CREATED");
        // If an exception is thrown here, both sends are rolled back
    }

    // Programmatic transaction (no @Transactional required)
    public void sendProgrammatic(String message) {
        kafkaTemplate.executeInTransaction(ops -> {
            ops.send("orders", message);
            ops.send("audit-log", message);
            return true;
        });
    }
}
Kafka transactions guarantee that all sends either commit or abort together. They do not span JPA / database transactions: for atomic DB + Kafka writes, use the Outbox pattern.

Custom ProducerFactory

Override the auto-configured producer by declaring your own ProducerFactory and KafkaTemplate beans with custom properties such as acks, delivery.timeout.ms, batch size, and linger.

Java
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.common.serialization.StringSerializer;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.kafka.core.DefaultKafkaProducerFactory;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.core.ProducerFactory;

import java.util.HashMap;
import java.util.Map;

@Configuration
public class KafkaProducerConfig {

    @Bean
    public ProducerFactory<String, String> producerFactory() {
        Map<String, Object> props = new HashMap<>();
        props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
        props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
        props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class);

        // Reliability: wait for all in-sync replicas to acknowledge
        props.put(ProducerConfig.ACKS_CONFIG, "all");

        // Retry transient failures for up to 2 minutes (retries stays at its default, unlimited)
        props.put(ProducerConfig.DELIVERY_TIMEOUT_MS_CONFIG, 120_000);

        // Throughput tuning: batch up to 16 KB before sending
        props.put(ProducerConfig.BATCH_SIZE_CONFIG, 16384);

        // Wait up to 5ms to fill the batch
        props.put(ProducerConfig.LINGER_MS_CONFIG, 5);

        return new DefaultKafkaProducerFactory<>(props);
    }

    @Bean
    public KafkaTemplate<String, String> kafkaTemplate(ProducerFactory<String, String> pf) {
        return new KafkaTemplate<>(pf);
    }
}
Retries can write a message twice if the broker stores it but the acknowledgement never reaches the producer. Idempotence (enable.idempotence=true, the default since Kafka 3.0) stops these duplicates, so leave it on.

Related articles

All Spring articles →
Spring Boot: WebSocket with STOMPSpring Boot WebSocket with STOMP, configuring WebSocket endpoints, message controllers, SimpMessagingTemplate, and JavaScript client with SockJS.Spring Kafka: Consuming Messages with @KafkaListenerSpring Kafka @KafkaListener: consuming messages with Spring Kafka, consumer groups, concurrency, manual acknowledgement, and error handling with Java examples.Spring Boot and AWS SQS with @SqsListenerSpring Boot AWS SQS integration: @SqsListener, message batching, visibility timeout, DLQ configuration, FIFO queues, and reliable message processing patterns.Spring Integration: Message Channels and GatewaysSpring Integration: message channels, gateways, service activators, transformers, routers, and enterprise integration patterns (EIP) with Spring Boot.