MINOR: Propagate ClusterInstance#createTopic errors and clean up callers

clients/clients-integration-tests/src/test/java/org/apache/kafka/clients/admin/DescribeProducersWithBrokerIdTest.java

@@ -57,7 +57,7 @@ private static void sendTestRecords(Producer<byte[], byte[]> producer) {
     }
 
     @BeforeEach
-    void setUp() throws InterruptedException {
+    void setUp() {
         clusterInstance.createTopic(TOPIC_NAME, NUM_PARTITIONS, REPLICATION_FACTOR);
     }
 

clients/clients-integration-tests/src/test/java/org/apache/kafka/clients/admin/ListOffsetsIntegrationTest.java

@@ -130,14 +130,14 @@ public void testThreeNonCompressedRecordsInSeparateBatch() throws ExecutionExcep
 
     @ClusterTest
     public void testThreeRecordsInOneBatchHavingDifferentCompressionTypeWithServer() throws InterruptedException, ExecutionException {
-        createTopicWithConfig(CUSTOM_CONFIG_TOPIC, Map.of(TopicConfig.COMPRESSION_TYPE_CONFIG, CompressionType.LZ4.name));
+        clusterInstance.createTopic(CUSTOM_CONFIG_TOPIC, PARTITION, REPLICAS, Map.of(TopicConfig.COMPRESSION_TYPE_CONFIG, CompressionType.LZ4.name));
         produceMessagesInOneBatch(CompressionType.NONE.name, CUSTOM_CONFIG_TOPIC);
         verifyListOffsets(CUSTOM_CONFIG_TOPIC, 1);
     }
 
     @ClusterTest
     public void testThreeRecordsInSeparateBatchHavingDifferentCompressionTypeWithServer() throws InterruptedException, ExecutionException {
-        createTopicWithConfig(CUSTOM_CONFIG_TOPIC, Map.of(TopicConfig.COMPRESSION_TYPE_CONFIG, CompressionType.LZ4.name));
+        clusterInstance.createTopic(CUSTOM_CONFIG_TOPIC, PARTITION, REPLICAS, Map.of(TopicConfig.COMPRESSION_TYPE_CONFIG, CompressionType.LZ4.name));
         produceMessagesInOneBatch(CompressionType.NONE.name, CUSTOM_CONFIG_TOPIC);
         verifyListOffsets(CUSTOM_CONFIG_TOPIC, 1);
     }
@@ -287,11 +287,7 @@ private ListOffsetsResultInfo runFetchOffsets(OffsetSpec offsetSpec,
         return adminClient.listOffsets(Map.of(tp, offsetSpec), new ListOffsetsOptions()).all().get().get(tp);
     }
 
-    private void setUpForLogAppendTimeCase() throws InterruptedException {
-        createTopicWithConfig(CUSTOM_CONFIG_TOPIC, Map.of(TopicConfig.MESSAGE_TIMESTAMP_TYPE_CONFIG, "LogAppendTime"));
-    }
-
-    private void createTopicWithConfig(String topic, Map<String, String> props) throws InterruptedException {
-        clusterInstance.createTopic(topic, PARTITION, REPLICAS, props);
+    private void setUpForLogAppendTimeCase() {
+        clusterInstance.createTopic(CUSTOM_CONFIG_TOPIC, PARTITION, REPLICAS, Map.of(TopicConfig.MESSAGE_TIMESTAMP_TYPE_CONFIG, "LogAppendTime"));
     }
 }

clients/clients-integration-tests/src/test/java/org/apache/kafka/clients/consumer/ConsumerBounceTest.java

@@ -127,7 +127,7 @@ public class ConsumerBounceTest {
     }
 
     @BeforeEach
-    void setUp() throws InterruptedException {
+    void setUp() {
         clusterInstance.createTopic(topic, numPartitions, numReplica);
     }
 
@@ -272,7 +272,7 @@ private void testSubscribeWhenTopicUnavailable(GroupProtocol groupProtocol) thro
         consumer.subscribe(List.of(newTopic));
         consumer.poll(Duration.ZERO);
         // Schedule topic creation after 2 seconds
-        executor.schedule(() -> assertDoesNotThrow(() -> clusterInstance.createTopic(newTopic, numPartitions, numReplica)),
+        executor.schedule(() -> clusterInstance.createTopic(newTopic, numPartitions, numReplica),
                 2, TimeUnit.SECONDS);
 
         // Start first poller

clients/clients-integration-tests/src/test/java/org/apache/kafka/clients/consumer/ConsumerWithLegacyMessageFormatIntegrationTest.java

@@ -107,7 +107,7 @@ private void appendLegacyRecords(int numRecords, TopicPartition tp, int brokerId
     }
 
     @BeforeEach
-    public void setupTopics() throws InterruptedException {
+    public void setupTopics() {
         cluster.createTopic(topic1, 2, (short) 1);
         cluster.createTopicWithAssignment(topic2, Map.of(0, List.of(0), 1, List.of(1)));
         cluster.createTopicWithAssignment(topic3, Map.of(0, List.of(0), 1, List.of(1)));

clients/clients-integration-tests/src/test/java/org/apache/kafka/clients/consumer/PlaintextConsumerAssignTest.java

@@ -72,7 +72,7 @@ public class PlaintextConsumerAssignTest {
     }
 
     @BeforeEach
-    public void setup() throws InterruptedException {
+    public void setup() {
         clusterInstance.createTopic(topic, 2, (short) BROKER_COUNT);
     }
 

clients/clients-integration-tests/src/test/java/org/apache/kafka/clients/consumer/PlaintextConsumerCommitTest.java

@@ -83,7 +83,7 @@ public PlaintextConsumerCommitTest(ClusterInstance clusterInstance) {
     }
 
     @BeforeEach
-    public void setup() throws InterruptedException {
+    public void setup() {
         cluster.createTopic(topic, 2, (short) BROKER_COUNT);
     }
 
@@ -580,7 +580,7 @@ public void testCommitAsyncFailsWhenCoordinatorUnavailableDuringClose() throws I
 
     // TODO: This only works in the new consumer, but should be fixed for the old consumer as well
     @ClusterTest
-    public void testCommitAsyncCompletedBeforeConsumerCloses() throws InterruptedException {
+    public void testCommitAsyncCompletedBeforeConsumerCloses() {
         // This is testing the contract that asynchronous offset commit are completed before the consumer
         // is closed, even when no commit sync is performed as part of the close (due to auto-commit
         // disabled, or simply because there are no consumed offsets).

clients/clients-integration-tests/src/test/java/org/apache/kafka/clients/consumer/PlaintextConsumerFetchTest.java

@@ -79,7 +79,7 @@ public PlaintextConsumerFetchTest(ClusterInstance cluster) {
     }
 
     @BeforeEach
-    public void setup() throws InterruptedException {
+    public void setup() {
         cluster.createTopic(topic, 2, (short) BROKER_COUNT);
     }
 

clients/clients-integration-tests/src/test/java/org/apache/kafka/clients/consumer/PlaintextConsumerPollTest.java

@@ -95,7 +95,7 @@ public PlaintextConsumerPollTest(ClusterInstance cluster) {
     }
 
     @BeforeEach
-    public void setup() throws InterruptedException {
+    public void setup() {
         cluster.createTopic(topic, 2, (short) BROKER_COUNT);
     }
 

clients/clients-integration-tests/src/test/java/org/apache/kafka/clients/consumer/PlaintextConsumerSubscriptionTest.java

@@ -81,7 +81,7 @@ public PlaintextConsumerSubscriptionTest(ClusterInstance cluster) {
     }
 
     @BeforeEach
-    public void setup() throws InterruptedException {
+    public void setup() {
         cluster.createTopic(topic, 2, (short) BROKER_COUNT);
     }
 

clients/clients-integration-tests/src/test/java/org/apache/kafka/clients/consumer/PlaintextConsumerTest.java

@@ -560,7 +560,6 @@ private void testSeek(Map<String, Object> consumerConfig) throws Exception {
 
             // Test seek compressed message
             var tp2 = new TopicPartition(TOPIC, 1);
-            cluster.createTopic(TOPIC, 2, (short) BROKER_COUNT);
             sendCompressedMessages(totalRecords, tp2);
             consumer.assign(List.of(tp2));
 

clients/clients-integration-tests/src/test/java/org/apache/kafka/clients/consumer/SaslPlainPlaintextConsumerTest.java

@@ -70,7 +70,7 @@ public SaslPlainPlaintextConsumerTest(ClusterInstance cluster) {
     }
 
     @BeforeEach
-    public void setUp() throws InterruptedException {
+    public void setUp() {
         cluster.createTopic(ClientsTestUtils.BaseConsumerTestcase.TOPIC, 2, (short) ClientsTestUtils.BaseConsumerTestcase.BROKER_COUNT);
     }
 

clients/clients-integration-tests/src/test/java/org/apache/kafka/clients/consumer/SaslPlaintextConsumerTest.java

@@ -115,7 +115,7 @@ public static void setup() throws Exception {
     }
 
     @BeforeEach
-    public void beforeEach() throws InterruptedException {
+    public void beforeEach() {
         cluster.createTopic(ClientsTestUtils.BaseConsumerTestcase.TOPIC, 2, (short) ClientsTestUtils.BaseConsumerTestcase.BROKER_COUNT);
     }
 

clients/clients-integration-tests/src/test/java/org/apache/kafka/clients/consumer/ShareConsumerDLQTest.java

@@ -52,7 +52,6 @@
 import static org.apache.kafka.test.TestUtils.DEFAULT_MAX_WAIT_MS;
 import static org.apache.kafka.test.TestUtils.waitForCondition;
 import static org.junit.jupiter.api.Assertions.assertArrayEquals;
-import static org.junit.jupiter.api.Assertions.assertDoesNotThrow;
 import static org.junit.jupiter.api.Assertions.assertEquals;
 import static org.junit.jupiter.api.Assertions.assertNotNull;
 import static org.junit.jupiter.api.Assertions.assertNull;
@@ -956,8 +955,8 @@ private void createDlqTopic(String topicName) {
     }
 
     private void createDlqTopic(String topicName, int numPartitions) {
-        assertDoesNotThrow(() -> cluster.createTopic(topicName, numPartitions, (short) 1,
-            Map.of(TopicConfig.ERRORS_DEADLETTERQUEUE_GROUP_ENABLE_CONFIG, "true")), "Failed to create DLQ topic");
+        cluster.createTopic(topicName, numPartitions, (short) 1,
+            Map.of(TopicConfig.ERRORS_DEADLETTERQUEUE_GROUP_ENABLE_CONFIG, "true"));
     }
 
     // Creates a single-partition source topic with tiered storage enabled and one log segment per record (via
@@ -973,8 +972,7 @@ private void createRemoteStorageSourceTopic(String topic, long retentionMs, long
             // Roll a segment for every record so each inactive segment can be offloaded then deleted locally.
             TopicConfig.INDEX_INTERVAL_BYTES_CONFIG, "1",
             TopicConfig.SEGMENT_INDEX_BYTES_CONFIG, "12");
-        assertDoesNotThrow(() -> cluster.createTopic(topic, 1, (short) 1, configs),
-            "Failed to create remote-storage source topic");
+        cluster.createTopic(topic, 1, (short) 1, configs);
     }
 
     // The earliest offset still held in local storage. Offsets below this have been removed locally (e.g. after

clients/clients-integration-tests/src/test/java/org/apache/kafka/clients/producer/ProducerFailureHandlingTest.java

@@ -107,7 +107,7 @@ public void testTooLargeRecordWithAckZero(ClusterInstance clusterInstance) throw
      * With ack == 1 the future metadata will throw ExecutionException caused by RecordTooLargeException
      */
     @ClusterTest
-    public void testTooLargeRecordWithAckOne(ClusterInstance clusterInstance) throws InterruptedException {
+    public void testTooLargeRecordWithAckOne(ClusterInstance clusterInstance) {
         clusterInstance.createTopic(topic1, 1, (short) clusterInstance.brokers().size());
 
         try (Producer<byte[], byte[]> producer = clusterInstance.producer(producerConfig(1))) {
@@ -156,7 +156,7 @@ public void testNonExistentTopic(ClusterInstance clusterInstance) {
      * With incorrect broker-list the future metadata should return ExecutionException caused by TimeoutException
      */
     @ClusterTest
-    public void testWrongBrokerList(ClusterInstance clusterInstance) throws InterruptedException {
+    public void testWrongBrokerList(ClusterInstance clusterInstance) {
         clusterInstance.createTopic(topic1, 1, (short) 1);
         // producer with incorrect broker list
         Map<String, Object> producerConfig = new HashMap<>(producerConfig(1));
@@ -174,7 +174,7 @@ public void testWrongBrokerList(ClusterInstance clusterInstance) throws Interrup
      * when partition is higher than the upper bound of partitions.
      */
     @ClusterTest
-    public void testInvalidPartition(ClusterInstance clusterInstance) throws InterruptedException {
+    public void testInvalidPartition(ClusterInstance clusterInstance) {
         // create topic with a single partition
         clusterInstance.createTopic(topic1, 1, (short) clusterInstance.brokers().size());
 

clients/clients-integration-tests/src/test/java/org/apache/kafka/clients/producer/ProducerIntegrationTest.java

@@ -120,7 +120,7 @@ public void testTransactionWithAndWithoutSend(ClusterInstance cluster) {
         @ClusterTest(features = {
             @ClusterFeature(feature = Feature.TRANSACTION_VERSION, version = 2)}),
     })
-    public void testTransactionWithInvalidSendAndEndTxnRequestSent(ClusterInstance cluster) throws InterruptedException {
+    public void testTransactionWithInvalidSendAndEndTxnRequestSent(ClusterInstance cluster) {
         var topic = "foobar";
         cluster.createTopic(topic, 1, (short) 1, Map.of(TopicConfig.MAX_MESSAGE_BYTES_CONFIG, "100"));
         String txnId = "test-txn";

server/src/test/java/org/apache/kafka/api/EndToEndClusterIdTest.java

@@ -83,7 +83,7 @@ public class EndToEndClusterIdTest {
     }
 
     @BeforeEach
-    public void setup() throws InterruptedException {
+    public void setup() {
         this.clusterInstance.createTopic(TOPIC, 2, (short) 1);
         clusterBrokerId = String.valueOf(clusterInstance.brokerIds().iterator().next());
         controllerId = String.valueOf(clusterInstance.controllerIds().iterator().next());

server/src/test/java/org/apache/kafka/server/CordonedLogDirsIntegrationTest.java

@@ -494,11 +494,11 @@ private void setCordonedLogDirs(Admin admin, List<String> logDirs, ConfigResourc
         }, 10_000, "Unable to set the " + CORDONED_LOG_DIRS_CONFIG + " configuration on " + cr + ".");
     }
 
-    private void createTopic(String name) throws InterruptedException {
+    private void createTopic(String name) {
         createTopic(name, (short) clusterInstance.brokers().size());
     }
 
-    private void createTopic(String name, short replicationFactor) throws InterruptedException {
+    private void createTopic(String name, short replicationFactor) {
         clusterInstance.createTopic(name, 1, replicationFactor);
     }
 

server/src/test/java/org/apache/kafka/server/MetadataRequestTest.java

@@ -48,7 +48,6 @@
 import java.util.Map;
 import java.util.Optional;
 import java.util.Set;
-import java.util.concurrent.ExecutionException;
 import java.util.stream.Collectors;
 
 import static org.junit.jupiter.api.Assertions.assertEquals;
@@ -156,7 +155,7 @@ public void testRack() throws IOException {
     }
 
     @ClusterTest
-    public void testIsInternal() throws InterruptedException, IOException {
+    public void testIsInternal() throws IOException {
         String internalTopic = Topic.GROUP_METADATA_TOPIC_NAME;
         String notInternalTopic = "notInternal";
         // create the topics
@@ -177,7 +176,7 @@ public void testIsInternal() throws InterruptedException, IOException {
     }
 
     @ClusterTest
-    public void testNoTopicsRequest() throws InterruptedException, IOException {
+    public void testNoTopicsRequest() throws IOException {
         // create some topics
         clusterInstance.createTopic("t1", 3, (short) 2);
         clusterInstance.createTopic("t2", 3, (short) 2);
@@ -233,7 +232,7 @@ public void testAutoCreateTopicWithInvalidReplicationFactor() throws IOException
     }
 
     @ClusterTest
-    public void testAllTopicsRequest() throws InterruptedException, IOException {
+    public void testAllTopicsRequest() throws IOException {
         // create some topics
         clusterInstance.createTopic("t1", 3, (short) 2);
         clusterInstance.createTopic("t2", 3, (short) 2);
@@ -250,7 +249,7 @@ public void testAllTopicsRequest() throws InterruptedException, IOException {
     }
 
     @ClusterTest
-    public void testTopicIdsInResponse() throws ExecutionException, InterruptedException, IOException {
+    public void testTopicIdsInResponse() throws IOException {
         Map<Integer, List<Integer>> replicaAssignment = Map.of(0, List.of(1, 2, 0), 1, List.of(2, 0, 1));
         String topic1 = "topic1";
         String topic2 = "topic2";
@@ -279,7 +278,7 @@ public void testTopicIdsInResponse() throws ExecutionException, InterruptedExcep
      * Preferred replica should be the first item in the replicas list
      */
     @ClusterTest
-    public void testPreferredReplica() throws ExecutionException, InterruptedException, IOException {
+    public void testPreferredReplica() throws IOException {
         Map<Integer, List<Integer>> replicaAssignment = Map.of(0, List.of(1, 2, 0), 1, List.of(2, 0, 1));
         clusterInstance.createTopicWithAssignment("t1", replicaAssignment);
         // Test metadata on two different brokers to ensure that metadata propagation works correctly
@@ -303,7 +302,7 @@ public void testPreferredReplica() throws ExecutionException, InterruptedExcepti
     }
 
     @ClusterTest
-    public void testPartitionInfoPreferredReplica() throws ExecutionException, InterruptedException, IOException {
+    public void testPartitionInfoPreferredReplica() throws IOException {
         Map<Integer, List<Integer>> replicaAssignment = Map.of(0, List.of(1, 2, 0));
         String topic = "testPartitionInfoPreferredReplicaTopic";
         clusterInstance.createTopicWithAssignment(topic, replicaAssignment);
@@ -384,7 +383,7 @@ private void checkIsr(List<KafkaBroker> brokers, String topic) {
     }
 
     @ClusterTest
-    public void testIsrAfterBrokerShutDownAndJoinsBack() throws InterruptedException {
+    public void testIsrAfterBrokerShutDownAndJoinsBack() {
         String topic = "isr-after-broker-shutdown";
         short replicaCount = 3;
         clusterInstance.createTopic(topic, 1, replicaCount);

storage/src/test/java/org/apache/kafka/admin/RemoteTopicCrudTest.java

@@ -55,7 +55,6 @@
 import java.util.Objects;
 import java.util.Optional;
 import java.util.Set;
-import java.util.concurrent.ExecutionException;
 import java.util.concurrent.atomic.AtomicInteger;
 
 import static org.apache.kafka.test.TestUtils.assertFutureThrows;
@@ -95,7 +94,7 @@ void setUp(TestInfo info) {
     }
 
     @ClusterTest
-    void testCreateRemoteTopicWithValidRetentionTime() throws InterruptedException {
+    void testCreateRemoteTopicWithValidRetentionTime() {
         var topicConfig = Map.of(
             TopicConfig.REMOTE_LOG_STORAGE_ENABLE_CONFIG, "true",
             TopicConfig.RETENTION_MS_CONFIG, "60000",
@@ -329,7 +328,7 @@ void testCreateTopicRetentionBytesValidationWithRemoteCopyDisabled() throws Exce
     @ClusterTest
     void testEnableRemoteLogOnExistingTopic() throws Exception {
         try (var admin = cluster.admin()) {
-            cluster.createTopic(testTopicName, numPartitions, numReplicationFactor, Map.of());
+            cluster.createTopic(testTopicName, numPartitions, numReplicationFactor);
 
             var configs = new HashMap<ConfigResource, Collection<AlterConfigOp>>();
             configs.put(new ConfigResource(ConfigResource.Type.TOPIC, testTopicName),
@@ -343,7 +342,7 @@ void testEnableRemoteLogOnExistingTopic() throws Exception {
     @ClusterTest(serverProperties = {
         @ClusterConfigProperty(key = RemoteLogManagerConfig.REMOTE_LOG_STORAGE_SYSTEM_ENABLE_PROP, value = "false")
     })
-    void testEnableRemoteLogWhenSystemRemoteStorageIsDisabled() throws ExecutionException, InterruptedException {
+    void testEnableRemoteLogWhenSystemRemoteStorageIsDisabled() {
         try (var admin = cluster.admin()) {
             var topicConfig = Map.of(
                 TopicConfig.REMOTE_LOG_STORAGE_ENABLE_CONFIG, "true"

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

@@ -35,7 +35,6 @@
 import java.util.concurrent.Phaser;
 import java.util.concurrent.TimeUnit;
 
-import static org.junit.jupiter.api.Assertions.assertDoesNotThrow;
 import static org.junit.jupiter.api.Assertions.assertEquals;
 import static org.junit.jupiter.api.Assertions.assertThrows;
 import static org.junit.jupiter.api.Assertions.assertTrue;
@@ -59,15 +58,15 @@ public void testMultiplePartitionSubscriptions() throws Exception {
         // Create topics.
         String leaderTopic = "leader";
         // Set broker id 0 as the first entry which is taken as the leader.
-        createTopic(leaderTopic, Map.of(0, List.of(0, 1, 2)));
+        clusterInstance.createTopicWithAssignment(leaderTopic, Map.of(0, List.of(0, 1, 2)));
 
         String followerTopic = "follower";
         // Set broker id 1 as the first entry which is taken as the leader.
-        createTopic(followerTopic, Map.of(0, List.of(1, 2, 0)));
+        clusterInstance.createTopicWithAssignment(followerTopic, Map.of(0, List.of(1, 2, 0)));
 
         String topicWithNoMessages = "no-messages-topic";
         // Set broker id 1 as the first entry which is taken as the leader.
-        createTopic(topicWithNoMessages, Map.of(0, List.of(1, 2, 0)));
+        clusterInstance.createTopicWithAssignment(topicWithNoMessages, Map.of(0, List.of(1, 2, 0)));
 
         final TopicIdPartition leaderTopicIdPartition = new TopicIdPartition(Uuid.randomUuid(), new TopicPartition(leaderTopic, 0));
         final TopicIdPartition followerTopicIdPartition = new TopicIdPartition(Uuid.randomUuid(), new TopicPartition(followerTopic, 0));
@@ -166,8 +165,4 @@ public int metadataPartition(TopicIdPartition topicIdPartition) {
             assertTrue(remoteLogMetadataManager.listRemoteLogSegments(followerTopicIdPartition).hasNext(), "No segments found");
         }
     }
-
-    private void createTopic(String topic, Map<Integer, List<Integer>> replicasAssignments) {
-        assertDoesNotThrow(() -> clusterInstance.createTopicWithAssignment(topic, replicasAssignments));
-    }
 }

storage/src/test/java/org/apache/kafka/tiered/storage/TieredStorageTestContext.java

@@ -142,7 +142,7 @@ private static List<BrokerLocalStorage> localStorages(Collection<KafkaBroker> br
                 .toList();
     }
 
-    public void createTopic(TopicSpec spec) throws InterruptedException {
+    public void createTopic(TopicSpec spec) {
         if (spec.assignment() == null || spec.assignment().isEmpty()) {
             cluster.createTopic(spec.topicName(), spec.partitionCount(), (short) spec.replicationFactor(), spec.properties());
         } else {