Auto-create topic with N partitions

kafka/src/main/java/io/micronaut/configuration/kafka/admin/KafkaNewTopics.java

@@ -0,0 +1,91 @@
+/*
+ * Copyright 2017-2020 original authors
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package io.micronaut.configuration.kafka.admin;
+
+import io.micronaut.context.ApplicationContext;
+import io.micronaut.context.annotation.Context;
+import io.micronaut.context.annotation.Requires;
+import io.micronaut.core.annotation.Experimental;
+import io.micronaut.core.annotation.NonNull;
+import io.micronaut.core.annotation.Nullable;
+import org.apache.kafka.clients.admin.AdminClient;
+import org.apache.kafka.clients.admin.CreateTopicsOptions;
+import org.apache.kafka.clients.admin.CreateTopicsResult;
+import org.apache.kafka.clients.admin.NewTopic;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.List;
+import java.util.Optional;
+
+import static java.util.function.Predicate.not;
+
+/**
+ * Creates Kafka topics via {@link AdminClient}.
+ *
+ * @author Guillermo Calvo
+ * @since 5.2
+ */
+@Context
+@Requires(bean = AdminClient.class)
+public class KafkaNewTopics {
+
+    private static final Logger LOG = LoggerFactory.getLogger(KafkaNewTopics.class);
+
+    @Nullable
+    private final CreateTopicsOptions options;
+
+    @Nullable
+    private final CreateTopicsResult result;
+
+    /**
+     * @param context The application context.
+     * @param options Optional {@link CreateTopicsOptions}.
+     * @param topics  Optional list of {@link NewTopic} beans.
+     */
+    KafkaNewTopics(
+        @NonNull ApplicationContext context,
+        @Nullable CreateTopicsOptions options,
+        @Nullable List<NewTopic> topics
+    ) {
+        this.options = options;
+        this.result = Optional.ofNullable(topics).filter(not(List::isEmpty))
+            .map(t -> createNewTopics(context, t))
+            .orElse(null);
+    }
+
+    /**
+     * @return An optional {@link CreateTopicsOptions} that was used if new topics were created.
+     */
+    @Experimental
+    public Optional<CreateTopicsOptions> getOptions() {
+        return Optional.ofNullable(options);
+    }
+
+    /**
+     * @return An optional {@link CreateTopicsResult} if new topics were created.
+     */
+    @Experimental
+    public Optional<CreateTopicsResult> getResult() {
+        return Optional.ofNullable(result);
+    }
+
+    private CreateTopicsResult createNewTopics(@NonNull ApplicationContext context, @NonNull List<NewTopic> topics) {
+        LOG.info("Creating new topics: {}", topics);
+        final AdminClient adminClient = context.getBean(AdminClient.class);
+        return adminClient.createTopics(topics, getOptions().orElseGet(CreateTopicsOptions::new));
+    }
+}

kafka/src/test/groovy/io/micronaut/configuration/kafka/admin/KafkaNewTopicsSpec.groovy

