Use ClusterInstance#createTopic(WithAssignment) in remote log metadata manager tests

Author: lytt925

storage/src/test/java/org/apache/kafka/server/log/remote/metadata/storage/TopicBasedRemoteLogMetadataManagerMultipleSubscriptionsTest.java

@@ -17,9 +17,6 @@
 package org.apache.kafka.server.log.remote.metadata.storage;
 
 
-import org.apache.kafka.clients.CommonClientConfigs;
-import org.apache.kafka.clients.admin.Admin;
-import org.apache.kafka.clients.admin.NewTopic;
 import org.apache.kafka.common.TopicIdPartition;
 import org.apache.kafka.common.TopicPartition;
 import org.apache.kafka.common.Uuid;
@@ -171,9 +168,6 @@ public int metadataPartition(TopicIdPartition topicIdPartition) {
     }
 
     private void createTopic(String topic, Map<Integer, List<Integer>> replicasAssignments) {
-        try (Admin admin = Admin.create(Map.of(CommonClientConfigs.BOOTSTRAP_SERVERS_CONFIG, clusterInstance.bootstrapServers()))) {
-            admin.createTopics(List.of(new NewTopic(topic, replicasAssignments)));
-            assertDoesNotThrow(() -> clusterInstance.waitTopicCreation(topic, replicasAssignments.size()));
-        }
+        assertDoesNotThrow(() -> clusterInstance.createTopicWithAssignment(topic, replicasAssignments));
     }
 }

storage/src/test/java/org/apache/kafka/server/log/remote/metadata/storage/TopicBasedRemoteLogMetadataManagerRestartTest.java

@@ -16,8 +16,6 @@
  */
 package org.apache.kafka.server.log.remote.metadata.storage;
 
-import org.apache.kafka.clients.admin.Admin;
-import org.apache.kafka.clients.admin.NewTopic;
 import org.apache.kafka.common.TopicIdPartition;
 import org.apache.kafka.common.TopicPartition;
 import org.apache.kafka.common.Uuid;
@@ -58,15 +56,10 @@ public void testRLMMAPIsAfterRestart() throws Exception {
         // Create topics.
         String leaderTopic = "new-leader";
         String followerTopic = "new-follower";
-        try (Admin admin = clusterInstance.admin()) {
-            // Set broker id 0 as the first entry which is taken as the leader.
-            NewTopic newLeaderTopic = new NewTopic(leaderTopic, Map.of(0, List.of(0, 1, 2)));
-            // Set broker id 1 as the first entry which is taken as the leader.
-            NewTopic newFollowerTopic = new NewTopic(followerTopic, Map.of(0, List.of(1, 2, 0)));
-            admin.createTopics(List.of(newLeaderTopic, newFollowerTopic)).all().get();
-        }
-        clusterInstance.waitTopicCreation(leaderTopic, 1);
-        clusterInstance.waitTopicCreation(followerTopic, 1);
+        // Set broker id 0 as the first entry which is taken as the leader.
+        clusterInstance.createTopicWithAssignment(leaderTopic, Map.of(0, List.of(0, 1, 2)));
+        // Set broker id 1 as the first entry which is taken as the leader.
+        clusterInstance.createTopicWithAssignment(followerTopic, Map.of(0, List.of(1, 2, 0)));
 
         TopicIdPartition leaderTopicIdPartition = new TopicIdPartition(Uuid.randomUuid(), new TopicPartition(leaderTopic, 0));
         TopicIdPartition followerTopicIdPartition = new TopicIdPartition(Uuid.randomUuid(), new TopicPartition(followerTopic, 0));

storage/src/test/java/org/apache/kafka/server/log/remote/metadata/storage/TopicBasedRemoteLogMetadataManagerTest.java

@@ -21,7 +21,6 @@
 import org.apache.kafka.clients.admin.ConfigEntry;
 import org.apache.kafka.clients.admin.DescribeConfigsResult;
 import org.apache.kafka.clients.admin.DescribeTopicsResult;
-import org.apache.kafka.clients.admin.NewTopic;
 import org.apache.kafka.clients.admin.TopicDescription;
 import org.apache.kafka.common.KafkaFuture;
 import org.apache.kafka.common.TopicIdPartition;
@@ -94,10 +93,9 @@ public void teardown() throws IOException {
 
     @ClusterTest
     public void testDoesTopicExist() throws ExecutionException, InterruptedException {
+        String topic = "test-topic-exist";
+        clusterInstance.createTopic(topic, 1, (short) 1);
         try (Admin admin = clusterInstance.admin()) {
-            String topic = "test-topic-exist";
-            admin.createTopics(List.of(new NewTopic(topic, 1, (short) 1))).all().get();
-            clusterInstance.waitTopicCreation(topic, 1);
             boolean doesTopicExist = topicBasedRlmm().doesTopicExist(admin, topic);
             assertTrue(doesTopicExist);
         }
@@ -144,13 +142,9 @@ public void testNewPartitionUpdates() throws Exception {
         // Create topics.
         String leaderTopic = "new-leader";
         String followerTopic = "new-follower";
-        try (Admin admin = clusterInstance.admin()) {
-            // Set broker id 0 as the first entry which is taken as the leader.
-            admin.createTopics(List.of(new NewTopic(leaderTopic, Map.of(0, List.of(0, 1, 2))))).all().get();
-            clusterInstance.waitTopicCreation(leaderTopic, 1);
-            admin.createTopics(List.of(new NewTopic(followerTopic, Map.of(0, List.of(1, 2, 0))))).all().get();
-            clusterInstance.waitTopicCreation(followerTopic, 1);
-        }
+        // Set broker id 0 as the first entry which is taken as the leader.
+        clusterInstance.createTopicWithAssignment(leaderTopic, Map.of(0, List.of(0, 1, 2)));
+        clusterInstance.createTopicWithAssignment(followerTopic, Map.of(0, List.of(1, 2, 0)));
 
         final TopicIdPartition newLeaderTopicIdPartition = new TopicIdPartition(Uuid.randomUuid(), new TopicPartition(leaderTopic, 0));
         final TopicIdPartition newFollowerTopicIdPartition = new TopicIdPartition(Uuid.randomUuid(), new TopicPartition(followerTopic, 0));