Repository: skiptests/astraea
Pull request #96 (2021-11-05 14:48:52)

Enable to create topic repeatedly
之前的版本我們允許重複建立topic (透過檢查，存在則忽略）

現在這個版本還原之前的邏輯，並新增更完整的確認，包含:

1. number of partitions
1. number of replicas
1. all configs

上述只要有任何一個不一樣則拋出錯誤


Patch:
diff --git a/app/src/main/java/org/astraea/topic/Builder.java b/app/src/main/java/org/astraea/topic/Builder.java
index 1e62ff97..ef3e7a20 100644
--- a/app/src/main/java/org/astraea/topic/Builder.java
+++ b/app/src/main/java/org/astraea/topic/Builder.java
@@ -1,6 +1,7 @@
 package org.astraea.topic;
 
 import java.util.*;
+import java.util.function.Function;
 import java.util.stream.Collectors;
 import org.apache.kafka.clients.admin.*;
 import org.apache.kafka.common.Node;
@@ -288,19 +289,27 @@ public class Builder {
 
     @Override
     public Creator creator() {
-      return new CreatorImpl(admin);
+      return new CreatorImpl(
+          admin, topic -> this.replicas(Set.of(topic)), topic -> topics().get(topic));
     }
   }
 
   private static class CreatorImpl implements Creator {
     private final Admin admin;
+    private final Function<String, Map<TopicPartition, List<Replica>>> replicasGetter;
+    private final Function<String, Map<String, String>> configsGetter;
     private String topic;
     private int numberOfPartitions = 1;
     private short numberOfReplicas = 1;
     private final Map<String, String> configs = new HashMap<>();
 
-    CreatorImpl(Admin admin) {
+    CreatorImpl(
+        Admin admin,
+        Function<String, Map<TopicPartition, List<Replica>>> replicasGetter,
+        Function<String, Map<String, String>> configsGetter) {
       this.admin = admin;
+      this.replicasGetter = replicasGetter;
+      this.configsGetter = configsGetter;
     }
 
     @Override
@@ -329,6 +338,48 @@ public class Builder {
 
     @Override
     public void create() {
+      if (Utils.handleException(() -> admin.listTopics().names().get()).contains(topic)) {
+        var partitionReplicas = replicasGetter.apply(topic);
+        partitionReplicas.forEach(
+            (tp, replicas) -> {
+              if (replicas.size() != numberOfReplicas)
+                throw new IllegalArgumentException(
+                    topic
+                        + " is existent but its replicas: "
+                        + replicas.size()
+                        + " is not equal to expected: "
+                        + numberOfReplicas);
+            });
+        var result =
+            Utils.handleException(() -> admin.describeTopics(Set.of(topic)).all().get().get(topic));
+        if (result.partitions().size() != numberOfPartitions)
+          throw new IllegalArgumentException(
+              topic
+                  + " is existent but its partitions: "
+                  + result.partitions().size()
+                  + " is not equal to expected: "
+                  + numberOfReplicas);
+
+        var actualConfigs = configsGetter.apply(topic);
+        this.configs.forEach(
+            (key, value) -> {
+              if (!actualConfigs.get(key).equals(value))
+                throw new IllegalArgumentException(
+                    topic
+                        + " is existent but its config: <"
+                        + key
+                        + ", "
+                        + actualConfigs.get(key)
+                        + "> is not equal to expected: "
+                        + key
+                        + ", "
+                        + value);
+            });
+
+        // ok, the existent topic is totally equal to what we want to create.
+        return;
+      }
+
       Utils.handleException(
           () ->
               admin


Test patch:
diff --git a/app/src/test/java/org/astraea/topic/TopicAdminTest.java b/app/src/test/java/org/astraea/topic/TopicAdminTest.java
index 166ccf22..b31a82a4 100644
--- a/app/src/test/java/org/astraea/topic/TopicAdminTest.java
+++ b/app/src/test/java/org/astraea/topic/TopicAdminTest.java
@@ -7,6 +7,7 @@ import java.util.Set;
 import java.util.concurrent.ExecutionException;
 import java.util.concurrent.TimeUnit;
 import java.util.stream.Collectors;
+import java.util.stream.IntStream;
 import org.apache.kafka.common.config.TopicConfig;
 import org.astraea.Utils;
 import org.astraea.producer.Producer;
@@ -30,6 +31,46 @@ public class TopicAdminTest extends RequireBrokerCluster {
     }
   }
 
+  @Test
+  void testCreateTopicRepeatedly() throws IOException {
+    var topicName = "testCreateTopicRepeatedly";
+    try (var topicAdmin = TopicAdmin.of(bootstrapServers())) {
+      Runnable createTopic =
+          () ->
+              topicAdmin
+                  .creator()
+                  .configs(Map.of(TopicConfig.COMPRESSION_TYPE_CONFIG, "lz4"))
+                  .numberOfReplicas((short) 1)
+                  .numberOfPartitions(3)
+                  .topic(topicName)
+                  .create();
+
+      createTopic.run();
+      Utils.waitFor(() -> topicAdmin.topics().containsKey(topicName));
+      IntStream.range(0, 10).forEach(i -> createTopic.run());
+
+      // changing number of partitions can producer error
+      Assertions.assertThrows(
+          IllegalArgumentException.class,
+          () -> topicAdmin.creator().numberOfPartitions(1).topic(topicName).create());
+
+      // changing number of replicas can producer error
+      Assertions.assertThrows(
+          IllegalArgumentException.class,
+          () -> topicAdmin.creator().numberOfReplicas((short) 2).topic(topicName).create());
+
+      // changing config can producer error
+      Assertions.assertThrows(
+          IllegalArgumentException.class,
+          () ->
+              topicAdmin
+                  .creator()
+                  .configs(Map.of(TopicConfig.COMPRESSION_TYPE_CONFIG, "gzip"))
+                  .topic(topicName)
+                  .create());
+    }
+  }
+
   @Test
   void testPartitions() throws IOException, InterruptedException {
     var topicName = "testPartitions";