Wiring for NewTopic collections

[StephenFlavin]

I've recently started playing around with spring-kafka and I like it so far however I was trying to make use of the programmatic creation of new topics when I hit a blocker.
I was hoping to have an app basically verify a bunch of topics are configured and then produce (or consume) to them but to have those topics be something that's passed in as an environment variable e.g.

    @Bean
    public List<NewTopic> topics(@Value("${ENVIRONMENT.topics:topicA,topicB,topicC}") List<String> topics) {
        return topics.stream()
                .map(this::toTopic)
                .collect(Collectors.toList());
    }

    private NewTopic toTopic(String topic) {
        return TopicBuilder.name(topic)
                .partitions(16)
                .replicas(2)
                .build();
    }

Apologies if there is an existing better way to do what I describe, best I've found as a workaround is to write the code using the org.apache.kafka.clients.admin.AdminClient.

[artembilan]

I think we can treat this as a New Feature Request and really add createTopics(NewTopic... topicsToCreate) API into a KafkaAdmin for use-case like this.

The AdminClient.createTopics() is the way to go at the moment.
Another workaround could be achieved with something like this:

		public ContextConfiguration(@Value("${ENVIRONMENT.topics:topicA,topicB,topicC}") String[] topics,
				GenericApplicationContext context) {

			Arrays.stream(topics)
					.forEach(topic ->
							context.registerBean(topic, NewTopic.class,
									() -> TopicBuilder.name(topic)
											.partitions(16)
											.replicas(2)
											.build())
					);
		}

Pay attention that I use String[], not List<String>. Looks like Spring doesn't split over comma for List expectations...

[garyrussell]

We could also do something like add NewTopics (a wrapper around List<NewTopic>).

Unfortunately, List<NewTopic> is not appropriate for a bean definition because of type erasure - we'd end up early accessing all beans of type List<?>. We had a similar problem in RabbitMQ and introduced the Declarables wrapper for collections of Declarable.

[StephenFlavin]

Thanks for the quick response, it would be nice if the KafkaAdmin class had the method as you describe, it would allow for dynamically creating topics after initialisation as well.
@garyrussell suggestion sounds like a good way to handle this too.