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.
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.