@@ -0,0 +1,57 @@
+package io.micronaut.configuration.kafka.admin
+
+import io.micronaut.configuration.kafka.AbstractKafkaContainerSpec
+import io.micronaut.context.annotation.Bean
+import io.micronaut.context.annotation.Factory
+import io.micronaut.context.annotation.Requires
+import org.apache.kafka.clients.admin.CreateTopicsOptions
+import org.apache.kafka.clients.admin.NewTopic
+import spock.util.concurrent.PollingConditions
+
+class KafkaNewTopicsSpec extends AbstractKafkaContainerSpec {
+
+    final static TOPIC_1 = 'new-topic-1'
+    final static TOPIC_2 = 'new-topic-2'
+    final static TOPIC_3 = '$illegal/topic:name!'
+
+    void "create kafka topics"() {
+        given:
+        final KafkaNewTopics kafkaNewTopics = context.getBean(KafkaNewTopics)
+
+        expect:
+        kafkaNewTopics.options.isPresent()
+        kafkaNewTopics.result.isPresent()
+
+        and:
+        final options = kafkaNewTopics.options.get()
+        options.timeoutMs == 5000
+        options.validateOnly == true
+        options.retryOnQuotaViolation == false
+
+        and:
+        final result = kafkaNewTopics.result.get()
+        new PollingConditions(timeout: 10).eventually {
+            result.all().done == true
+        }
+        result.numPartitions(TOPIC_1).get() == 1
+        result.numPartitions(TOPIC_2).get() == 2
+        result.values()[TOPIC_3].completedExceptionally == true
+    }
+
+    @Requires(property = 'spec.name', value = 'KafkaNewTopicsSpec')
+    @Factory
+    static class MyTopicFactory {
+
+        @Bean
+        CreateTopicsOptions options() { new CreateTopicsOptions().timeoutMs(5000).validateOnly(true).retryOnQuotaViolation(false) }
+
+        @Bean
+        NewTopic topic1() { new NewTopic(TOPIC_1, Optional.of(1), Optional.empty()) }
+
+        @Bean
+        NewTopic topic2() { new NewTopic(TOPIC_2, Optional.of(2), Optional.empty()) }
+
+        @Bean
+        NewTopic topic3() { new NewTopic(TOPIC_3, Optional.of(4), Optional.empty()) }
+    }
+}

src/main/docs/guide/kafkaApplications/kafkaNewTopics.adoc

@@ -0,0 +1,14 @@
+
+You can automatically add topics to the broker when your application starts. To do so, you can add a link:{kafkaapi}/org/apache/kafka/clients/admin/NewTopic[`NewTopic`] bean for each topic you want to create. `NewTopic` instances let you specify the name, the number of partitions, the replication factor, the replicas assignments and the configuration properties you want to associate with the new topic. Additionally, you can add one link:{kafkaapi}/org/apache/kafka/clients/admin/CreateTopicsOptions[`CreateTopicsOptions`] bean that will be used when the new topics are created.
+
+.Creating New Kafka Topics with Options
+
+snippet::io.micronaut.kafka.docs.admin.MyTopicFactory[tags = 'imports,clazz']
+
+NOTE: Creating topics is not a transactional operation, so it may succeed for some topics while fail for others. This operation also executes asynchronously, so it may take several seconds until all the brokers become aware that the topics have been created.
+
+If you ever need to check if the operation has completed, you can `@Inject` or retrieve the api:io.micronaut.configuration.kafka.admin.KafkaNewTopics[] bean from the application context and then retrieve the link:{kafkaapi}/org/apache/kafka/clients/admin/CreateTopicsResult[operation result] that Kafka returned when the topics were created.
+
+.Checking if Topic Creation is Done
+
+snippet::io.micronaut.kafka.docs.admin.MyTopicFactoryTest[tags = 'result']

src/main/docs/guide/toc.yml

@@ -29,6 +29,7 @@ kafkaApplications:
   kafkaHealth: Kafka Health Checks
   kafkaMetrics: Kafka Metrics
   kafkaTracing: Kafka Distributed Tracing
+  kafkaNewTopics: Creating New Topics
   kafkaDisabled: Disabling Micronaut-Kafka
 kafkaStreams:
   title: Kafka Streams

test-suite-groovy/src/test/groovy/io/micronaut/kafka/docs/admin/MyTopicFactory.groovy

