From 4fc55b706fce1a8e8eb8b014d6a4f556c3a9c573 Mon Sep 17 00:00:00 2001
From: Adam <897017+aSemy@users.noreply.github.com>
Date: Tue, 25 Jan 2022 00:11:38 +0100
Subject: [PATCH] only try to create topics if there are topics to be created

---
 .../adamko/kafkatorio/admin/CreateTopics.kt   | 41 +++++++++----------
 1 file changed, 20 insertions(+), 21 deletions(-)

diff --git a/modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/admin/CreateTopics.kt b/modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/admin/CreateTopics.kt
index 5f25d92..0857e83 100644
--- a/modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/admin/CreateTopics.kt
+++ b/modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/admin/CreateTopics.kt
@@ -30,6 +30,8 @@ object CreateTopics {
 
     KafkatorioPacket.PacketType.values().forEach { packetType ->
 
+      val currentTopics = currentTopics().map { it.name() }
+
       val kafkatorioTopics = buildSet {
         add(KafkatorioTopology.sourceTopic)
         when (packetType) {
@@ -42,33 +44,30 @@ object CreateTopics {
           KafkatorioPacket.PacketType.PROTOTYPES ->
             add("kafkatorio.${packetType.name}.all")
         }
-      }
+      }.minus(currentTopics.toSet())
 
-      val currentTopics = currentTopics().map { it.name() }
+      if (currentTopics.isNotEmpty()) {
 
-      val result = kafkatorioTopics
-        .filterNot { it in currentTopics }
-        .run {
-          logger.info("Creating $size topics")
-          kafkaAdmin.createTopics { this }
+        logger.info("Creating ${kafkatorioTopics.size} topics")
+        val result = kafkaAdmin.createTopics { kafkatorioTopics }
+
+        // wait for all brokers to become aware of the created topics
+        Thread.sleep(TimeUnit.SECONDS.toMillis(5))
+
+        result.values().forEach {
+          logger.info("Created topic: ${it.key} ${it.value.get()}")
         }
 
-      // wait for all brokers to become aware of the created topics
-      Thread.sleep(TimeUnit.SECONDS.toMillis(5))
+        Thread.sleep(TimeUnit.SECONDS.toMillis(5))
 
-      result.values().forEach {
-        logger.info("Created topic: ${it.key} ${it.value.get()}")
+        logger.info("Listing topics")
+        kafkaAdmin.listTopics()
+          .listings()
+          .get()
+          .forEach {
+            logger.info("Topic: $it")
+          }
       }
-
-      Thread.sleep(TimeUnit.SECONDS.toMillis(5))
-
-      logger.info("Listing topics")
-      kafkaAdmin.listTopics()
-        .listings()
-        .get()
-        .forEach {
-          logger.info("Topic: $it")
-        }
     }
   }