[FIXED] Spring Boot @KafkaListener not consuming from all partitions

Issue

I've got a kafka topic setup with 9 partitions. With my Kafka Consumer annotated like so

@KafkaListener(topics = "myTopic",
              groupId = "topic-group",
              autoStartup = "true") //if this is false listener doesn’t startup
public void consume(ConsumerRecord<?, ?> record) {

when my application starts up, I see

11-08-2022 17:14:53.290 [33m[main][0;39m sub= gsk= [34mINFO [0;39m c.a.b.d.e.push.Application.logStarted - Started Application in 8.221 seconds (JVM running for 9.007)
11-08-2022 17:14:56.331 [33m[org.springframework.kafka.KafkaListenerEndpointContainer#0-0-C-1][0;39m sub= gsk= [34mINFO [0;39m o.s.k.l.KafkaMessageListenerContainer.info - topic-group: partitions assigned: [myTopic-2, myTopic-1, myTopic-4, myTopic-3, myTopic-0]

I am fairly new to this, my understanding is the KafkaMessageListenerContainer somehow is now subscribed/polling/listening to 5 partitions. Why? Why isn't it looking at all the 9 partitions?

When I add concurrency to the kafka listener

@KafkaListener(topics = "myTopic",
               concurrency = "9",    
               groupId = "topic-group",
                autoStartup = "true")

My startup logs are :

11-08-2022 18:41:20.586 [33m[main][0;39m sub= gsk= [34mINFO [0;39m c.a.b.d.e.push.Application.logStarted - Started Application in 7.161 seconds (JVM running for 7.87)
11-08-2022 18:41:23.066 [33m[org.springframework.kafka.KafkaListenerEndpointContainer#0-5-C-1][0;39m sub= gsk= [34mINFO [0;39m o.s.k.l.KafkaMessageListenerContainer.info - topic-group: partitions assigned: [myTopic-5]
11-08-2022 18:41:23.071 [33m[org.springframework.kafka.KafkaListenerEndpointContainer#0-2-C-1][0;39m sub= gsk= [34mINFO [0;39m o.s.k.l.KafkaMessageListenerContainer.info - topic-group: partitions assigned: [myTopic-2]
11-08-2022 18:41:23.141 [33m[org.springframework.kafka.KafkaListenerEndpointContainer#0-7-C-1][0;39m sub= gsk= [34mINFO [0;39m o.s.k.l.KafkaMessageListenerContainer.info - topic-group: partitions assigned: [myTopic-7]
11-08-2022 18:41:23.141 [33m[org.springframework.kafka.KafkaListenerEndpointContainer#0-6-C-1][0;39m sub= gsk= [34mINFO [0;39m o.s.k.l.KafkaMessageListenerContainer.info - topic-group: partitions assigned: [myTopic-6]
11-08-2022 18:41:23.141 [33m[org.springframework.kafka.KafkaListenerEndpointContainer#0-3-C-1][0;39m sub= gsk= [34mINFO [0;39m o.s.k.l.KafkaMessageListenerContainer.info - topic-group: partitions assigned: [myTopic-3]
11-08-2022 18:41:23.141 [33m[org.springframework.kafka.KafkaListenerEndpointContainer#0-4-C-1][0;39m sub= gsk= [34mINFO [0;39m o.s.k.l.KafkaMessageListenerContainer.info - topic-group: partitions assigned: [myTopic-4]
11-08-2022 18:41:23.141 [33m[org.springframework.kafka.KafkaListenerEndpointContainer#0-1-C-1][0;39m sub= gsk= [34mINFO [0;39m o.s.k.l.KafkaMessageListenerContainer.info - topic-group: partitions assigned: [myTopic-1]
11-08-2022 18:41:23.146 [33m[org.springframework.kafka.KafkaListenerEndpointContainer#0-8-C-1][0;39m sub= gsk= [34mINFO [0;39m o.s.k.l.KafkaMessageListenerContainer.info - topic-group: partitions assigned: [myTopic-8]
11-08-2022 18:41:23.146 [33m[org.springframework.kafka.KafkaListenerEndpointContainer#0-0-C-1][0;39m sub= gsk= [34mINFO [0;39m o.s.k.l.KafkaMessageListenerContainer.info - topic-group: partitions assigned: [myTopic-0]

This starts up 9 containers - which I think isn't the right way to go about this and I dont want to bind myself to 9 partitions.

How do I consume from all 9 partitions?

P.S: Tagging reactor-kafka as the producer is reactive and works alright.

Solution

partitions assigned: [myTopic-2, myTopic-1, myTopic-4, myTopic-3, myTopic-0]

Most likely you had another instance running that is assigned the other 4 partitions.

But it was not running when you ran the version with concurrency = 9.

If that's not the case, provide an MCRE so we can see what's wrong.

PS: autoStartup is true by default.

Answered By - Gary Russell
Answer Checked By - Cary Denson (JavaFixing Admin)