Updated on 2026-06-29 GMT+08:00

Example Code for Spring Boot to Access Kafka Using Spring Kafka Module

This section applies to MRS 3.6.0 and later versions.

Function

Spring Kafka is a module integrated with Apache Kafka within the Spring ecosystem. With simplified APIs and configuration options, it allows you to implement Kafka producer and consumer in Spring applications with ease. Spring Kafka provides flexible and efficient solutions for both simple message queuing scenarios and complex event-driven architectures.

The following example demonstrates how Spring Boot uses the Spring Kafka module to access MRS Kafka.

Example Code

The following example code enables Spring Boot to access MRS Kafka using the Spring Kafka module.

Producer is the message production interface, Consumer is the message consumption interface, and KafkaProperties contains client parameters. You can modify the parameters based on the actual business needs.

Producer:

@RestController
public class MessageController {
    private final static Logger LOG = LoggerFactory.getLogger(MessageController.class);
    @Autowired
    private ProducerThread producerThread;
    @Value("${topic:example-metric1}")
    protected String topic;
    @GetMapping("/produce")
    public String produce() {
        String message = "Start to produce message";
        producerThread.start();
        LOG.info(message);
        return message;
    }
}

Consumer:
```
@Service
public class ConsumerService {
  private static final Logger LOG = LoggerFactory.getLogger(ConsumerService.class);
  @Value("${topic:example-metric1}")
  private String topic;
  @KafkaListener(
    containerFactory = "KafkaListenerContainerFactory",
    id = "id",
    idIsGroup = false,
    groupId = "groupid",
    topics = "${topic}"
  )
  public void listen(ConsumerRecord<?, ?> record) {
    LOG.info("The consumer poll 1 record from kafka, the topic is {}, the partition is {}, the offset is {}, the " +
      "key is {}, the value is {}", record.topic(), record.partition(), record.offset(), record.key(), record.value());
  }
}
```
Introduction to Consumer Parameters
- @KafkaListener: KafkaListener is an annotation provided by Spring Kafka, which allows you easily consume Kafka messages in Spring applications. You can use this annotation to define a method to listen to Kafka messages. The method will be automatically executed when a message is sent to a specified topic. In general, KafkaListener simplifies the integration of Kafka message consumption so that you only need to focus on service logic rather than underlying implementation.
- containerFactory: This parameter specifies the name of the factory bean used to create the MessageListenerContainer.
- MessageListenerContainer: MessageListenerContainer is a core interface in Spring-Kafka, used to manage and control the lifecycle of Kafka message consumers. It creates, starts, stops, and manages Kafka consumer instances, and interacts with the Kafka server to consume messages. In addition, MessageListenerContainer can handle messages in both single-thread and concurrent message consumption scenarios.
- id: This parameter specifies the unique identifier for each listener instance. If the groupId is not specified, the id will be used as the groupId. In an application with multiple listeners, different IDs can be used to distinguish different listener containers.
- idIsGroup: This parameter determines the relationship between the id and the groupId, precisely, determines whether the id is used as the groupId. The default value is false.
- groupId: This parameter specifies the ID of a Kafka consumer group. Each consumer belongs to a group. A group can contain multiple consumers who process messages concurrently.
- topics: This parameter specifies the names of the topics for production and consumption.

KafkaProperties:

