Skip to content

MINOR: Propagate ClusterInstance#deleteTopic errors and use it in more tests #23715

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
lytt925 wants to merge 2 commits into
base: trunk
Choose a base branch
from
Open

MINOR: Propagate ClusterInstance#deleteTopic errors and use it in more tests #23715

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
e91e66c
Propagate ClusterInstance#deleteTopic errors and use it in more tests
lytt925 Oct 5, 2026
d264b28
MINOR: Merge trunk and resolve topic deletion test conflicts
lytt925 Oct 15, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
67 changes: 26 additions & 41 deletions ...ients-integration-tests/src/test/java/org/apache/kafka/clients/admin/DeleteTopicTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -67,26 +67,17 @@ public class DeleteTopicTest {
@ClusterTest
public void testCreateAndDeleteTopic(ClusterInstance cluster) throws Exception {
String testTopic = "test-topic";
try (Admin admin = cluster.admin()) {
// Create a test topic
cluster.createTopic(testTopic, 1, (short) 3);

// Delete topic
DeleteTopicsResult deleteResult = admin.deleteTopics(List.of(testTopic));
deleteResult.all().get();
// Create a test topic
cluster.createTopic(testTopic, 1, (short) 3);

// Wait for topic deletion
cluster.waitTopicDeletion(testTopic);
}
// Delete topic and wait for topic deletion
cluster.deleteTopic(testTopic);
}

@ClusterTest
public void testDeleteTopicWithAllAliveReplicas(ClusterInstance cluster) throws Exception {
try (Admin admin = cluster.admin()) {
cluster.createTopicWithAssignment(DEFAULT_TOPIC, expectedReplicaAssignment);
admin.deleteTopics(List.of(DEFAULT_TOPIC)).all().get();
cluster.waitTopicDeletion(DEFAULT_TOPIC);
}
cluster.createTopicWithAssignment(DEFAULT_TOPIC, expectedReplicaAssignment);
cluster.deleteTopic(DEFAULT_TOPIC);
}

@ClusterTest
Expand Down Expand Up @@ -207,15 +198,12 @@ public void testAddPartitionDuringDeleteTopic(ClusterInstance cluster) throws Ex

@ClusterTest
public void testRecreateTopicAfterDeletion(ClusterInstance cluster) throws Exception {
try (Admin admin = cluster.admin()) {
cluster.createTopicWithAssignment(DEFAULT_TOPIC, expectedReplicaAssignment);
TopicPartition topicPartition = new TopicPartition(DEFAULT_TOPIC, 0);
admin.deleteTopics(List.of(DEFAULT_TOPIC)).all().get();
cluster.waitTopicDeletion(DEFAULT_TOPIC);
// re-create topic on same replicas
cluster.createTopicWithAssignment(DEFAULT_TOPIC, expectedReplicaAssignment);
waitForReplicaCreated(cluster.brokers(), topicPartition, "Replicas for topic " + DEFAULT_TOPIC + " not created.");
}
cluster.createTopicWithAssignment(DEFAULT_TOPIC, expectedReplicaAssignment);
TopicPartition topicPartition = new TopicPartition(DEFAULT_TOPIC, 0);
cluster.deleteTopic(DEFAULT_TOPIC);
// re-create topic on same replicas
cluster.createTopicWithAssignment(DEFAULT_TOPIC, expectedReplicaAssignment);
waitForReplicaCreated(cluster.brokers(), topicPartition, "Replicas for topic " + DEFAULT_TOPIC + " not created.");
}

@ClusterTest
Expand Down Expand Up @@ -246,23 +234,20 @@ public void testDeleteNonExistingTopic(ClusterInstance cluster) throws Exception
@ClusterConfigProperty(key = "log.cleaner.dedupe.buffer.size", value = "1048577")
})
public void testDeleteTopicWithCleaner(ClusterInstance cluster) throws Exception {
try (Admin admin = cluster.admin()) {
cluster.createTopicWithAssignment(DEFAULT_TOPIC, expectedReplicaAssignment);
TopicPartition topicPartition = new TopicPartition(DEFAULT_TOPIC, 0);
// for simplicity, we are validating cleaner offsets on a single broker
KafkaBroker server = cluster.brokers().values().stream().findFirst().orElseThrow();
TestUtils.waitForCondition(() -> server.logManager().getLog(topicPartition, false).isPresent(),
"Replicas for topic test not created.");
UnifiedLog log = server.logManager().getLog(topicPartition, false).get();
writeDups(100, 3, log);
// force roll the segment so that cleaner can work on it
server.logManager().getLog(topicPartition, false).get().roll(Optional.empty());
// wait for cleaner to clean
server.logManager().cleaner().awaitCleaned(topicPartition, 0, 60000);
admin.deleteTopics(List.of(DEFAULT_TOPIC)).all().get();

cluster.waitTopicDeletion(DEFAULT_TOPIC);
}
cluster.createTopicWithAssignment(DEFAULT_TOPIC, expectedReplicaAssignment);
TopicPartition topicPartition = new TopicPartition(DEFAULT_TOPIC, 0);
// for simplicity, we are validating cleaner offsets on a single broker
KafkaBroker server = cluster.brokers().values().stream().findFirst().orElseThrow();
TestUtils.waitForCondition(() -> server.logManager().getLog(topicPartition, false).isPresent(),
"Replicas for topic test not created.");
UnifiedLog log = server.logManager().getLog(topicPartition, false).get();
writeDups(100, 3, log);
// force roll the segment so that cleaner can work on it
server.logManager().getLog(topicPartition, false).get().roll(Optional.empty());
// wait for cleaner to clean
server.logManager().cleaner().awaitCleaned(topicPartition, 0, 60000);

cluster.deleteTopic(DEFAULT_TOPIC);
}

