core/src/test/java/kafka/admin/DeleteTopicTest.java - kafka-merge-queue-sandbox - Git at Google

 /*
  * Licensed to the Apache Software Foundation (ASF) under one or more
  * contributor license agreements. See the NOTICE file distributed with
  * this work for additional information regarding copyright ownership.
  * The ASF licenses this file to You under the Apache License, Version 2.0
  * (the "License"); you may not use this file except in compliance with
  * the License. You may obtain a copy of the License at
  *
  *    http://www.apache.org/licenses/LICENSE-2.0
  *
  * Unless required by applicable law or agreed to in writing, software
  * distributed under the License is distributed on an "AS IS" BASIS,
  * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  * See the License for the specific language governing permissions and
  * limitations under the License.
  */
 package kafka.admin;

 import kafka.log.UnifiedLog;
 import kafka.server.KafkaBroker;

 import org.apache.kafka.clients.admin.Admin;
 import org.apache.kafka.clients.admin.AdminClientConfig;
 import org.apache.kafka.clients.admin.NewPartitionReassignment;
 import org.apache.kafka.clients.admin.NewPartitions;
 import org.apache.kafka.clients.admin.NewTopic;
 import org.apache.kafka.common.TopicPartition;
 import org.apache.kafka.common.compress.Compression;
 import org.apache.kafka.common.errors.TopicDeletionDisabledException;
 import org.apache.kafka.common.errors.UnknownTopicOrPartitionException;
 import org.apache.kafka.common.record.MemoryRecords;
 import org.apache.kafka.common.record.SimpleRecord;
 import org.apache.kafka.common.test.ClusterInstance;
 import org.apache.kafka.common.test.TestUtils;
 import org.apache.kafka.common.test.api.ClusterConfigProperty;
 import org.apache.kafka.common.test.api.ClusterTest;
 import org.apache.kafka.common.test.api.ClusterTestDefaults;
 import org.apache.kafka.common.test.api.Type;
 import org.apache.kafka.coordinator.group.GroupCoordinatorConfig;
 import org.apache.kafka.metadata.BrokerState;
 import org.apache.kafka.server.common.RequestLocal;
 import org.apache.kafka.server.config.ServerConfigs;
 import org.apache.kafka.storage.internals.log.AppendOrigin;
 import org.apache.kafka.storage.internals.log.VerificationGuard;

 import java.util.ArrayList;
 import java.util.Collection;
 import java.util.List;
 import java.util.Map;
 import java.util.Optional;
 import java.util.Properties;
 import java.util.concurrent.ExecutionException;
 import java.util.function.Supplier;
 import java.util.stream.Collectors;

 import scala.Option;
 import scala.jdk.javaapi.OptionConverters;

 import static org.junit.jupiter.api.Assertions.assertDoesNotThrow;
 import static org.junit.jupiter.api.Assertions.assertThrows;
 import static org.junit.jupiter.api.Assertions.fail;


 @ClusterTestDefaults(types = {Type.KRAFT},
     brokers = 3,
     serverProperties = {
         @ClusterConfigProperty(key = GroupCoordinatorConfig.OFFSETS_TOPIC_PARTITIONS_CONFIG, value = "1"),
         @ClusterConfigProperty(key = GroupCoordinatorConfig.OFFSETS_TOPIC_REPLICATION_FACTOR_CONFIG, value = "1"),
         @ClusterConfigProperty(key = "log.initial.task.delay.ms", value = "100"),
         @ClusterConfigProperty(key = "log.segment.delete.delay.ms", value = "1000")
     })
 public class DeleteTopicTest {
     private static final String DEFAULT_TOPIC = "topic";
     private final Map<Integer, List<Integer>> expectedReplicaAssignment = Map.of(0, List.of(0, 1, 2));

     @ClusterTest
     public void testDeleteTopicWithAllAliveReplicas(ClusterInstance cluster) throws Exception {
         try (Admin admin = cluster.admin()) {
             admin.createTopics(List.of(new NewTopic(DEFAULT_TOPIC, expectedReplicaAssignment))).all().get();
             admin.deleteTopics(List.of(DEFAULT_TOPIC)).all().get();
             cluster.waitForTopic(DEFAULT_TOPIC, 0);
         }
     }

     @ClusterTest
     public void testResumeDeleteTopicWithRecoveredFollower(ClusterInstance cluster) throws Exception {
         try (Admin admin = cluster.admin()) {
             admin.createTopics(List.of(new NewTopic(DEFAULT_TOPIC, expectedReplicaAssignment))).all().get();
             TopicPartition topicPartition = new TopicPartition(DEFAULT_TOPIC, 0);
             int leaderId = waitUtilLeaderIsKnown(cluster.brokers(), topicPartition);
             KafkaBroker follower = findFollower(cluster.brokers().values(), leaderId);

             // shutdown one follower replica
             follower.shutdown();
             admin.deleteTopics(List.of(DEFAULT_TOPIC)).all().get();

             TestUtils.waitForCondition(() -> cluster.brokers().values()
                     .stream()
                     .filter(broker -> broker.config().brokerId() != follower.config().brokerId())
                     .allMatch(b -> b.logManager().getLog(topicPartition, false).isEmpty()),
                 "Online replicas have not deleted log.");

             follower.startup();
             cluster.waitForTopic(DEFAULT_TOPIC, 0);
         }
     }

     @ClusterTest(brokers = 4)
     public void testPartitionReassignmentDuringDeleteTopic(ClusterInstance cluster) throws Exception {
         try (Admin admin = cluster.admin()) {
             admin.createTopics(List.of(new NewTopic(DEFAULT_TOPIC, expectedReplicaAssignment))).all().get();
             TopicPartition topicPartition = new TopicPartition(DEFAULT_TOPIC, 0);
             Map<Integer, KafkaBroker> servers = findPartitionHostingBrokers(cluster.brokers());
             int leaderId = waitUtilLeaderIsKnown(cluster.brokers(), topicPartition);
             KafkaBroker follower = findFollower(servers.values(), leaderId);
             follower.shutdown();

             admin.deleteTopics(List.of(DEFAULT_TOPIC)).all().get();
             Properties properties = new Properties();
             properties.setProperty(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, cluster.bootstrapServers());

             try (Admin otherAdmin = Admin.create(properties)) {
                 waitUtilTopicGone(otherAdmin);
                 assertThrows(ExecutionException.class, () -> otherAdmin.alterPartitionReassignments(
                     Map.of(topicPartition, Optional.of(new NewPartitionReassignment(List.of(1, 2, 3))))
                 ).all().get());
             }

             follower.startup();
             cluster.waitForTopic(DEFAULT_TOPIC, 0);
         }
     }

     @ClusterTest(brokers = 4)
     public void testIncreasePartitionCountDuringDeleteTopic(ClusterInstance cluster) throws Exception {
         try (Admin admin = cluster.admin()) {
             admin.createTopics(List.of(new NewTopic(DEFAULT_TOPIC, expectedReplicaAssignment))).all().get();
             TopicPartition topicPartition = new TopicPartition(DEFAULT_TOPIC, 0);
             Map<Integer, KafkaBroker> partitionHostingBrokers = findPartitionHostingBrokers(cluster.brokers());
             waitForReplicaCreated(partitionHostingBrokers, topicPartition, "Replicas for topic test not created.");
             int leaderId = waitUtilLeaderIsKnown(partitionHostingBrokers, topicPartition);
             KafkaBroker follower = findFollower(partitionHostingBrokers.values(), leaderId);
             // shutdown a broker to make sure the following topic deletion will be suspended
             follower.shutdown();
             admin.deleteTopics(List.of(DEFAULT_TOPIC)).all().get();

             // increase the partition count for topic
             Properties properties = new Properties();
             properties.setProperty(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, cluster.bootstrapServers());

             try (Admin otherAdmin = Admin.create(properties)) {
                 otherAdmin.createPartitions(Map.of(DEFAULT_TOPIC, NewPartitions.increaseTo(2))).all().get();
             } catch (ExecutionException ignored) {
                 // do nothing
             }

             follower.startup();
             cluster.waitForTopic(DEFAULT_TOPIC, 0);
         }
     }

     @ClusterTest
     public void testDeleteTopicDuringAddPartition(ClusterInstance cluster) throws Exception {
         try (Admin admin = cluster.admin()) {
             admin.createTopics(List.of(new NewTopic(DEFAULT_TOPIC, expectedReplicaAssignment))).all().get();
             int leaderId = waitUtilLeaderIsKnown(cluster.brokers(), new TopicPartition(DEFAULT_TOPIC, 0));
             TopicPartition newTopicPartition = new TopicPartition(DEFAULT_TOPIC, 1);
             KafkaBroker follower = findFollower(cluster.brokers().values(), leaderId);
             follower.shutdown();

             // wait until the broker is in shutting down state
             int followerBrokerId = follower.config().brokerId();
             TestUtils.waitForCondition(() -> follower.brokerState().equals(BrokerState.SHUTTING_DOWN),
                 "Follower " + followerBrokerId + " was not shutdown");
             Map<String, NewPartitions> newPartitionSet = Map.of(DEFAULT_TOPIC, NewPartitions.increaseTo(3));
             admin.createPartitions(newPartitionSet);
             cluster.waitForTopic(DEFAULT_TOPIC, 3);
             admin.deleteTopics(List.of(DEFAULT_TOPIC)).all().get();

             follower.startup();
             // test if topic deletion is resumed
             cluster.waitForTopic(DEFAULT_TOPIC, 0);
             waitForReplicaDeleted(cluster.brokers(), newTopicPartition, "Replica logs not for new partition [" + DEFAULT_TOPIC + ",1] not deleted after delete topic is complete.");
         }
     }

     @ClusterTest
     public void testAddPartitionDuringDeleteTopic(ClusterInstance cluster) throws Exception {
         try (Admin admin = cluster.admin()) {
             admin.createTopics(List.of(new NewTopic(DEFAULT_TOPIC, expectedReplicaAssignment))).all().get();
             // partitions to be added to the topic later
             TopicPartition newTopicPartition = new TopicPartition(DEFAULT_TOPIC, 1);
             admin.deleteTopics(List.of(DEFAULT_TOPIC)).all().get();
             Map<String, NewPartitions> newPartitionSet = Map.of(DEFAULT_TOPIC, NewPartitions.increaseTo(3));
             admin.createPartitions(newPartitionSet);
             cluster.waitForTopic(DEFAULT_TOPIC, 0);
             waitForReplicaDeleted(cluster.brokers(), newTopicPartition, "Replica logs not deleted after delete topic is complete");
         }
     }

     @ClusterTest
     public void testRecreateTopicAfterDeletion(ClusterInstance cluster) throws Exception {
         try (Admin admin = cluster.admin()) {
             admin.createTopics(List.of(new NewTopic(DEFAULT_TOPIC, expectedReplicaAssignment))).all().get();
             TopicPartition topicPartition = new TopicPartition(DEFAULT_TOPIC, 0);
             admin.deleteTopics(List.of(DEFAULT_TOPIC)).all().get();
             cluster.waitForTopic(DEFAULT_TOPIC, 0);
             // re-create topic on same replicas
             admin.createTopics(List.of(new NewTopic(DEFAULT_TOPIC, expectedReplicaAssignment))).all().get();
             waitForReplicaCreated(cluster.brokers(), topicPartition, "Replicas for topic " + DEFAULT_TOPIC + " not created.");
         }
     }
     @ClusterTest
     public void testDeleteNonExistingTopic(ClusterInstance cluster) throws Exception {
         try (Admin admin = cluster.admin()) {
             admin.createTopics(List.of(new NewTopic(DEFAULT_TOPIC, expectedReplicaAssignment))).all().get();
             TopicPartition topicPartition = new TopicPartition(DEFAULT_TOPIC, 0);
             String topic = "test2";
             TestUtils.waitForCondition(() -> {
                 try {
                     admin.deleteTopics(List.of(topic)).all().get();
                     return false;
                 } catch (Exception exception) {
                     return exception.getCause() instanceof UnknownTopicOrPartitionException;
                 }
             }, "Topic test2 should not exist.");

             cluster.waitForTopic(topic, 0);

             waitForReplicaCreated(cluster.brokers(), topicPartition, "Replicas for topic test not created.");
             TestUtils.waitUntilLeaderIsElectedOrChangedWithAdmin(admin, DEFAULT_TOPIC, 0, 1000);
         }
     }

     @ClusterTest(serverProperties = {
         @ClusterConfigProperty(key = "log.cleaner.enable", value = "true"),
         @ClusterConfigProperty(key = "log.cleanup.policy", value = "compact"),
         @ClusterConfigProperty(key = "log.cleaner.dedupe.buffer.size", value = "1048577")
     })
     public void testDeleteTopicWithCleaner(ClusterInstance cluster) throws Exception {
         try (Admin admin = cluster.admin()) {
             admin.createTopics(List.of(new NewTopic(DEFAULT_TOPIC, expectedReplicaAssignment))).all().get();
             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).isDefined(),
                 "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(Option.empty());
             // wait for cleaner to clean
             server.logManager().cleaner().awaitCleaned(topicPartition, 0, 60000);
             admin.deleteTopics(List.of(DEFAULT_TOPIC)).all().get();

             cluster.waitForTopic(DEFAULT_TOPIC, 0);
         }
     }

     @ClusterTest
     public void testDeleteTopicAlreadyMarkedAsDeleted(ClusterInstance cluster) throws Exception {
         try (Admin admin = cluster.admin()) {
             admin.createTopics(List.of(new NewTopic(DEFAULT_TOPIC, expectedReplicaAssignment))).all().get();
             admin.deleteTopics(List.of(DEFAULT_TOPIC)).all().get();

             TestUtils.waitForCondition(() -> {
                 try {
                     admin.deleteTopics(List.of(DEFAULT_TOPIC)).all().get();
                     return false;
                 } catch (Exception exception) {
                     return exception.getCause() instanceof UnknownTopicOrPartitionException;
                 }
             }, "Topic " + DEFAULT_TOPIC + " should be marked for deletion or already deleted.");

             cluster.waitForTopic(DEFAULT_TOPIC, 0);
         }
     }

     @ClusterTest(controllers = 1,
         serverProperties = {@ClusterConfigProperty(key = ServerConfigs.DELETE_TOPIC_ENABLE_CONFIG, value = "false")})
     public void testDisableDeleteTopic(ClusterInstance cluster) throws Exception {
         try (Admin admin = cluster.admin()) {
             admin.createTopics(List.of(new NewTopic(DEFAULT_TOPIC, expectedReplicaAssignment))).all().get();
             TopicPartition topicPartition = new TopicPartition(DEFAULT_TOPIC, 0);
             TestUtils.waitForCondition(() -> {
                 try {
                     admin.deleteTopics(List.of(DEFAULT_TOPIC)).all().get();
                     return false;
                 } catch (Exception exception) {
                     return exception.getCause() instanceof TopicDeletionDisabledException;
                 }
             }, "TopicDeletionDisabledException should be returned when deleting " + DEFAULT_TOPIC);

             waitForReplicaCreated(cluster.brokers(), topicPartition, "TopicDeletionDisabledException should be returned when deleting " + DEFAULT_TOPIC);
             assertDoesNotThrow(() -> admin.describeTopics(List.of(DEFAULT_TOPIC)).allTopicNames().get());
             assertDoesNotThrow(() -> waitUtilLeaderIsKnown(cluster.brokers(), topicPartition));
         }
     }

     private int waitUtilLeaderIsKnown(Map<Integer, KafkaBroker> idToBroker,
                                       TopicPartition topicPartition) throws InterruptedException {
         TestUtils.waitForCondition(() -> isLeaderKnown(idToBroker, topicPartition).get().isPresent(), 15000,
             "Partition " + topicPartition + " not made yet" + " after 15 seconds");
         return isLeaderKnown(idToBroker, topicPartition).get().get();
     }

     private void waitForReplicaCreated(Map<Integer, KafkaBroker> clusters,
                                        TopicPartition topicPartition,
                                        String failMessage) throws InterruptedException {
         TestUtils.waitForCondition(() -> clusters.values().stream().allMatch(broker ->
                 broker.logManager().getLog(topicPartition, false).isDefined()),
             failMessage);
     }

     private void waitForReplicaDeleted(Map<Integer, KafkaBroker> clusters,
                                        TopicPartition newTopicPartition,
                                        String failMessage) throws InterruptedException {
         TestUtils.waitForCondition(() -> clusters.values().stream().allMatch(broker ->
                 broker.logManager().getLog(newTopicPartition, false).isEmpty()),
             failMessage);
     }

     private Supplier<Optional<Integer>> isLeaderKnown(Map<Integer, KafkaBroker> idToBroker, TopicPartition topicPartition) {
         return () -> idToBroker.values()
             .stream()
             .filter(broker -> OptionConverters.toJava(broker.replicaManager().onlinePartition(topicPartition))
                 .stream().anyMatch(tp -> tp.leaderIdIfLocal().isDefined()))
             .map(broker -> broker.config().brokerId())
             .findFirst();
     }

     private KafkaBroker findFollower(Collection<KafkaBroker> idToBroker, int leaderId) {
         return idToBroker.stream()
             .filter(broker -> broker.config().brokerId() != leaderId)
             .findFirst()
             .orElseGet(() -> fail("Can't find any follower"));
     }

     private void waitUtilTopicGone(Admin admin) throws Exception {
         TestUtils.waitForCondition(() -> {
             try {
                 admin.describeTopics(List.of(DEFAULT_TOPIC)).allTopicNames().get();
                 return false;
             } catch (Exception exception) {
                 return exception.getCause() instanceof UnknownTopicOrPartitionException;
             }
         }, "Topic" + DEFAULT_TOPIC + " should be deleted");
     }

     private Map<Integer, KafkaBroker> findPartitionHostingBrokers(Map<Integer, KafkaBroker> brokers) {
         return brokers.entrySet()
                 .stream()
                 .filter(broker -> expectedReplicaAssignment.get(0).contains(broker.getValue().config().brokerId()))
                 .collect(Collectors.toMap(Map.Entry::getKey, Map.Entry::getValue));
     }

     private List<int[]> writeDups(int numKeys, int numDups, UnifiedLog log) {
         int counter = 0;
         List<int[]> result = new ArrayList<>();

         for (int i = 0; i < numDups; i++) {
             for (int key = 0; key < numKeys; key++) {
                 int count = counter;
                 log.appendAsLeader(
                     MemoryRecords.withRecords(
                         Compression.NONE,
                         new SimpleRecord(
                             String.valueOf(key).getBytes(),
                             String.valueOf(counter).getBytes()
                         )
                     ),
                     0,
                     AppendOrigin.CLIENT,
                     RequestLocal.noCaching(),
                     VerificationGuard.SENTINEL
                 );
                 counter++;
                 result.add(new int[] {key, count});
             }
         }
         return result;
     }
 }
	/*
	* Licensed to the Apache Software Foundation (ASF) under one or more
	* contributor license agreements. See the NOTICE file distributed with
	* this work for additional information regarding copyright ownership.
	* The ASF licenses this file to You under the Apache License, Version 2.0
	* (the "License"); you may not use this file except in compliance with
	* the License. You may obtain a copy of the License at
	*
	* http://www.apache.org/licenses/LICENSE-2.0
	*
	* Unless required by applicable law or agreed to in writing, software
	* distributed under the License is distributed on an "AS IS" BASIS,
	* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
	* See the License for the specific language governing permissions and
	* limitations under the License.
	*/
	package kafka.admin;

	import kafka.log.UnifiedLog;
	import kafka.server.KafkaBroker;

	import org.apache.kafka.clients.admin.Admin;
	import org.apache.kafka.clients.admin.AdminClientConfig;
	import org.apache.kafka.clients.admin.NewPartitionReassignment;
	import org.apache.kafka.clients.admin.NewPartitions;
	import org.apache.kafka.clients.admin.NewTopic;
	import org.apache.kafka.common.TopicPartition;
	import org.apache.kafka.common.compress.Compression;
	import org.apache.kafka.common.errors.TopicDeletionDisabledException;
	import org.apache.kafka.common.errors.UnknownTopicOrPartitionException;
	import org.apache.kafka.common.record.MemoryRecords;
	import org.apache.kafka.common.record.SimpleRecord;
	import org.apache.kafka.common.test.ClusterInstance;
	import org.apache.kafka.common.test.TestUtils;
	import org.apache.kafka.common.test.api.ClusterConfigProperty;
	import org.apache.kafka.common.test.api.ClusterTest;
	import org.apache.kafka.common.test.api.ClusterTestDefaults;
	import org.apache.kafka.common.test.api.Type;
	import org.apache.kafka.coordinator.group.GroupCoordinatorConfig;
	import org.apache.kafka.metadata.BrokerState;
	import org.apache.kafka.server.common.RequestLocal;
	import org.apache.kafka.server.config.ServerConfigs;
	import org.apache.kafka.storage.internals.log.AppendOrigin;
	import org.apache.kafka.storage.internals.log.VerificationGuard;

	import java.util.ArrayList;
	import java.util.Collection;
	import java.util.List;
	import java.util.Map;
	import java.util.Optional;
	import java.util.Properties;
	import java.util.concurrent.ExecutionException;
	import java.util.function.Supplier;
	import java.util.stream.Collectors;

	import scala.Option;
	import scala.jdk.javaapi.OptionConverters;

	import static org.junit.jupiter.api.Assertions.assertDoesNotThrow;
	import static org.junit.jupiter.api.Assertions.assertThrows;
	import static org.junit.jupiter.api.Assertions.fail;


	@ClusterTestDefaults(types = {Type.KRAFT},
	brokers = 3,
	serverProperties = {
	@ClusterConfigProperty(key = GroupCoordinatorConfig.OFFSETS_TOPIC_PARTITIONS_CONFIG, value = "1"),
	@ClusterConfigProperty(key = GroupCoordinatorConfig.OFFSETS_TOPIC_REPLICATION_FACTOR_CONFIG, value = "1"),
	@ClusterConfigProperty(key = "log.initial.task.delay.ms", value = "100"),
	@ClusterConfigProperty(key = "log.segment.delete.delay.ms", value = "1000")
	})
	public class DeleteTopicTest {
	private static final String DEFAULT_TOPIC = "topic";
	private final Map<Integer, List<Integer>> expectedReplicaAssignment = Map.of(0, List.of(0, 1, 2));

	@ClusterTest
	public void testDeleteTopicWithAllAliveReplicas(ClusterInstance cluster) throws Exception {
	try (Admin admin = cluster.admin()) {
	admin.createTopics(List.of(new NewTopic(DEFAULT_TOPIC, expectedReplicaAssignment))).all().get();
	admin.deleteTopics(List.of(DEFAULT_TOPIC)).all().get();
	cluster.waitForTopic(DEFAULT_TOPIC, 0);
	}
	}

	@ClusterTest
	public void testResumeDeleteTopicWithRecoveredFollower(ClusterInstance cluster) throws Exception {
	try (Admin admin = cluster.admin()) {
	admin.createTopics(List.of(new NewTopic(DEFAULT_TOPIC, expectedReplicaAssignment))).all().get();
	TopicPartition topicPartition = new TopicPartition(DEFAULT_TOPIC, 0);
	int leaderId = waitUtilLeaderIsKnown(cluster.brokers(), topicPartition);
	KafkaBroker follower = findFollower(cluster.brokers().values(), leaderId);

	// shutdown one follower replica
	follower.shutdown();
	admin.deleteTopics(List.of(DEFAULT_TOPIC)).all().get();

	TestUtils.waitForCondition(() -> cluster.brokers().values()
	.stream()
	.filter(broker -> broker.config().brokerId() != follower.config().brokerId())
	.allMatch(b -> b.logManager().getLog(topicPartition, false).isEmpty()),
	"Online replicas have not deleted log.");

	follower.startup();
	cluster.waitForTopic(DEFAULT_TOPIC, 0);
	}
	}

	@ClusterTest(brokers = 4)
	public void testPartitionReassignmentDuringDeleteTopic(ClusterInstance cluster) throws Exception {
	try (Admin admin = cluster.admin()) {
	admin.createTopics(List.of(new NewTopic(DEFAULT_TOPIC, expectedReplicaAssignment))).all().get();
	TopicPartition topicPartition = new TopicPartition(DEFAULT_TOPIC, 0);
	Map<Integer, KafkaBroker> servers = findPartitionHostingBrokers(cluster.brokers());
	int leaderId = waitUtilLeaderIsKnown(cluster.brokers(), topicPartition);
	KafkaBroker follower = findFollower(servers.values(), leaderId);
	follower.shutdown();

	admin.deleteTopics(List.of(DEFAULT_TOPIC)).all().get();
	Properties properties = new Properties();
	properties.setProperty(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, cluster.bootstrapServers());

	try (Admin otherAdmin = Admin.create(properties)) {
	waitUtilTopicGone(otherAdmin);
	assertThrows(ExecutionException.class, () -> otherAdmin.alterPartitionReassignments(
	Map.of(topicPartition, Optional.of(new NewPartitionReassignment(List.of(1, 2, 3))))
	).all().get());
	}

	follower.startup();
	cluster.waitForTopic(DEFAULT_TOPIC, 0);
	}
	}

	@ClusterTest(brokers = 4)
	public void testIncreasePartitionCountDuringDeleteTopic(ClusterInstance cluster) throws Exception {
	try (Admin admin = cluster.admin()) {
	admin.createTopics(List.of(new NewTopic(DEFAULT_TOPIC, expectedReplicaAssignment))).all().get();
	TopicPartition topicPartition = new TopicPartition(DEFAULT_TOPIC, 0);
	Map<Integer, KafkaBroker> partitionHostingBrokers = findPartitionHostingBrokers(cluster.brokers());
	waitForReplicaCreated(partitionHostingBrokers, topicPartition, "Replicas for topic test not created.");
	int leaderId = waitUtilLeaderIsKnown(partitionHostingBrokers, topicPartition);
	KafkaBroker follower = findFollower(partitionHostingBrokers.values(), leaderId);
	// shutdown a broker to make sure the following topic deletion will be suspended
	follower.shutdown();
	admin.deleteTopics(List.of(DEFAULT_TOPIC)).all().get();

	// increase the partition count for topic
	Properties properties = new Properties();
	properties.setProperty(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, cluster.bootstrapServers());

	try (Admin otherAdmin = Admin.create(properties)) {
	otherAdmin.createPartitions(Map.of(DEFAULT_TOPIC, NewPartitions.increaseTo(2))).all().get();
	} catch (ExecutionException ignored) {
	// do nothing
	}

	follower.startup();
	cluster.waitForTopic(DEFAULT_TOPIC, 0);
	}
	}

	@ClusterTest
	public void testDeleteTopicDuringAddPartition(ClusterInstance cluster) throws Exception {
	try (Admin admin = cluster.admin()) {
	admin.createTopics(List.of(new NewTopic(DEFAULT_TOPIC, expectedReplicaAssignment))).all().get();
	int leaderId = waitUtilLeaderIsKnown(cluster.brokers(), new TopicPartition(DEFAULT_TOPIC, 0));
	TopicPartition newTopicPartition = new TopicPartition(DEFAULT_TOPIC, 1);
	KafkaBroker follower = findFollower(cluster.brokers().values(), leaderId);
	follower.shutdown();

	// wait until the broker is in shutting down state
	int followerBrokerId = follower.config().brokerId();
	TestUtils.waitForCondition(() -> follower.brokerState().equals(BrokerState.SHUTTING_DOWN),
	"Follower " + followerBrokerId + " was not shutdown");
	Map<String, NewPartitions> newPartitionSet = Map.of(DEFAULT_TOPIC, NewPartitions.increaseTo(3));
	admin.createPartitions(newPartitionSet);
	cluster.waitForTopic(DEFAULT_TOPIC, 3);
	admin.deleteTopics(List.of(DEFAULT_TOPIC)).all().get();

	follower.startup();
	// test if topic deletion is resumed
	cluster.waitForTopic(DEFAULT_TOPIC, 0);
	waitForReplicaDeleted(cluster.brokers(), newTopicPartition, "Replica logs not for new partition [" + DEFAULT_TOPIC + ",1] not deleted after delete topic is complete.");
	}
	}

	@ClusterTest
	public void testAddPartitionDuringDeleteTopic(ClusterInstance cluster) throws Exception {
	try (Admin admin = cluster.admin()) {
	admin.createTopics(List.of(new NewTopic(DEFAULT_TOPIC, expectedReplicaAssignment))).all().get();
	// partitions to be added to the topic later
	TopicPartition newTopicPartition = new TopicPartition(DEFAULT_TOPIC, 1);
	admin.deleteTopics(List.of(DEFAULT_TOPIC)).all().get();
	Map<String, NewPartitions> newPartitionSet = Map.of(DEFAULT_TOPIC, NewPartitions.increaseTo(3));
	admin.createPartitions(newPartitionSet);
	cluster.waitForTopic(DEFAULT_TOPIC, 0);
	waitForReplicaDeleted(cluster.brokers(), newTopicPartition, "Replica logs not deleted after delete topic is complete");
	}
	}

	@ClusterTest
	public void testRecreateTopicAfterDeletion(ClusterInstance cluster) throws Exception {
	try (Admin admin = cluster.admin()) {
	admin.createTopics(List.of(new NewTopic(DEFAULT_TOPIC, expectedReplicaAssignment))).all().get();
	TopicPartition topicPartition = new TopicPartition(DEFAULT_TOPIC, 0);
	admin.deleteTopics(List.of(DEFAULT_TOPIC)).all().get();
	cluster.waitForTopic(DEFAULT_TOPIC, 0);
	// re-create topic on same replicas
	admin.createTopics(List.of(new NewTopic(DEFAULT_TOPIC, expectedReplicaAssignment))).all().get();
	waitForReplicaCreated(cluster.brokers(), topicPartition, "Replicas for topic " + DEFAULT_TOPIC + " not created.");
	}
	}
	@ClusterTest
	public void testDeleteNonExistingTopic(ClusterInstance cluster) throws Exception {
	try (Admin admin = cluster.admin()) {
	admin.createTopics(List.of(new NewTopic(DEFAULT_TOPIC, expectedReplicaAssignment))).all().get();
	TopicPartition topicPartition = new TopicPartition(DEFAULT_TOPIC, 0);
	String topic = "test2";
	TestUtils.waitForCondition(() -> {
	try {
	admin.deleteTopics(List.of(topic)).all().get();
	return false;
	} catch (Exception exception) {
	return exception.getCause() instanceof UnknownTopicOrPartitionException;
	}
	}, "Topic test2 should not exist.");

	cluster.waitForTopic(topic, 0);

	waitForReplicaCreated(cluster.brokers(), topicPartition, "Replicas for topic test not created.");
	TestUtils.waitUntilLeaderIsElectedOrChangedWithAdmin(admin, DEFAULT_TOPIC, 0, 1000);
	}
	}

	@ClusterTest(serverProperties = {
	@ClusterConfigProperty(key = "log.cleaner.enable", value = "true"),
	@ClusterConfigProperty(key = "log.cleanup.policy", value = "compact"),
	@ClusterConfigProperty(key = "log.cleaner.dedupe.buffer.size", value = "1048577")
	})
	public void testDeleteTopicWithCleaner(ClusterInstance cluster) throws Exception {
	try (Admin admin = cluster.admin()) {
	admin.createTopics(List.of(new NewTopic(DEFAULT_TOPIC, expectedReplicaAssignment))).all().get();
	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).isDefined(),
	"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(Option.empty());
	// wait for cleaner to clean
	server.logManager().cleaner().awaitCleaned(topicPartition, 0, 60000);
	admin.deleteTopics(List.of(DEFAULT_TOPIC)).all().get();

	cluster.waitForTopic(DEFAULT_TOPIC, 0);
	}
	}

	@ClusterTest
	public void testDeleteTopicAlreadyMarkedAsDeleted(ClusterInstance cluster) throws Exception {
	try (Admin admin = cluster.admin()) {
	admin.createTopics(List.of(new NewTopic(DEFAULT_TOPIC, expectedReplicaAssignment))).all().get();
	admin.deleteTopics(List.of(DEFAULT_TOPIC)).all().get();

	TestUtils.waitForCondition(() -> {
	try {
	admin.deleteTopics(List.of(DEFAULT_TOPIC)).all().get();
	return false;
	} catch (Exception exception) {
	return exception.getCause() instanceof UnknownTopicOrPartitionException;
	}
	}, "Topic " + DEFAULT_TOPIC + " should be marked for deletion or already deleted.");

	cluster.waitForTopic(DEFAULT_TOPIC, 0);
	}
	}

	@ClusterTest(controllers = 1,
	serverProperties = {@ClusterConfigProperty(key = ServerConfigs.DELETE_TOPIC_ENABLE_CONFIG, value = "false")})
	public void testDisableDeleteTopic(ClusterInstance cluster) throws Exception {
	try (Admin admin = cluster.admin()) {
	admin.createTopics(List.of(new NewTopic(DEFAULT_TOPIC, expectedReplicaAssignment))).all().get();
	TopicPartition topicPartition = new TopicPartition(DEFAULT_TOPIC, 0);
	TestUtils.waitForCondition(() -> {
	try {
	admin.deleteTopics(List.of(DEFAULT_TOPIC)).all().get();
	return false;
	} catch (Exception exception) {
	return exception.getCause() instanceof TopicDeletionDisabledException;
	}
	}, "TopicDeletionDisabledException should be returned when deleting " + DEFAULT_TOPIC);

	waitForReplicaCreated(cluster.brokers(), topicPartition, "TopicDeletionDisabledException should be returned when deleting " + DEFAULT_TOPIC);
	assertDoesNotThrow(() -> admin.describeTopics(List.of(DEFAULT_TOPIC)).allTopicNames().get());
	assertDoesNotThrow(() -> waitUtilLeaderIsKnown(cluster.brokers(), topicPartition));
	}
	}

	private int waitUtilLeaderIsKnown(Map<Integer, KafkaBroker> idToBroker,
	TopicPartition topicPartition) throws InterruptedException {
	TestUtils.waitForCondition(() -> isLeaderKnown(idToBroker, topicPartition).get().isPresent(), 15000,
	"Partition " + topicPartition + " not made yet" + " after 15 seconds");
	return isLeaderKnown(idToBroker, topicPartition).get().get();
	}

	private void waitForReplicaCreated(Map<Integer, KafkaBroker> clusters,
	TopicPartition topicPartition,
	String failMessage) throws InterruptedException {
	TestUtils.waitForCondition(() -> clusters.values().stream().allMatch(broker ->
	broker.logManager().getLog(topicPartition, false).isDefined()),
	failMessage);
	}

	private void waitForReplicaDeleted(Map<Integer, KafkaBroker> clusters,
	TopicPartition newTopicPartition,
	String failMessage) throws InterruptedException {
	TestUtils.waitForCondition(() -> clusters.values().stream().allMatch(broker ->
	broker.logManager().getLog(newTopicPartition, false).isEmpty()),
	failMessage);
	}

	private Supplier<Optional<Integer>> isLeaderKnown(Map<Integer, KafkaBroker> idToBroker, TopicPartition topicPartition) {
	return () -> idToBroker.values()
	.stream()
	.filter(broker -> OptionConverters.toJava(broker.replicaManager().onlinePartition(topicPartition))
	.stream().anyMatch(tp -> tp.leaderIdIfLocal().isDefined()))
	.map(broker -> broker.config().brokerId())
	.findFirst();
	}

	private KafkaBroker findFollower(Collection<KafkaBroker> idToBroker, int leaderId) {
	return idToBroker.stream()
	.filter(broker -> broker.config().brokerId() != leaderId)
	.findFirst()
	.orElseGet(() -> fail("Can't find any follower"));
	}

	private void waitUtilTopicGone(Admin admin) throws Exception {
	TestUtils.waitForCondition(() -> {
	try {
	admin.describeTopics(List.of(DEFAULT_TOPIC)).allTopicNames().get();
	return false;
	} catch (Exception exception) {
	return exception.getCause() instanceof UnknownTopicOrPartitionException;
	}
	}, "Topic" + DEFAULT_TOPIC + " should be deleted");
	}

	private Map<Integer, KafkaBroker> findPartitionHostingBrokers(Map<Integer, KafkaBroker> brokers) {
	return brokers.entrySet()
	.stream()
	.filter(broker -> expectedReplicaAssignment.get(0).contains(broker.getValue().config().brokerId()))
	.collect(Collectors.toMap(Map.Entry::getKey, Map.Entry::getValue));
	}

	private List<int[]> writeDups(int numKeys, int numDups, UnifiedLog log) {
	int counter = 0;
	List<int[]> result = new ArrayList<>();

	for (int i = 0; i < numDups; i++) {
	for (int key = 0; key < numKeys; key++) {
	int count = counter;
	log.appendAsLeader(
	MemoryRecords.withRecords(
	Compression.NONE,
	new SimpleRecord(
	String.valueOf(key).getBytes(),
	String.valueOf(counter).getBytes()
	)
	),
	0,
	AppendOrigin.CLIENT,
	RequestLocal.noCaching(),
	VerificationGuard.SENTINEL
	);
	counter++;
	result.add(new int[] {key, count});
	}
	}
	return result;
	}
	}