@@ -0,0 +1,33 @@
+package io.micronaut.kafka.docs.admin
+
+// tag::imports[]
+import io.micronaut.context.annotation.Bean
+import io.micronaut.context.annotation.Factory
+import io.micronaut.context.annotation.Requires
+import org.apache.kafka.clients.admin.AdminClient
+import org.apache.kafka.clients.admin.CreateTopicsOptions
+import org.apache.kafka.clients.admin.NewTopic
+// end::imports[]
+
+@Requires(property = "spec.name", value = "MyTopicFactoryTest")
+// tag::clazz[]
+@Requires(bean = AdminClient)
+@Factory
+class MyTopicFactory {
+
+    @Bean
+    CreateTopicsOptions options() {
+        new CreateTopicsOptions().timeoutMs(5000).validateOnly(true).retryOnQuotaViolation(false)
+    }
+
+    @Bean
+    NewTopic topic1() {
+        new NewTopic("my-new-topic-1", 1, (short) 1)
+    }
+
+    @Bean
+    NewTopic topic2() {
+        new NewTopic("my-new-topic-2", 2, (short) 1)
+    }
+}
+// end::clazz[]

test-suite-groovy/src/test/groovy/io/micronaut/kafka/docs/admin/MyTopicFactoryTest.groovy

@@ -0,0 +1,32 @@
+package io.micronaut.kafka.docs.admin
+
+import io.micronaut.configuration.kafka.admin.KafkaNewTopics
+import io.micronaut.context.annotation.Property
+import io.micronaut.test.extensions.spock.annotation.MicronautTest
+import jakarta.inject.Inject
+import spock.lang.Specification
+import spock.util.concurrent.PollingConditions
+
+@MicronautTest
+@Property(name = "spec.name", value = "MyTopicFactoryTest")
+class MyTopicFactoryTest extends Specification {
+
+    @Inject
+    KafkaNewTopics newTopics
+
+    void "test new topics"() {
+        expect:
+        new PollingConditions(timeout: 5).eventually {
+            areNewTopicsDone(newTopics)
+            final result = newTopics.getResult().orElseThrow()
+            result.numPartitions("my-new-topic-1").get() == 1
+            result.numPartitions("my-new-topic-2").get() == 2
+        }
+    }
+
+    // tag::result[]
+    boolean areNewTopicsDone(KafkaNewTopics newTopics) {
+        newTopics.result.map { it.all().isDone() }.orElse(false)
+    }
+    // end::result[]
+}

test-suite-kotlin/src/test/kotlin/io/micronaut/kafka/docs/admin/MyTopicFactory.kt

@@ -0,0 +1,33 @@
+package io.micronaut.kafka.docs.admin
+
+// tag::imports[]
+import io.micronaut.context.annotation.Bean
+import io.micronaut.context.annotation.Factory
+import io.micronaut.context.annotation.Requires
+import org.apache.kafka.clients.admin.AdminClient
+import org.apache.kafka.clients.admin.CreateTopicsOptions
+import org.apache.kafka.clients.admin.NewTopic
+// end::imports[]
+
+@Requires(property = "spec.name", value = "MyTopicFactoryTest")
+// tag::clazz[]
+@Requires(bean = AdminClient::class)
+@Factory
+class MyTopicFactory {
+
+    @Bean
+    fun options(): CreateTopicsOptions {
+        return CreateTopicsOptions().timeoutMs(5000).validateOnly(true).retryOnQuotaViolation(false)
+    }
+
+    @Bean
+    fun topic1(): NewTopic {
+        return NewTopic("my-new-topic-1", 1, 1)
+    }
+
+    @Bean
+    fun topic2(): NewTopic {
+        return NewTopic("my-new-topic-2", 2, 1)
+    }
+}
+// end::clazz[]

test-suite-kotlin/src/test/kotlin/io/micronaut/kafka/docs/admin/MyTopicFactoryTest.kt

