spring kafka - @KafkaListener Annotation
This post was migrated from Tistory. You can find the original here.
@KafkaListener Annotation
This is one of the many ways to implement a consumer.
1
2
3
4
@KafkaListener(id = "foo", topics = MSG_TOPIC, clientIdPrefix = "myClientId")
public void listen(String data) {
System.out.println(data);
}
Here’s how it’s used.
id is equivalent to GROUP_ID_CONFIG. So if you specify it on the @KafkaListener, you need to remove the group id setting from the consumer config.
1
2
3
[Consumer clientId=myClientId-0, groupId=foo]
[Consumer clientId=myClientId-1, groupId=foo]
[Consumer clientId=myClientId-2, groupId=foo]
clientIdPrefix produces logs like the ones above when connecting to Kafka.
When factory.setConcurrency(3); is set, and it runs across 3 consumer instances on multiple threads,
it appends the clientId to each one as shown in the log.
@EnableKafka Configuration
To use @KafkaListener, one of your @Configuration classes needs the @EnableKafka annotation, and that requires a container factory used to configure the default ConcurrentMessageListenerContainer.
The default bean name expected is kafkaListenerContainerFactory.
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
@Configuration
@EnableKafka
public class MsgListenerContainerFactory {
public final String GROUP1 = "ConsumerGroup1";
@Bean
KafkaListenerContainerFactory<ConcurrentMessageListenerContainer<Integer, String>>
kafkaListenerContainerFactory() {
ConcurrentKafkaListenerContainerFactory<Integer, String> factory =
new ConcurrentKafkaListenerContainerFactory<>();
factory.setConsumerFactory(consumerFactory());
factory.setConcurrency(3);
factory.getContainerProperties().setPollTimeout(3000);
return factory;
}
@Bean
public ConsumerFactory<Integer, String> consumerFactory() {
return new DefaultKafkaConsumerFactory<>(consumerConfigs());
}
@Bean
public Map<String, Object> consumerConfigs() {
Map<String, Object> props = new HashMap<>();
props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
//props.put(ConsumerConfig.GROUP_ID_CONFIG, GROUP1); //Comment this out if the id is specified on the Listener
props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, IntegerDeserializer.class);
props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
return props;
}
}
Following the default name mentioned in the docs, I registered the kafkaListenerContainerFactory() bean.
AUTO_OFFSET_RESET_CONFIG in consumerConfigs determines the behavior when there’s no prior access or no committed offset.
The default value is latest, which uses the most recent offset, while earliest uses the very first offset.
References
https://docs.spring.io/spring-kafka/reference/html/#kafka-listener-annotation
4.1.4 Receiving Messages @KafkaListener Annotation