What is the simplest Spring Kafka @KafkaListener configuration to consume all records from a set of compacted topics?

Question

I have the names of several compacted Kafka topics (topic1, topic2, ..., topicN) defined in my spring application.yaml file. I want to be able to consume all of the records on each topic partition on startup. The number of partitions on each topic is not known in advance.

The official Spring Kafka 2.6.1 documentation suggests the simplest way to do this is to implement a PartitionFinder and use it in a SpEL expresssion to dynamically look up the number of partitions for a topic, and to then use a * wildcard in the partitions attribute of a @TopicPartition annotation (see Explicit Partition Assignment in the @KafkaListener Annotation documentation):

@KafkaListener(topicPartitions = @TopicPartition(topic = "compacted",
            partitions = "#{@finder.partitions('compacted')}"),
            partitionOffsets = @PartitionOffset(partition = "*", initialOffset = "0")))
public void listen(@Header(KafkaHeaders.RECEIVED_MESSAGE_KEY) String key, String payload) {
    // process record
}

Since I have several topics, the resulting code is very verbose:

@KafkaListener(topicPartitions = {
        @TopicPartition(
                topic = "${topic1}",
                partitions = "#{@finder.partitions('${topic1}')}",
                partitionOffsets = @PartitionOffset(partition = "*", initialOffset = "0")
        ),
        @TopicPartition(
                topic = "${topic2}",
                partitions = "#{@finder.partitions('${topic2}')}",
                partitionOffsets = @PartitionOffset(partition = "*", initialOffset = "0")
        ),
        // and many more @TopicPartitions...
        @TopicPartition(
                topic = "${topicN}",
                partitions = "#{@finder.partitions('${topicN}')}",
                partitionOffsets = @PartitionOffset(partition = "*", initialOffset = "0")
        )
})
public void listen(@Header(KafkaHeaders.RECEIVED_MESSAGE_KEY) String key, String payload) {
    // process record
}

How can I make this repetitive configuration more concise by configuring the topicPartitions attribute of the @KafkaListener annotation with a dynamically generated array of @TopicPartions (one for each of my N topics)?

Gary Russell Gary Russell · Accepted Answer · 2020-09-23T13:22:21

It's not currently possible with @KafkaListener - please open a new feature issue on GitHub.

The only work around I can think of is to programmatically create a listener container from the container factory and create a listener adapter. I can provide an example if you need it.

EDIT

Here is an example:

@SpringBootApplication
public class So64022266Application {

    public static void main(String[] args) {
        SpringApplication.run(So64022266Application.class, args);
    }

    @Bean
    public NewTopic topic1() {
        return TopicBuilder.name("so64022266-1").partitions(10).replicas(1).build();
    }

    @Bean
    public NewTopic topic2() {
        return TopicBuilder.name("so64022266-2").partitions(10).replicas(1).build();
    }

    @Bean
    ConcurrentMessageListenerContainer<String, String> container(@Value("${topics}") String[] topics,
            PartitionFinder finder,
            ConcurrentKafkaListenerContainerFactory<String, String> factory,
            MyListener listener) throws Exception {

        MethodKafkaListenerEndpoint<String, String> endpoint = endpoint(topics, finder, listener);
        ConcurrentMessageListenerContainer<String, String> container = factory.createListenerContainer(endpoint);
        container.getContainerProperties().setGroupId("someGroup");
        return container;
    }

    @Bean
    MethodKafkaListenerEndpoint<String, String> endpoint(String[] topics, PartitionFinder finder,
            MyListener listener) throws NoSuchMethodException {

        MethodKafkaListenerEndpoint<String, String> endpoint = new MethodKafkaListenerEndpoint<>();
        endpoint.setBean(listener);
        endpoint.setMethod(MyListener.class.getDeclaredMethod("listen", String.class, String.class));
        endpoint.setTopicPartitions(Arrays.stream(topics)
            .flatMap(topic -> finder.partitions(topic))
            .toArray(TopicPartitionOffset[]::new));
        endpoint.setMessageHandlerMethodFactory(methodFactory());
        return endpoint;
    }

    @Bean
    DefaultMessageHandlerMethodFactory methodFactory() {
        return new DefaultMessageHandlerMethodFactory();
    }

    @Bean
    public ApplicationRunner runner(KafkaTemplate<String, String> template,
            ConcurrentMessageListenerContainer<String, String> container) {

        return args -> {
            System.out.println(container.getAssignedPartitions());
            template.send("so64022266-1", "key1", "foo");
            template.send("so64022266-2", "key2", "bar");
        };
    }

}

@Component
class MyListener {

    public void listen(@Header(KafkaHeaders.RECEIVED_MESSAGE_KEY) String key, String payload) {
        System.out.println(key + ":" + payload);
    }

}

@Component
class PartitionFinder {

    private final ConsumerFactory<String, String> consumerFactory;

    public PartitionFinder(ConsumerFactory<String, String> consumerFactory) {
        this.consumerFactory = consumerFactory;
    }

    public Stream<TopicPartitionOffset> partitions(String topic) {
        System.out.println("+" + topic + "+");
        try (Consumer<String, String> consumer = consumerFactory.createConsumer()) {
            return consumer.partitionsFor(topic).stream()
                    .map(part -> new TopicPartitionOffset(topic, part.partition(), 0L));
        }
    }

}

topics=so64022266-1, so64022266-2

If you need to deal with tombstone records (null values) we need to enhance the handler factory; we currently don't expose the framework's handler factory.

What is the simplest Spring Kafka @KafkaListener configuration to consume all records from a set of compacted topics?

1 Answers