MINOR: add helper function createTopic to ClusterInstance (apache#1…

…6852) Reviewers: Chia-Ping Tsai <[email protected]>
TongchengOpenSource · Aug 31, 2024 · 8f4d856 · 8f4d856
1 parent 1841c07
commit 8f4d856
Show file tree

Hide file tree

Showing 2 changed files with 27 additions and 0 deletions.
diff --git a/core/src/test/java/kafka/test/ClusterInstance.java b/core/src/test/java/kafka/test/ClusterInstance.java
@@ -25,6 +25,7 @@
 import kafka.test.annotation.Type;
 
 import org.apache.kafka.clients.admin.Admin;
+import org.apache.kafka.clients.admin.NewTopic;
 import org.apache.kafka.clients.consumer.GroupProtocol;
 import org.apache.kafka.common.network.ListenerName;
 import org.apache.kafka.test.TestUtils;
@@ -186,6 +187,13 @@ default Set<GroupProtocol> supportedGroupProtocols() {
     default void waitTopicDeletion(String topic) throws InterruptedException {
         waitForTopic(topic, 0);
     }
+
+    default void createTopic(String topicName, int partitions, short replicas) throws InterruptedException {
+        try (Admin admin = createAdminClient()) {
+            admin.createTopics(Collections.singletonList(new NewTopic(topicName, partitions, replicas)));
+            waitForTopic(topicName, partitions);
+        }
+    }
 
     void waitForReadyBrokers() throws InterruptedException;
 

diff --git a/core/src/test/java/kafka/test/ClusterTestExtensionsTest.java b/core/src/test/java/kafka/test/ClusterTestExtensionsTest.java
@@ -32,6 +32,7 @@
 import org.apache.kafka.clients.admin.DescribeLogDirsResult;
 import org.apache.kafka.clients.admin.NewTopic;
 import org.apache.kafka.clients.consumer.GroupProtocol;
+import org.apache.kafka.common.TopicPartitionInfo;
 import org.apache.kafka.common.config.ConfigResource;
 import org.apache.kafka.server.common.MetadataVersion;
 
@@ -232,6 +233,24 @@ public void testNotSupportedNewGroupProtocols(ClusterInstance clusterInstance) {
         Assertions.assertEquals(1, clusterInstance.supportedGroupProtocols().size());
     }
 
+
+
+    @ClusterTest(types = {Type.ZK, Type.CO_KRAFT, Type.KRAFT}, brokers = 3)
+    public void testCreateTopic(ClusterInstance clusterInstance) throws Exception {
+        String topicName = "test";
+        int numPartition = 3;
+        short numReplicas = 3;
+        clusterInstance.createTopic(topicName, numPartition, numReplicas);
+
+        try (Admin admin = clusterInstance.createAdminClient()) {
+            Assertions.assertTrue(admin.listTopics().listings().get().stream().anyMatch(s -> s.name().equals(topicName)));
+            List<TopicPartitionInfo> partitions = admin.describeTopics(Collections.singleton(topicName)).allTopicNames().get()
+                    .get(topicName).partitions();
+            Assertions.assertEquals(numPartition, partitions.size());
+            Assertions.assertTrue(partitions.stream().allMatch(partition -> partition.replicas().size() == numReplicas));
+        }
+    }
+
     @ClusterTest(types = {Type.ZK, Type.CO_KRAFT, Type.KRAFT}, brokers = 4)
     public void testClusterAliveBrokers(ClusterInstance clusterInstance) throws Exception {
         clusterInstance.waitForReadyBrokers();