Thanks to suggestions by @Gary Russel, I was able to come up with the following solution which creates a @KafkaListener bean instance (or consumer thread) per Kafka topic. This way, if there is an issue with messages belonging to a particular topic, it will not affect the processing of other topics.
Note - The following code throws a InstanceAlreadyExistsException exception during startup. However, this does not seem to affect the functionality. Using the log outputs I'm able to verify that there is one bean instance (or thread) per topic, and they are able to process messages.
@SpringBootApplication
@EnableScheduling
@Slf4j
public class KafkaConsumerApp {
public static void main(String[] args) {
log.info("Starting spring boot KafkaConsumerApp..");
SpringApplication.run(KafkaConsumerApp.class, args);
}
}
@EnableKafka
@Configuration
public class KafkaConfiguration {
private final KafkaProperties kafkaProperties;
@Value("${kafka.brokers:localhost:9092}")
private String bootstrapServer;
@Value("${kafka.consumerClientId}")
private String consumerClientId;
@Value("${kafka.consumerGroupId}")
private String consumerGroupId;
@Value("${kafka.topicMonitorClientId}")
private String topicMonitorClientId;
@Value("${kafka.topicMonitorGroupId}")
private String topicMonitorGroupId;
@Autowired
private ConfigurableApplicationContext context;
@Autowired
public KafkaConfiguration( KafkaProperties kafkaProperties ) {
this.kafkaProperties = kafkaProperties;
}
@Bean
public ConcurrentKafkaListenerContainerFactory<String, String> kafkaListenerContainerFactory() {
ConcurrentKafkaListenerContainerFactory<String, String> factory = new ConcurrentKafkaListenerContainerFactory<>();
factory.setConsumerFactory( consumerFactory( consumerClientId, consumerGroupId ) );
factory.getContainerProperties().setAckMode( ContainerProperties.AckMode.MANUAL );
return factory;
}
@Bean
public ConcurrentKafkaListenerContainerFactory<String, String> topicMonitorContainerFactory() {
ConcurrentKafkaListenerContainerFactory<String, String> factory = new ConcurrentKafkaListenerContainerFactory<>();
factory.setConsumerFactory( consumerFactory( topicMonitorClientId, topicMonitorGroupId ) );
factory.getContainerProperties().setAckMode( ContainerProperties.AckMode.MANUAL );
factory.getContainerProperties().setConsumerRebalanceListener( new KafkaRebalanceListener( context ) );
return factory;
}
private ConsumerFactory<String, String> consumerFactory( String clientId, String groupId ) {
Map<String, Object> config = new HashMap<>();
config.putAll( kafkaProperties.buildConsumerProperties() );
config.put( ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServer );
config.put( ConsumerConfig.CLIENT_ID_CONFIG, clientId );
config.put( ConsumerConfig.GROUP_ID_CONFIG, groupId );
config.put( ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, false ); // needs to be turned off for rebalancing during topic addition and deletion
// check -> https://stackguides.com/questions/56264681/is-it-possible-to-have-one-kafka-consumer-thread-per-topic/56274988?noredirect=1#comment99401765_56274988
return new DefaultKafkaConsumerFactory<>( config, new StringDeserializer(), new StringDeserializer() );
}
}
@Configuration
public class KafkaListenerConfiguration {
@Bean
@Scope("prototype")
public KafkaMessageListener kafkaMessageListener() {
return new KafkaMessageListener();
}
}
@Slf4j
public class KafkaMessageListener {
/*
* This is the actual message listener that will process messages. It will be instantiated per topic.
*/
@KafkaListener( topics = "${topic}", containerFactory = "kafkaListenerContainerFactory" )
public void receiveHyperscalerMessage( ConsumerRecord<String, String> record, Acknowledgment acknowledgment, Consumer<String, String> consumer ) {
log.debug("Kafka message - ThreadName={}, Hashcode={}, Partition={}, Topic={}, Value={}",
Thread.currentThread().getName(), Thread.currentThread().hashCode(), record.partition(), record.topic(), record.value() );
// do processing
// this is just a sample acknowledgment. it can be optimized to acknowledge after processing a batch of messages.
acknowledgment.acknowledge();
}
}
@Service
public class KafkaTopicMonitor {
/*
* The main purpose of this listener is to detect the rebalance events on our topic pattern, so that
* we can create a listener bean instance (consumer thread) per topic.
*
* Note that we use the wildcard topic pattern here.
*/
@KafkaListener( topicPattern = ".*abc.def.ghi", containerFactory = "topicMonitorContainerFactory" )
public void monitorTopics( ConsumerRecord<String, String> record ) {
// do nothing
}
}
@Slf4j
public class KafkaRebalanceListener implements ConsumerAwareRebalanceListener {
private static final ConcurrentMap<String, KafkaMessageListener> listenerMap = new ConcurrentHashMap<>();
private final ConfigurableApplicationContext context;
public KafkaRebalanceListener( ConfigurableApplicationContext context ) {
this.context = context;
}
public void onPartitionsRevokedBeforeCommit(Consumer<?, ?> consumer, Collection<TopicPartition> partitions) {
// do nothing
}
public void onPartitionsRevokedAfterCommit(Consumer<?, ?> consumer, Collection<TopicPartition> partitions) {
// do nothing
}
public void onPartitionsAssigned(Consumer<?, ?> consumer, Collection<TopicPartition> partitions) {
log.info("OnPartitionsAssigned - partitions={} - {}", partitions.size(), partitions);
Properties props = new Properties();
context.getEnvironment().getPropertySources().addLast( new PropertiesPropertySource("topics", props) );
for( TopicPartition tp: partitions ) {
listenerMap.computeIfAbsent( tp.topic(), key -> {
log.info("Creating messageListener bean instance for topic - {}", key );
props.put( "topic", key );
// create new KafkaMessageListener bean instance
return context.getBean( "kafkaMessageListener", KafkaMessageListener.class );
});
}
}
}