@ClusterTest
Expand Down
11 changes: 8 additions & 3 deletions ...ommon/test-common-runtime/src/main/java/org/apache/kafka/common/test/ClusterInstance.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -356,12 +356,17 @@ default void createTopicWithAssignment(String topicName, Map<Integer, List<Integ
* Deletes a topic and waits for the deletion to complete.
*
* @param topicName The name of the topic to delete
* @throws InterruptedException If the operation is interrupted
*/
default void deleteTopic(String topicName) throws InterruptedException, ExecutionException {
default void deleteTopic(String topicName) {
try (Admin admin = admin()) {
admin.deleteTopics(List.of(topicName));
admin.deleteTopics(List.of(topicName)).all().get();
waitTopicDeletion(topicName);
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
throw new RuntimeException("Failed to delete topic " + topicName, e);
} catch (ExecutionException e) {
if (e.getCause() instanceof RuntimeException re) throw re;
throw new RuntimeException("Failed to delete topic " + topicName, e.getCause());
}
}

Expand Down
8 changes: 5 additions & 3 deletions ...n-runtime/src/test/java/org/apache/kafka/common/test/junit/ClusterTestExtensionsTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -41,6 +41,7 @@
import org.apache.kafka.common.errors.ClusterAuthorizationException;
import org.apache.kafka.common.errors.SaslAuthenticationException;
import org.apache.kafka.common.errors.TopicAuthorizationException;
import org.apache.kafka.common.errors.UnknownTopicOrPartitionException;
import org.apache.kafka.common.security.auth.SecurityProtocol;
import org.apache.kafka.common.serialization.StringDeserializer;
import org.apache.kafka.common.serialization.StringSerializer;
Expand Down Expand Up @@ -292,15 +293,16 @@ public void testClusterAliveBrokers(ClusterInstance clusterInstance) throws Exce


@ClusterTest(types = {Type.CO_KRAFT, Type.KRAFT}, brokers = 4)
public void testVerifyTopicDeletion(ClusterInstance clusterInstance) throws Exception {
public void testDeleteTopic(ClusterInstance clusterInstance) throws Exception {
String testTopic = "testTopic";
clusterInstance.createTopic(testTopic, 1, (short) 1);
try (Admin admin = clusterInstance.admin()) {
admin.deleteTopics(List.of(testTopic));
clusterInstance.waitTopicDeletion(testTopic);
clusterInstance.deleteTopic(testTopic);
Assertions.assertTrue(admin.listTopics().listings().get().stream().noneMatch(
topic -> topic.name().equals(testTopic)
));

assertThrows(UnknownTopicOrPartitionException.class, () -> clusterInstance.deleteTopic("non-existent-topic"));
}
}

Expand Down
6 changes: 2 additions & 4 deletions tools/src/test/java/org/apache/kafka/tools/TopicCommandTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -380,8 +380,7 @@ public void testCreate(ClusterInstance clusterInstance) throws InterruptedExcept
Assertions.assertTrue(adminClient.listTopics().names().get().contains(testTopicName),
"Admin client didn't see the created topic. It saw: " + adminClient.listTopics().names().get());

adminClient.deleteTopics(List.of(testTopicName));
clusterInstance.waitTopicDeletion(testTopicName);
clusterInstance.deleteTopic(testTopicName);
Assertions.assertTrue(adminClient.listTopics().names().get().isEmpty(),
"Admin client see the created topic. It saw: " + adminClient.listTopics().names().get());
}
Expand All @@ -405,8 +404,7 @@ public void testCreateWithDefaults(ClusterInstance clusterInstance) throws Inter
Assertions.assertEquals(defaultNumPartitions, partitions.size(), "Unequal partition size: " + partitions.size());
Assertions.assertEquals(defaultReplicationFactor, (short) partitions.get(0).replicas().size(), "Unequal replication factor: " + partitions.get(0).replicas().size());

adminClient.deleteTopics(List.of(testTopicName));
clusterInstance.waitTopicDeletion(testTopicName);
clusterInstance.deleteTopic(testTopicName);
Assertions.assertTrue(adminClient.listTopics().names().get().isEmpty(),
"Admin client see the created topic. It saw: " + adminClient.listTopics().names().get());
}
Expand Down
8 changes: 1 addition & 7 deletions tools/src/test/java/org/apache/kafka/tools/consumer/group/DescribeConsumerGroupTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -439,7 +439,7 @@ public void testDescribeAllExistingGroups(ClusterInstance clusterInstance) throw
}
// remove previous consumer groups, so we can have a clean cluster for next consumer group protocol test.
deleteConsumerGroups(groups);
deleteTopic(topic);
clusterInstance.deleteTopic(topic);
}
}
}
Expand Down Expand Up @@ -1182,12 +1182,6 @@ private void deleteConsumerGroups(Collection<String> groupIds) {
}
}

private void deleteTopic(String topic) {
try (Admin admin = Admin.create(Map.of(CommonClientConfigs.BOOTSTRAP_SERVERS_CONFIG, clusterInstance.bootstrapServers()))) {
Assertions.assertDoesNotThrow(() -> admin.deleteTopics(List.of(topic)).topicNameValues().get(topic).get());
}
}

private AutoCloseable consumerGroupClosable(GroupProtocol protocol, String groupId, Set<TopicPartition> topicPartitions, Map<String, Object> customConfigs) {
Map<String, Object> configs = composeConfigs(
groupId,
Expand Down
Loading