According to Kafka documentation:
Kafka provides the guarantee that a topic-partition is assigned to only one consumer within a group.
But I'm observing different behaviour in my service. Here are some details:
I'm using Kafka 2.8 and spring-kafka 2.2.13.
Initially I had one Kafka topic topic.1 with 5 partitions in it, this topic was consumed in my service using @KafkaListener annotation from Spring and ConcurrentKafkaListenerContainerFactory with concurrency == 5. This configuration worked fine for me.
Later I started consuming topic.2 with 3 partitions in it in the same service using the same ConcurrentKafkaListenerContainerFactory and the same group ID.
For some time it also worked correctly but in the one of rebalancings one of the consumer threads was not included in the rebalancing process and for some reasons continued processing previously assigned partition, i.e. a new consumer was assigned to that partition and the old consumer kept processing the same partition, so the same records were processed twice. In the logs I see that records from that partition were consumed and processed twice in two different consumer threads during several days.
After service restart consumers were assigned correctly again.
Here is my code, ConcurrentKafkaListenerContainerFactory bean creation:
@Configuration
public class Config {
@Autowired
private KafkaProperties kafkaProperties;
@Bean
public ConcurrentKafkaListenerContainerFactory<Object, Object> transactionalKafkaContainerFactory(
ConcurrentKafkaListenerContainerFactoryConfigurer configurer,
KafkaTransactionManager kafkaTransactionManager,
ConsumerFactory<Object, Object> consumerFactory) {
ConcurrentKafkaListenerContainerFactory<Object, Object> factory = new ConcurrentKafkaListenerContainerFactory<>();
configurer.configure(factory, consumerFactory);
//If an exception is thrown, then we want to seek back for the whole batch
factory.setBatchErrorHandler(new FixedBackoffSeekToCurrentBatchErrorHandler());
//Enable transactional consumer for exactly-once transitivity
factory.getContainerProperties().setTransactionManager(kafkaTransactionManager);
//Enable batch-processing
factory.getContainerProperties().setAckMode(ContainerProperties.AckMode.BATCH);
factory.setBatchListener(true);
factory.setConcurrency(5);
return factory;
}
@Bean
public KafkaTransactionManager<?, ?> kafkaTransactionManager(
@Qualifier("transactionalProducerFactory") ProducerFactory producerFactory) {
return new KafkaTransactionManager(producerFactory);
}
@Bean
public ProducerFactory<?, ?> transactionalProducerFactory() {
DefaultKafkaProducerFactory<?, ?> producerFactory = new DefaultKafkaProducerFactory<>(
kafkaProperties.buildProducerProperties()
);
producerFactory.setTransactionIdPrefix("tx-");
return producerFactory;
}
}
@KafkaListener-s:
@Service
public class Processor {
@KafkaListener(topics = "topic.1",
groupId = "my-group",
containerFactory = "transactionalKafkaContainerFactory")
public void listener1(List<ConsumerRecord<String, MyObject>> records) {
// process records
}
// Listener for topic.2, added later
@KafkaListener(topics = "topic.2",
groupId = "my-group",
containerFactory = "transactionalKafkaContainerFactory")
public void listener2(List<ConsumerRecord<String, MyObjec>> records) {
// process records
}
}
Here are some logs, one of consumers (consumer-7) was not included in the rebalancing process and continued consuming his old partition (topic.1-3) while new consumer (consumer-4) was assigned to the same partition:
03/30/2022 10:23:47.484 [Consumer clientId=consumer-8, groupId=my-group] Setting newly assigned partitions [topic.1-1, topic.1-0]
03/30/2022 10:23:47.484 [Consumer clientId=consumer-4, groupId=my-group] Setting newly assigned partitions [topic.1-3]
03/30/2022 10:23:47.484 [Consumer clientId=consumer-2, groupId=my-group] Setting newly assigned partitions [topic.2-0]
03/30/2022 10:23:47.484 [Consumer clientId=consumer-6, groupId=my-group] Setting newly assigned partitions [topic.1-4]
03/30/2022 10:23:47.483 [Consumer clientId=consumer-5, groupId=my-group] Setting newly assigned partitions [topic.2-2]
03/30/2022 10:23:47.483 [Consumer clientId=consumer-1, groupId=my-group] Setting newly assigned partitions [topic.1-2]
03/30/2022 10:23:47.483 [Consumer clientId=consumer-3, groupId=my-group] Setting newly assigned partitions [topic.2-1]
...
03/30/2022 10:53:55.728 [Consumer clientId=consumer-7, groupId=my-group] Discovered group coordinator ... (id: ... rack: null)
03/30/2022 10:53:55.627 [Consumer clientId=consumer-7, groupId=my-group] Group coordinator ... (id: ... rack: null) is unavailable or invalid, will attempt rediscovery
03/30/2022 10:53:55.627 [Consumer clientId=consumer-7, groupId=my-group] Discovered group coordinator ... (id: ... rack: null)
03/30/2022 10:53:55.507 [Consumer clientId=consumer-7, groupId=my-group] Group coordinator ... (id: ... rack: null) is unavailable or invalid, will attempt rediscovery
03/30/2022 10:53:55.507 [Consumer clientId=consumer-7, groupId=my-group] Discovered group coordinator ... (id: ... rack: null)
Example of processing the same record twice (domain specific details are excluded):
03/30/2022 11:55:05.144 PM +0300 Processing payload -> EventId: 289f43b4-b07b-4a1f-b768-0453e0c42719, Topic: topic.1, Partition: 3, Offset: 10903844 org.springframework.kafka.KafkaListenerEndpointContainer#1-4-C-1
03/30/2022 11:55:05.143 PM +0300 Processing payload -> EventId: 289f43b4-b07b-4a1f-b768-0453e0c42719, Topic: topic.1, Partition: 3, Offset: 10903844 org.springframework.kafka.KafkaListenerEndpointContainer#1-0-C-1
Note: business logic details are not important here, so I used simple names: topic.1, Processor, my-group etc
My question is how this behaviour can be explained? Why consumer-7 were able to consume messages from its old partition after rebalancing?
Could adding new @KafkaListener with the same group ID cause this issue? (at least I didn't see this behaviour when I had only one @KafkaListener in my service)