// KafkaProperties 
@Configuration
public class KafkaProperties {
    // Common Client Config
    @Value("${bootstrap.servers:}")
    private String bootstrapServers;
    @Value("${security.protocol:SASL_PLAINTEXT}")
    private String securityProtocol;
    @Value("${sasl.mechanism:PLAIN}")
    private String saslMechanism;
    @Value("${manager_username:}")
    private String username;
    @Value("${manager_password:}")
    private String password;
    @Value("${topic:example-metric1}")
    private String topic;
    @Value("${is.security.mode:true}")
    private boolean isSecurityMode;
    // producer config
    @Value("${isAsync:false}")
    private String isAsync;
    // consumer config
    @Value("${consumer.alive.time:180000}")
    private String consumerAliveTime;
    public KafkaProperties() {
    }
    /**
     * Producer configuration
     */
    @Bean(name = "kafkaProducerTemplate")
    public KafkaTemplate kafkaProducerTemplate() {
        Map<String, Object> props = new HashMap<>();
        this.initPropertiesByResources(props);
        this.initProducerProperties(props);
        return new KafkaTemplate<>(new DefaultKafkaProducerFactory<>(props));
    }
    /**
     * Consumer configuration
     */
    @Bean(name = "KafkaListenerContainerFactory")
    public KafkaListenerContainerFactory integratedEnergyKafkaListenerContainerFactory() {
        ConcurrentKafkaListenerContainerFactory factory = new ConcurrentKafkaListenerContainerFactory<>();
        Map<String, Object> props = new HashMap<>();
        this.initConsumerProperties(props);
        factory.setConsumerFactory(new DefaultKafkaConsumerFactory<>(props));
        // Specify the number of concurrent factories.
        factory.setConcurrency(3);
        return factory;
    }
    // Set the parameters with the application.properties file in resources.
    public void initPropertiesByResources(Map<String, Object> properties) {
        // Specify broker connection address.
        if (isEmpty(this.bootstrapServers)) {
            throw new IllegalArgumentException("The bootstrap.servers is null or empty.");
        }
        properties.put(CommonClientConfigs.BOOTSTRAP_SERVERS_CONFIG, this.bootstrapServers);
        // Specify security protocol type.
        properties.put(CommonClientConfigs.SECURITY_PROTOCOL_CONFIG, this.securityProtocol);
        // Specify the authentication mechanism used in the security protocol.
        properties.put(SaslConfigs.SASL_MECHANISM, this.saslMechanism);
        // Specify dynamic JAAS configurations.
        if (this.isSecurityMode) {
            if (isEmpty(this.username)|| isEmpty(this.password)) {
                throw new IllegalArgumentException("The properties manager_username or manager_password is null or empty.");
            }
            String jaasConfig = String.format("org.apache.kafka.common.security.plain.PlainLoginModule required username=%s password=%s;", this.username, this.password);
            properties.put(SaslConfigs.SASL_JAAS_CONFIG, jaasConfig);
        }
        properties.put("topic", this.topic);
        properties.put("isAsync", this.isAsync);
        properties.put("consumer.alive.time", this.consumerAliveTime);
    }
    // Specify the producer properties.
    public void initProducerProperties(Map<String, Object> properties) {
        // Specify retry times.
        properties.put(ProducerConfig.RETRIES_CONFIG, 3);
        // acks=0: If the message is sent to Kafka, it is deemed a successful sending.
        // acks=1: If the message is sent to the Kafka leader partition and written to disk, it is deemed a successful sending.
        // acks=all: If the message is sent to the Kafka leader partition and synchronized by the follower of the partition, it is deemed a successful sending.
        properties.put(ProducerConfig.ACKS_CONFIG, "all");
        // Specify the maximum blocking time (in milliseconds) of KafkaProducer.send() and partitionsFor().
        properties.put(ProducerConfig.MAX_BLOCK_MS_CONFIG, 60000);
        // Specify the maximum number (in bytes) of messages that can be produced in a batch.
        properties.put(ProducerConfig.BATCH_SIZE_CONFIG, 4096);
        // Specify the time the producer waits before sending a message. When the accumulated messages on the production end reach the batch-size, or when the message linger.ms is received, the producer will send the messages to Kafka.
        properties.put(ProducerConfig.LINGER_MS_CONFIG, 1000);
        // Specify the maximum size (in bytes) of the available buffer for the producer.
        properties.put(ProducerConfig.BUFFER_MEMORY_CONFIG, 33554432);
        // Specify the maximum size of each message.
        properties.put(ProducerConfig.MAX_REQUEST_SIZE_CONFIG, 1048576);
        // Enter the client ID.
        properties.put(ProducerConfig.CLIENT_ID_CONFIG, "client-1");
        // Specify the key serialization method.
        properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
        // Specify the value serialization method.
        properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
        // Specify the message compression type: none, lz4, gzip, or snappy.
        properties.put(ProducerConfig.COMPRESSION_TYPE_CONFIG, "none");
    }
    // Specify the consumer properties.
    private Map initConsumerProperties(Map<String, Object> properties) {
        this.initPropertiesByResources(properties);
        // Determine whether to automatically submit the offset.
        properties.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, true);
        // Specify the interval for automatically submitting the offset.
        properties.put(ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG, 1000);
        // Specify the maximum number of messages that can be consumed in a batch.
        properties.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, 100);
        // Specify the consumer group.
        properties.put(ConsumerConfig.GROUP_ID_CONFIG, "testGroup");
        // Specify the session timeout interval of Kafka consumer. If the consumer fails to send a heartbeat message within the specified interval, rebalancing is initiated.
        properties.put(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG, 120000);
        // Specify the request timeout interval, in seconds.
        properties.put(ConsumerConfig.REQUEST_TIMEOUT_MS_CONFIG, 120000);
        // Specify the key deserialization class.
        properties.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        // Specify the value deserialization class.
        properties.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        properties.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "latest");
        return properties;
    }
    public static boolean isEmpty(final CharSequence cs) {
        return cs == null || cs.length() == 0;
    }
}

The KafkaProperties parameter in the example code can be configured in springboot > kafka-examples > module-spring-kafka-examples > src > main > resources > application.properties or added to the application.properties file in the example running environment. If no default value is specified, the parameter is mandatory.

server.port: port listened by the server.
server.address: IP address of the server.
bootstrap.servers: Broker address list of the Kafka cluster. The format is IP address:Port,IP address:Port,IP address:Port. In an IPv6 environment, add square brackets [] to the IP address, for example, [::1]:21007.
security.protocol: authentication protocol used by the Kafka client. The default value is PLAINTEXT.
sasl.mechanism: authentication mechanism used by the Kafka client. The default value is PLAIN.
topic: name of the topic for production and consumption. The default value is example-metric1.
isAsync: whether to use asynchronous production. The default value is false.
consumer.alive.time: lifetime of the consumer thread. The default value is 180000, in milliseconds.
is.security.mode: whether the client connects to the cluster in security mode. The default value is false.

Parent topic: Developing Kafka Applications

Example Code for Spring Boot to Access Kafka Using Spring Kafka Module

Function

Example Code

What is your overall rating for this page?