@@ -0,0 +1,35 @@
+package io.micronaut.kafka.docs.admin
+
+import io.micronaut.configuration.kafka.admin.KafkaNewTopics
+import io.micronaut.context.ApplicationContext
+import org.awaitility.Awaitility
+import org.junit.jupiter.api.Assertions.assertEquals
+import org.junit.jupiter.api.Test
+import java.util.concurrent.ExecutionException
+import java.util.concurrent.TimeUnit
+
+internal class MyTopicFactoryTest {
+
+    @Test
+    @Throws(ExecutionException::class, InterruptedException::class)
+    fun testNewTopics() {
+        ApplicationContext.run(
+            mapOf(
+                "kafka.enabled" to "true",
+                "spec.name" to "MyTopicFactoryTest"
+            )
+        ).use { ctx ->
+            val newTopics = ctx.getBean(KafkaNewTopics::class.java)
+            Awaitility.await().atMost(5, TimeUnit.SECONDS).until { areNewTopicsDone(newTopics) }
+            val result = newTopics.result.orElseThrow()
+            assertEquals(1, result.numPartitions("my-new-topic-1").get())
+            assertEquals(2, result.numPartitions("my-new-topic-2").get())
+        }
+    }
+
+    // tag::result[]
+    fun areNewTopicsDone(newTopics: KafkaNewTopics): Boolean {
+        return newTopics.result.map { it.all().isDone }.orElse(false)
+    }
+    // end::result[]
+}

test-suite/src/test/java/io/micronaut/kafka/docs/admin/MyTopicFactory.java

@@ -0,0 +1,33 @@
+package io.micronaut.kafka.docs.admin;
+
+// tag::imports[]
+import io.micronaut.context.annotation.Bean;
+import io.micronaut.context.annotation.Factory;
+import io.micronaut.context.annotation.Requires;
+import org.apache.kafka.clients.admin.AdminClient;
+import org.apache.kafka.clients.admin.CreateTopicsOptions;
+import org.apache.kafka.clients.admin.NewTopic;
+// end::imports[]
+
+@Requires(property = "spec.name", value = "MyTopicFactoryTest")
+// tag::clazz[]
+@Requires(bean = AdminClient.class)
+@Factory
+public class MyTopicFactory {
+
+    @Bean
+    CreateTopicsOptions options() {
+        return new CreateTopicsOptions().timeoutMs(5000).validateOnly(true).retryOnQuotaViolation(false);
+    }
+
+    @Bean
+    NewTopic topic1() {
+        return new NewTopic("my-new-topic-1", 1, (short) 1);
+    }
+
+    @Bean
+    NewTopic topic2() {
+        return new NewTopic("my-new-topic-2", 2, (short) 1);
+    }
+}
+// end::clazz[]

test-suite/src/test/java/io/micronaut/kafka/docs/admin/MyTopicFactoryTest.java

@@ -0,0 +1,36 @@
+package io.micronaut.kafka.docs.admin;
+
+import io.micronaut.configuration.kafka.admin.KafkaNewTopics;
+import io.micronaut.context.ApplicationContext;
+import org.apache.kafka.clients.admin.CreateTopicsResult;
+import org.junit.jupiter.api.Test;
+
+import java.util.Map;
+import java.util.concurrent.ExecutionException;
+
+import static java.util.concurrent.TimeUnit.SECONDS;
+import static org.awaitility.Awaitility.await;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+
+class MyTopicFactoryTest {
+
+    @Test
+    void testNewTopics() throws ExecutionException, InterruptedException {
+        try (ApplicationContext ctx = ApplicationContext.run(Map.of(
+            "kafka.enabled", "true",
+            "spec.name", "MyTopicFactoryTest"
+        ))) {
+            final KafkaNewTopics newTopics = ctx.getBean(KafkaNewTopics.class);
+            await().atMost(5, SECONDS).until(() -> areNewTopicsDone(newTopics));
+            final CreateTopicsResult result = newTopics.getResult().orElseThrow();
+            assertEquals(1, result.numPartitions("my-new-topic-1").get());
+            assertEquals(2, result.numPartitions("my-new-topic-2").get());
+        }
+    }
+
+    // tag::result[]
+    boolean areNewTopicsDone(KafkaNewTopics newTopics) {
+        return newTopics.getResult().map(result -> result.all().isDone()).orElse(false);
+    }
+    // end::result[]
+}