I would like to create different consumer groups dynamically for my kafka listener, therefore I have given up the standard way with @KafkaListener and went throw the manual way with a ContainerProperties which is working fine like the example below. However I can't use setRecordFilterStrategy because it is only present in the kafka listener version ConcurrentKafkaListenerContainerFactory. Do you know a way to do it or another to have dynamic consumer group and the possibility to use record filter strategy at the same time ?
Regards, Luc
Map<String, Object> consumerConfig = ImmutableMap.of(
BOOTSTRAP_SERVERS_CONFIG, "brokerAddress",
GROUP_ID_CONFIG, "groupId"
);
DefaultKafkaConsumerFactory<String, String> kafkaConsumerFactory =
new DefaultKafkaConsumerFactory<>(
consumerConfig,
new StringDeserializer(),
new StringDeserializer());
ContainerProperties containerProperties = new ContainerProperties("topicName");
containerProperties.setMessageListener((MessageListener<String, String>) record -> {
//do something with received record
}
ConcurrentMessageListenerContainer container =
new ConcurrentMessageListenerContainer<>(
kafkaConsumerFactory,
containerProperties);
container.start();