dynamic consumer group in kafka listeners combined with RecordFilterStrategy

Viewed 157

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();
1 Answers
Related