You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@kafka.apache.org by ch...@apache.org on 2024/11/09 19:06:57 UTC
(kafka) branch trunk updated: KAFKA-17837 Rewrite DeleteTopicTest (#17579)
This is an automated email from the ASF dual-hosted git repository.
chia7712 pushed a commit to branch trunk
in repository https://gitbox.apache.org/repos/asf/kafka.git
The following commit(s) were added to refs/heads/trunk by this push:
new 393455eb1ad KAFKA-17837 Rewrite DeleteTopicTest (#17579)
393455eb1ad is described below
commit 393455eb1ada86eaea84f9485efe7f133de2583c
Author: TengYao Chi <ki...@gmail.com>
AuthorDate: Sun Nov 10 03:06:52 2024 +0800
KAFKA-17837 Rewrite DeleteTopicTest (#17579)
Reviewers: Chia-Ping Tsai <ch...@gmail.com>
---
build.gradle | 1 +
checkstyle/import-control-core.xml | 4 +
checkstyle/import-control-test-common.xml | 1 +
.../src/test/java/kafka/admin/DeleteTopicTest.java | 387 ++++++++++++++
.../scala/unit/kafka/admin/DeleteTopicTest.scala | 573 ---------------------
.../org/apache/kafka/common/test/TestUtils.java | 54 ++
.../kafka/common/test/api/ClusterInstance.java | 65 ++-
.../common/test/api/ClusterTestExtensionsTest.java | 9 +-
.../org/apache/kafka/tools/TopicCommandTest.java | 2 +
9 files changed, 519 insertions(+), 577 deletions(-)
diff --git a/build.gradle b/build.gradle
index afc142a9684..20eeb3eff47 100644
--- a/build.gradle
+++ b/build.gradle
@@ -1577,6 +1577,7 @@ project(':test-common:test-common-api') {
implementation project(':raft')
implementation project(':server')
implementation project(':server-common')
+ implementation project(':storage')
implementation project(':test-common')
implementation libs.junitJupiterApi
diff --git a/checkstyle/import-control-core.xml b/checkstyle/import-control-core.xml
index b56e4c09670..a8dc78160e3 100644
--- a/checkstyle/import-control-core.xml
+++ b/checkstyle/import-control-core.xml
@@ -123,14 +123,18 @@
<subpackage name="admin">
<allow pkg="kafka.admin"/>
<allow pkg="kafka.cluster"/>
+ <allow pkg="kafka.log"/>
<allow pkg="kafka.security.authorizer"/>
<allow pkg="kafka.server"/>
<allow pkg="kafka.zk"/>
<allow pkg="org.apache.kafka.clients.admin"/>
<allow pkg="org.apache.kafka.coordinator.group"/>
+ <allow pkg="org.apache.kafka.metadata"/>
<allow pkg="org.apache.kafka.metadata.authorizer"/>
<allow pkg="org.apache.kafka.security"/>
<allow pkg="org.apache.kafka.server"/>
+ <allow pkg="org.apache.kafka.storage.internals.checkpoint"/>
+ <allow pkg="org.apache.kafka.storage.internals.log"/>
<allow pkg="org.apache.kafka.test"/>
<allow pkg="org.apache.log4j"/>
<allow pkg="org.apache.kafka.common.test"/>
diff --git a/checkstyle/import-control-test-common.xml b/checkstyle/import-control-test-common.xml
index 015aed73fc7..9fe7f4d4844 100644
--- a/checkstyle/import-control-test-common.xml
+++ b/checkstyle/import-control-test-common.xml
@@ -23,4 +23,5 @@
<allow pkg="java" />
<allow pkg="org" />
<allow pkg="kafka" />
+ <allow pkg="scala.jdk.javaapi" />
</import-control>
diff --git a/core/src/test/java/kafka/admin/DeleteTopicTest.java b/core/src/test/java/kafka/admin/DeleteTopicTest.java
new file mode 100644
index 00000000000..76173378e05
--- /dev/null
+++ b/core/src/test/java/kafka/admin/DeleteTopicTest.java
@@ -0,0 +1,387 @@
+/*
+ * 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.TestUtils;
+import org.apache.kafka.common.test.api.ClusterConfigProperty;
+import org.apache.kafka.common.test.api.ClusterInstance;
+import org.apache.kafka.common.test.api.ClusterTest;
+import org.apache.kafka.common.test.api.ClusterTestDefaults;
+import org.apache.kafka.common.test.api.ClusterTestExtensions;
+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.MetadataVersion;
+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 org.junit.jupiter.api.extension.ExtendWith;
+
+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.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;
+
+
+@ExtendWith(value = ClusterTestExtensions.class)
+@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.createAdminClient()) {
+ 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.createAdminClient()) {
+ 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.createAdminClient()) {
+ 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.createAdminClient()) {
+ 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.createAdminClient()) {
+ 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.createAdminClient()) {
+ 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.createAdminClient()) {
+ 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.createAdminClient()) {
+ 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.segment.bytes", value = "100"),
+ @ClusterConfigProperty(key = "log.cleaner.dedupe.buffer.size", value = "1048577")
+ })
+ public void testDeleteTopicWithCleaner(ClusterInstance cluster) throws Exception {
+ try (Admin admin = cluster.createAdminClient()) {
+ 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);
+ // 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.createAdminClient()) {
+ 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.createAdminClient()) {
+ 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,
+ MetadataVersion.LATEST_PRODUCTION,
+ RequestLocal.noCaching(),
+ VerificationGuard.SENTINEL
+ );
+ counter++;
+ result.add(new int[] {key, count});
+ }
+ }
+ return result;
+ }
+}
\ No newline at end of file
diff --git a/core/src/test/scala/unit/kafka/admin/DeleteTopicTest.scala b/core/src/test/scala/unit/kafka/admin/DeleteTopicTest.scala
deleted file mode 100644
index ef4585d2ab4..00000000000
--- a/core/src/test/scala/unit/kafka/admin/DeleteTopicTest.scala
+++ /dev/null
@@ -1,573 +0,0 @@
-/**
- * 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.controller.{OfflineReplica, PartitionAndReplica, ReplicaAssignment, ReplicaDeletionSuccessful}
-import kafka.log.UnifiedLog
-import kafka.server.{KafkaBroker, KafkaConfig, KafkaServer, QuorumTestHarness}
-import kafka.utils.TestUtils.waitForAllPartitionsMetadata
-import kafka.utils._
-import kafka.zk.TopicPartitionZNode
-import org.apache.kafka.clients.admin.{Admin, AdminClientConfig, NewPartitionReassignment, NewPartitions}
-import org.apache.kafka.common.TopicPartition
-import org.apache.kafka.common.errors.{TopicDeletionDisabledException, UnknownTopicOrPartitionException}
-import org.apache.kafka.common.network.ListenerName
-import org.apache.kafka.common.security.auth.SecurityProtocol
-import org.apache.kafka.metadata.BrokerState
-import org.apache.kafka.server.config.ServerConfigs
-import org.junit.jupiter.api.Assertions._
-import org.junit.jupiter.api.{AfterEach, Test}
-import org.junit.jupiter.params.ParameterizedTest
-import org.junit.jupiter.params.provider.ValueSource
-
-import java.util
-import java.util.concurrent.ExecutionException
-import java.util.{Collections, Optional, Properties}
-import scala.collection.{Map, Seq}
-import scala.jdk.CollectionConverters._
-import scala.util.Using
-
-class DeleteTopicTest extends QuorumTestHarness {
-
- var brokers: Seq[KafkaBroker] = Seq()
-
- var admin: Admin = _
-
- val topic = "test"
-
- val expectedReplicaAssignment = Map(0 -> List(0, 1, 2))
- val expectedReplicaFullAssignment = expectedReplicaAssignment.map { case (k, v) =>
- k -> ReplicaAssignment(v, List(), List())
- }
-
- @AfterEach
- override def tearDown(): Unit = {
- TestUtils.shutdownServers(brokers)
- if (admin != null) admin.close()
- super.tearDown()
- }
-
- @ParameterizedTest
- @ValueSource(strings = Array("zk", "kraft"))
- def testDeleteTopicWithAllAliveReplicas(quorum: String): Unit = {
- brokers = createTestTopicAndCluster(topic)
- // start topic deletion
- admin.deleteTopics(Collections.singletonList(topic)).all().get()
- TestUtils.verifyTopicDeletion(zkClientOrNull, topic, 1, brokers)
- }
-
- @ParameterizedTest
- @ValueSource(strings = Array("zk", "kraft"))
- def testResumeDeleteTopicWithRecoveredFollower(quorum: String): Unit = {
- val topicPartition = new TopicPartition("test", 0)
- val topic = topicPartition.topic
- brokers = createTestTopicAndCluster(topic)
- // shut down one follower replica
- val leaderId = TestUtils.waitUntilLeaderIsKnown(brokers, new TopicPartition(topic, 0))
- val follower = brokers.filter(s => s.config.brokerId != leaderId).last
- follower.shutdown()
- // start topic deletion
- admin.deleteTopics(Collections.singletonList(topic)).all().get()
- // check if all replicas but the one that is shut down has deleted the log
- TestUtils.waitUntilTrue(() =>
- brokers.filter(s => s.config.brokerId != follower.config.brokerId)
- .forall(_.logManager.getLog(topicPartition).isEmpty), "Replicas 0,1 have not deleted log.")
-
- if (!isKRaftTest()) {
- // ensure topic deletion is halted
- TestUtils.waitUntilTrue(() => zkClient.isTopicMarkedForDeletion(topic),
- "Admin path /admin/delete_topics/test path deleted even when a follower replica is down")
- }
-
- // restart follower replica
- follower.startup()
- TestUtils.verifyTopicDeletion(zkClientOrNull, topic, 1, brokers)
- }
-
- // This test is not enabled with KRaft because it doesn't have the same controller failover logic as Zookeeper mode.
- @Test
- def testResumeDeleteTopicOnControllerFailover(): Unit = {
- val topicPartition = new TopicPartition("test", 0)
- val topic = topicPartition.topic
- brokers = createTestTopicAndCluster(topic)
- val controllerId = zkClient.getControllerId.getOrElse(fail("Controller doesn't exist"))
- val controller = brokers.filter(s => s.config.brokerId == controllerId).head
- val leaderIdOpt = zkClient.getLeaderForPartition(new TopicPartition(topic, 0))
- val follower = brokers.filter(s => s.config.brokerId != leaderIdOpt.get && s.config.brokerId != controllerId).last
- follower.shutdown()
-
- // start topic deletion
- adminZkClient.deleteTopic(topic)
- // shut down the controller to trigger controller failover during delete topic
- controller.shutdown()
-
- // ensure topic deletion is halted
- TestUtils.waitUntilTrue(() => zkClient.isTopicMarkedForDeletion(topic),
- "Admin path /admin/delete_topics/test path deleted even when a replica is down")
-
- controller.startup()
- follower.startup()
-
- TestUtils.verifyTopicDeletion(zkClient, topic, 1, brokers)
- }
-
- @ParameterizedTest
- @ValueSource(strings = Array("zk", "kraft"))
- def testPartitionReassignmentDuringDeleteTopic(quorum: String): Unit = {
- val topicPartition = new TopicPartition(topic, 0)
-
- // create brokers
- brokers = createTestTopicAndCluster(topic, 4)
- val servers = brokers.filter(s => expectedReplicaAssignment(0).contains(s.config.brokerId))
-
- val leaderIdOpt = TestUtils.waitUntilLeaderIsKnown(brokers, topicPartition)
- val follower = servers.filter(s => s.config.brokerId != leaderIdOpt).last
- follower.shutdown()
- // start topic deletion
- admin.deleteTopics(Collections.singletonList(topic)).all().get()
- // verify that a partition from the topic cannot be reassigned
- val props = new Properties()
- props.setProperty(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, TestUtils.plaintextBootstrapServers(servers))
- val adminClient = Admin.create(props)
- try {
- waitUntilTopicGone(adminClient, "test")
- verifyReassignmentFailsForMissing(adminClient, new TopicPartition(topic, 0),
- new NewPartitionReassignment(util.Arrays.asList(1, 2, 3)))
- } finally {
- adminClient.close()
- }
- follower.startup()
- TestUtils.verifyTopicDeletion(zkClientOrNull, topic, 1, servers)
- }
-
- private def waitUntilTopicGone(adminClient: Admin, topicName: String): Unit = {
- TestUtils.waitUntilTrue(() => {
- try {
- adminClient.describeTopics(util.Collections.singletonList(topicName)).allTopicNames().get()
- false
- } catch {
- case e: ExecutionException =>
- classOf[UnknownTopicOrPartitionException].equals(e.getCause.getClass)
- }
- }, s"Topic ${topicName} should be deleted.")
- }
-
- private def verifyReassignmentFailsForMissing(adminClient: Admin,
- partition: TopicPartition,
- reassignment: NewPartitionReassignment): Unit = {
- val e = assertThrows(classOf[ExecutionException], () => adminClient.alterPartitionReassignments(Collections.singletonMap(partition,
- Optional.of(reassignment))).all().get())
- assertEquals(classOf[UnknownTopicOrPartitionException], e.getCause.getClass)
- }
-
- private def getController() : (KafkaServer, Int) = {
- val controllerId = zkClient.getControllerId.getOrElse(throw new AssertionError("Controller doesn't exist"))
- val controller = brokers.find(s => s.config.brokerId == controllerId).get.asInstanceOf[KafkaServer]
- (controller, controllerId)
- }
-
- private def ensureControllerExists() = {
- TestUtils.waitUntilTrue(() => {
- try {
- getController()
- true
- } catch {
- case _: Throwable => false
- }
- }, "Controller should eventually exist")
- }
-
- private def getAllReplicasFromAssignment(topic : String, assignment : Map[Int, Seq[Int]]) : Set[PartitionAndReplica] = {
- assignment.flatMap { case (partition, replicas) =>
- replicas.map {r => new PartitionAndReplica(new TopicPartition(topic, partition), r)}
- }.toSet
- }
-
- @ParameterizedTest
- @ValueSource(strings = Array("zk", "kraft"))
- def testIncreasePartitionCountDuringDeleteTopic(quorum: String): Unit = {
- val topicPartition = new TopicPartition(topic, 0)
- val allBrokers = createTestTopicAndCluster(topic, 4)
- this.brokers = allBrokers
- val partitionHostingBrokers = allBrokers.filter(b => expectedReplicaAssignment(0).contains(b.config.brokerId))
-
- // wait until replica log is created on every broker
- TestUtils.waitUntilTrue(() => partitionHostingBrokers.forall(_.logManager.getLog(topicPartition).isDefined),
- "Replicas for topic test not created.")
-
- val leaderIdOpt = TestUtils.waitUntilLeaderIsKnown(partitionHostingBrokers, topicPartition)
-
- if (isKRaftTest()) {
- // shutdown a broker to make sure the following topic deletion will be suspended
- val follower = partitionHostingBrokers.filter(s => s.config.brokerId != leaderIdOpt).last
- follower.shutdown()
- // start topic deletion
- admin.deleteTopics(Collections.singletonList(topic)).all().get()
-
- // increase the partition count for topic
- val props = new Properties()
- props.setProperty(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, TestUtils.plaintextBootstrapServers(partitionHostingBrokers))
- Using(Admin.create(props)) { adminClient =>
- try {
- adminClient.createPartitions(Map(topic -> NewPartitions.increaseTo(2)).asJava).all().get()
- } catch {
- case _: ExecutionException =>
- }
- }
-
- // bring back the failed broker
- follower.startup()
- } else {
- // convert brokers to KafkaServer objects
- val partitionHostingServers = partitionHostingBrokers.map(_.asInstanceOf[KafkaServer]).toList
-
- // shutdown a broker to make sure the following topic deletion will be suspended
-// val leaderIdOpt = zkClient.getLeaderForPartition(topicPartition)
-// assertTrue(leaderIdOpt.isDefined, "Leader should exist for partition [test,0]")
- val follower = partitionHostingServers.filter(s => s.config.brokerId != leaderIdOpt).last
- follower.shutdown()
- // start topic deletion
- adminZkClient.deleteTopic(topic)
-
- // make sure deletion of all of the topic's replicas have been tried
- ensureControllerExists()
- val (controller, controllerId) = getController()
- val allReplicasForTopic = getAllReplicasFromAssignment(topic, expectedReplicaAssignment)
- TestUtils.waitUntilTrue(() => {
- val replicasInDeletionSuccessful = controller.kafkaController.controllerContext.replicasInState(topic, ReplicaDeletionSuccessful)
- val offlineReplicas = controller.kafkaController.controllerContext.replicasInState(topic, OfflineReplica)
- allReplicasForTopic == (replicasInDeletionSuccessful union offlineReplicas)
- }, s"Not all replicas for topic $topic are in states of either ReplicaDeletionSuccessful or OfflineReplica")
-
- // increase the partition count for topic
- val props = new Properties()
- props.setProperty(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, TestUtils.plaintextBootstrapServers(partitionHostingServers))
- Using(Admin.create(props)) { adminClient =>
- try {
- adminClient.createPartitions(Map(topic -> NewPartitions.increaseTo(2)).asJava).all().get()
- } catch {
- case _: ExecutionException =>
- }
- }
- // trigger a controller switch now
- val previousControllerId = controllerId
-
- controller.shutdown()
-
- ensureControllerExists()
- // wait until a new controller to show up
- TestUtils.waitUntilTrue(() => {
- val (_, newControllerId) = getController()
- newControllerId != previousControllerId
- }, "The new controller should not have the failed controller id")
-
- // bring back the failed brokers
- follower.startup()
- controller.startup()
- }
-
- TestUtils.verifyTopicDeletion(zkClientOrNull, topic, 2, partitionHostingBrokers)
- }
-
- @ParameterizedTest
- @ValueSource(strings = Array("zk", "kraft"))
- def testDeleteTopicDuringAddPartition(quorum: String): Unit = {
- brokers = createTestTopicAndCluster(topic)
- val leaderIdOpt = TestUtils.waitUntilLeaderIsKnown(brokers, new TopicPartition(topic, 0))
- val follower = brokers.filter(_.config.brokerId != leaderIdOpt).last
- val newPartition = new TopicPartition(topic, 1)
-
- if (isKRaftTest()) {
- follower.shutdown()
- // wait until the broker is in shutting down state
- TestUtils.waitUntilTrue(() => follower.brokerState == BrokerState.SHUTTING_DOWN,
- s"Follower ${follower.config.brokerId} was not shut down")
-
- increasePartitions(admin, topic, 3, brokers.filter(_.config.brokerId != follower.config.brokerId))
- } else {
- // capture the brokers before we shutdown so that we don't fail validation in `addPartitions`
- val brokersMetadata = adminZkClient.getBrokerMetadatas()
-
- follower.shutdown()
- // wait until the broker has been removed from ZK to reduce non-determinism
- TestUtils.waitUntilTrue(() => zkClient.getBroker(follower.config.brokerId).isEmpty,
- s"Follower ${follower.config.brokerId} was not removed from ZK")
-
- // add partitions to topic
- adminZkClient.addPartitions(topic, expectedReplicaFullAssignment, brokersMetadata, 2,
- Some(Map(1 -> Seq(0, 1, 2), 2 -> Seq(0, 1, 2))))
- }
-
- // start topic deletion
- admin.deleteTopics(Collections.singletonList(topic)).all().get()
- follower.startup()
- // test if topic deletion is resumed
- TestUtils.verifyTopicDeletion(zkClientOrNull, topic, 1, brokers)
- // verify that new partition doesn't exist on any broker either
- TestUtils.waitUntilTrue(() =>
- brokers.forall(_.logManager.getLog(newPartition).isEmpty),
- "Replica logs not for new partition [test,1] not deleted after delete topic is complete.")
- }
-
- @ParameterizedTest
- @ValueSource(strings = Array("kraft"))
- def testAddPartitionDuringDeleteTopic(quorum: String): Unit = {
- brokers = createTestTopicAndCluster(topic)
- // partitions to be added to the topic later
- val newPartition = new TopicPartition(topic, 1)
-
- if (isKRaftTest()) {
- admin.deleteTopics(Collections.singletonList(topic)).all().get()
- // pass empty list of brokers to avoid validating metadata since the topic is being deleted.
- increasePartitions(admin, topic, 3, Seq.empty)
- } else {
- zkClient.createTopLevelPaths()
- val brokersMetadata = adminZkClient.getBrokerMetadatas()
-
- // start topic deletion
- admin.deleteTopics(Collections.singletonList(topic)).all().get()
- adminZkClient.addPartitions(topic, expectedReplicaFullAssignment, brokersMetadata, 2,
- Some(Map(1 -> Seq(0, 1, 2), 2 -> Seq(0, 1, 2))))
- }
-
- TestUtils.verifyTopicDeletion(zkClientOrNull, topic, 1, brokers)
- // verify that new partition doesn't exist on any broker either
- assertTrue(brokers.forall(_.logManager.getLog(newPartition).isEmpty), "Replica logs not deleted after delete topic is complete")
- }
-
- @ParameterizedTest
- @ValueSource(strings = Array("zk", "kraft"))
- def testRecreateTopicAfterDeletion(quorum: String): Unit = {
- val expectedReplicaAssignment = Map(0 -> List(0, 1, 2))
- val topicPartition = new TopicPartition(topic, 0)
- brokers = createTestTopicAndCluster(topic)
- // start topic deletion
- admin.deleteTopics(Collections.singletonList(topic)).all().get()
- TestUtils.verifyTopicDeletion(zkClientOrNull, topic, 1, brokers)
- // re-create topic on same replicas
- TestUtils.createTopicWithAdmin(admin, topic, brokers, controllerServers, replicaAssignment = expectedReplicaAssignment)
- // check if all replica logs are created
- TestUtils.waitUntilTrue(() => brokers.forall(_.logManager.getLog(topicPartition).isDefined),
- "Replicas for topic test not created.")
- }
-
- @ParameterizedTest
- @ValueSource(strings = Array("zk", "kraft"))
- def testDeleteNonExistingTopic(quorum: String): Unit = {
- val topicPartition = new TopicPartition("test", 0)
- val topic = topicPartition.topic
- brokers = createTestTopicAndCluster(topic)
-
- // start topic deletion
- TestUtils.waitUntilTrue(() => {
- try {
- admin.deleteTopics(Collections.singletonList("test2")).all().get()
- false
- } catch {
- case e: ExecutionException =>
- classOf[UnknownTopicOrPartitionException].equals(e.getCause.getClass)
- }
- }, s"Topic test2 should not exist.")
-
- // verify delete topic path for test2 is removed from ZooKeeper
- TestUtils.verifyTopicDeletion(zkClientOrNull, "test2", 1, brokers)
- // verify that topic test is untouched
- TestUtils.waitUntilTrue(() => brokers.forall(_.logManager.getLog(topicPartition).isDefined),
- "Replicas for topic test not created")
-
- if (!isKRaftTest()) {
- // test the topic path exists
- assertTrue(zkClient.topicExists(topic), "Topic test mistakenly deleted")
- }
- TestUtils.waitUntilLeaderIsElectedOrChangedWithAdmin(admin, topic, 0, 1000)
- }
-
- @ParameterizedTest
- @ValueSource(strings = Array("zk", "kraft"))
- def testDeleteTopicWithCleaner(quorum: String): Unit = {
- val topicName = "test"
- val topicPartition = new TopicPartition(topicName, 0)
- val topic = topicPartition.topic
-
- val brokerConfigs = TestUtils.createBrokerConfigs(3, zkConnectOrNull, false)
- brokerConfigs.head.setProperty("delete.topic.enable", "true")
- brokerConfigs.head.setProperty("log.cleaner.enable","true")
- brokerConfigs.head.setProperty("log.cleanup.policy","compact")
- brokerConfigs.head.setProperty("log.segment.bytes","100")
- brokerConfigs.head.setProperty("log.cleaner.dedupe.buffer.size","1048577")
-
- brokers = createTestTopicAndCluster(topic, brokerConfigs, expectedReplicaAssignment)
-
- // for simplicity, we are validating cleaner offsets on a single broker
- val server = brokers.head
- val log = server.logManager.getLog(topicPartition).get
-
- // write to the topic to activate cleaner
- writeDups(numKeys = 100, numDups = 3,log)
-
- // wait for cleaner to clean
- server.logManager.cleaner.awaitCleaned(new TopicPartition(topicName, 0), 0)
-
- // delete topic
- admin.deleteTopics(Collections.singletonList(topic)).all().get()
- TestUtils.verifyTopicDeletion(zkClientOrNull, "test", 1, brokers)
- }
-
- @Test
- def testDeleteTopicAlreadyMarkedAsDeleted(): Unit = {
- val topicPartition = new TopicPartition("test", 0)
- val topic = topicPartition.topic
- brokers = createTestTopicAndCluster(topic)
- // start topic deletion
- admin.deleteTopics(Collections.singletonList(topic)).all().get()
- // try to delete topic marked as deleted
- // start topic deletion
- TestUtils.waitUntilTrue(() => {
- try {
- admin.deleteTopics(Collections.singletonList(topic)).all().get()
- false
- } catch {
- case e: ExecutionException =>
- classOf[UnknownTopicOrPartitionException].equals(e.getCause.getClass)
- }
- }, s"Topic ${topic} should be marked for deletion or already deleted.")
-
- TestUtils.verifyTopicDeletion(zkClient, topic, 1, brokers)
- }
-
- private def createTestTopicAndCluster(topic: String, numOfConfigs: Int = 3, deleteTopicEnabled: Boolean = true, replicaAssignment: Map[Int, List[Int]] = expectedReplicaAssignment): Seq[KafkaBroker] = {
- val brokerConfigs = TestUtils.createBrokerConfigs(numOfConfigs, zkConnectOrNull, enableControlledShutdown = false)
- brokerConfigs.foreach(_.setProperty("delete.topic.enable", deleteTopicEnabled.toString))
- createTestTopicAndCluster(topic, brokerConfigs, replicaAssignment)
- }
-
- private def createTestTopicAndCluster(topic: String, brokerConfigs: Seq[Properties], replicaAssignment: Map[Int, List[Int]]): Seq[KafkaBroker] = {
- val topicPartition = new TopicPartition(topic, 0)
- // create brokers
- val brokers = brokerConfigs.map(b => createBroker(KafkaConfig.fromProps(b)))
-
- admin = TestUtils.createAdminClient(brokers, ListenerName.forSecurityProtocol(SecurityProtocol.PLAINTEXT))
- TestUtils.createTopicWithAdmin(admin, topic, brokers, controllerServers, replicaAssignment = replicaAssignment)
-
- val brokersHostingTopicPartition = brokers.filter(s => expectedReplicaAssignment(0).contains(s.config.brokerId))
- // create the topic
- // wait until replica log is created on every broker
- TestUtils.waitUntilTrue(() => brokersHostingTopicPartition.forall(_.logManager.getLog(topicPartition).isDefined),
- "Replicas for topic test not created")
-
- brokers
- }
-
- private def writeDups(numKeys: Int, numDups: Int, log: UnifiedLog): Seq[(Int, Int)] = {
- var counter = 0
- for (_ <- 0 until numDups; key <- 0 until numKeys) yield {
- val count = counter
- log.appendAsLeader(TestUtils.singletonRecords(value = counter.toString.getBytes, key = key.toString.getBytes), leaderEpoch = 0)
- counter += 1
- (key, count)
- }
- }
-
- @ParameterizedTest
- @ValueSource(strings = Array("zk", "kraft"))
- def testDisableDeleteTopic(quorum: String): Unit = {
- val topicPartition = new TopicPartition(topic, 0)
-
- if (isKRaftTest()) {
- // Restart KRaft quorum with the updated config
- val overridingProps = new Properties()
- overridingProps.put(ServerConfigs.DELETE_TOPIC_ENABLE_CONFIG, false.toString)
- if (implementation != null)
- implementation.shutdown()
- implementation = newKRaftQuorum(new EmptyTestInfo(), overridingProps)
- }
-
- brokers = createTestTopicAndCluster(topic, deleteTopicEnabled = false)
- // in ZK mode, exception is returned when delete topic not enabled
- TestUtils.waitUntilTrue(() => {
- try {
- admin.deleteTopics(Collections.singletonList(topic)).all().get()
- false
- } catch {
- case e: ExecutionException =>
- classOf[TopicDeletionDisabledException].equals(e.getCause.getClass)
- }
- }, s"TopicDeletionDisabledException should be returned when deleting ${topic}.")
-
- if (!isKRaftTest()) {
- TestUtils.waitUntilTrue(() => !zkClient.isTopicMarkedForDeletion(topic),
- "Admin path /admin/delete_topics/%s path not deleted even if deleteTopic is disabled".format(topic))
- // test the topic path exists
- assertTrue(zkClient.topicExists(topic), "Topic path disappeared")
- }
- // verify that topic test is untouched
- assertTrue(brokers.forall(_.logManager.getLog(topicPartition).isDefined))
- assertDoesNotThrow(() => TestUtils.describeTopic(admin, topic))
- // topic test should have a leader
- TestUtils.waitUntilLeaderIsKnown(brokers, topicPartition)
- }
-
- // This test is not enabled with KRaft, it's not possible to partially delete topics with KRaft
- @Test
- def testDeletingPartiallyDeletedTopic(): Unit = {
- /**
- * A previous controller could have deleted some partitions of a topic from ZK, but not all partitions, and then crashed.
- * In that case, the new controller should be able to handle the partially deleted topic, and finish the deletion.
- */
-
- val replicaAssignment = Map(0 -> List(0, 1, 2), 1 -> List(0, 1, 2))
- brokers = createTestTopicAndCluster(topic, 4, deleteTopicEnabled = true, replicaAssignment)
-
- /**
- * shutdown all brokers in order to create a partially deleted topic on ZK
- */
- brokers.foreach(_.shutdown())
-
- /**
- * delete the partition znode at /brokers/topics/test/partition/0
- * to simulate the case that a previous controller crashed right after deleting the partition znode
- */
- zkClient.deleteRecursive(TopicPartitionZNode.path(new TopicPartition(topic, 0)))
- adminZkClient.deleteTopic(topic)
-
- /**
- * start up all brokers and verify that topic deletion eventually finishes.
- */
- brokers.foreach(_.startup())
- TestUtils.waitUntilTrue(() => brokers.exists(_.asInstanceOf[KafkaServer].kafkaController.isActive), "No controller is elected")
- TestUtils.verifyTopicDeletion(zkClient, topic, 2, brokers)
- }
-
- private def increasePartitions[B <: KafkaBroker](admin: Admin,
- topic: String,
- totalPartitionCount: Int,
- brokersToValidate: Seq[B]
- ): Unit = {
- val newPartitionSet: Map[String, NewPartitions] = Map.apply(topic -> NewPartitions.increaseTo(totalPartitionCount))
- admin.createPartitions(newPartitionSet.asJava)
-
- if (brokersToValidate.nonEmpty) {
- // wait until we've propagated all partitions metadata to all brokers
- val allPartitionsMetadata = waitForAllPartitionsMetadata(brokersToValidate, topic, totalPartitionCount)
- (0 until totalPartitionCount - 1).foreach(i => {
- allPartitionsMetadata.get(new TopicPartition(topic, i)).foreach { partitionMetadata =>
- assertEquals(totalPartitionCount, partitionMetadata.replicas.size)
- }
- })
- }
- }
-}
diff --git a/test-common/src/main/java/org/apache/kafka/common/test/TestUtils.java b/test-common/src/main/java/org/apache/kafka/common/test/TestUtils.java
index 151184596fe..b818a083c84 100644
--- a/test-common/src/main/java/org/apache/kafka/common/test/TestUtils.java
+++ b/test-common/src/main/java/org/apache/kafka/common/test/TestUtils.java
@@ -16,6 +16,10 @@
*/
package org.apache.kafka.common.test;
+import org.apache.kafka.clients.admin.Admin;
+import org.apache.kafka.clients.admin.TopicDescription;
+import org.apache.kafka.common.Node;
+import org.apache.kafka.common.TopicPartition;
import org.apache.kafka.common.errors.TimeoutException;
import org.apache.kafka.common.utils.Exit;
import org.apache.kafka.common.utils.Utils;
@@ -26,6 +30,9 @@ import org.slf4j.LoggerFactory;
import java.io.File;
import java.io.IOException;
import java.nio.file.Files;
+import java.util.Collections;
+import java.util.Optional;
+import java.util.function.BiFunction;
import java.util.function.Supplier;
import static java.lang.String.format;
@@ -125,4 +132,51 @@ public class TestUtils {
Thread.sleep(DEFAULT_POLL_INTERVAL_MS);
}
}
+
+ public static int waitUntilLeaderIsElectedOrChangedWithAdmin(Admin admin,
+ String topic,
+ int partitionNumber,
+ long timeoutMs) throws Exception {
+ BiFunction<String, Integer, Optional<Integer>> getPartitionLeader = (t, p) -> {
+ try {
+ return Optional.ofNullable(getLeaderFromAdmin(admin, t, p));
+ } catch (Exception e) {
+ throw new RuntimeException(e);
+ }
+ };
+ return doWaitUntilLeaderIsElectedOrChanged(getPartitionLeader, topic, partitionNumber, timeoutMs);
+ }
+
+ private static Integer getLeaderFromAdmin(Admin admin, String topic, int partition) throws Exception {
+ TopicDescription topicDescription = admin.describeTopics(Collections.singletonList(topic)).allTopicNames().get().get(topic);
+ return topicDescription.partitions().stream()
+ .filter(partitionInfo -> partitionInfo.partition() == partition)
+ .findFirst()
+ .map(partitionInfo -> partitionInfo.leader().id() == Node.noNode().id() ? null : partitionInfo.leader().id())
+ .orElse(null);
+ }
+
+ private static int doWaitUntilLeaderIsElectedOrChanged(BiFunction<String, Integer, Optional<Integer>> getPartitionLeader,
+ String topic,
+ int partition,
+ long timeoutMs) throws Exception {
+ long startTime = System.currentTimeMillis();
+ TopicPartition topicPartition = new TopicPartition(topic, partition);
+ Optional<Integer> electedLeader = Optional.empty();
+
+ while (electedLeader.isEmpty() && System.currentTimeMillis() < startTime + timeoutMs) {
+ Optional<Integer> leader = getPartitionLeader.apply(topic, partition);
+ if (leader.isPresent()) {
+ log.trace("Leader {} is elected for partition {}", leader.get(), topicPartition);
+ electedLeader = leader;
+ } else {
+ log.trace("Leader for partition {} is not elected yet", topicPartition);
+ }
+ Thread.sleep(Math.min(timeoutMs, 100L));
+ }
+
+ Optional<Integer> finalLeader = electedLeader;
+ return electedLeader.orElseThrow(() -> new AssertionError("Timing out after " + timeoutMs
+ + " ms since a leader was not elected for partition " + topicPartition + ", leader is " + finalLeader));
+ }
}
diff --git a/test-common/test-common-api/src/main/java/org/apache/kafka/common/test/api/ClusterInstance.java b/test-common/test-common-api/src/main/java/org/apache/kafka/common/test/api/ClusterInstance.java
index 81fc5965bcf..020f3ed7753 100644
--- a/test-common/test-common-api/src/main/java/org/apache/kafka/common/test/api/ClusterInstance.java
+++ b/test-common/test-common-api/src/main/java/org/apache/kafka/common/test/api/ClusterInstance.java
@@ -17,6 +17,7 @@
package org.apache.kafka.common.test.api;
+import kafka.log.UnifiedLog;
import kafka.network.SocketServer;
import kafka.server.BrokerServer;
import kafka.server.ControllerServer;
@@ -25,23 +26,32 @@ import kafka.server.KafkaBroker;
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.TopicPartition;
import org.apache.kafka.common.acl.AccessControlEntry;
import org.apache.kafka.common.acl.AclBindingFilter;
import org.apache.kafka.common.network.ListenerName;
import org.apache.kafka.common.test.TestUtils;
import org.apache.kafka.server.authorizer.Authorizer;
+import org.apache.kafka.storage.internals.checkpoint.OffsetCheckpointFile;
+import java.io.File;
+import java.io.IOException;
import java.util.ArrayList;
+import java.util.Arrays;
import java.util.Collection;
import java.util.Collections;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
+import java.util.Objects;
import java.util.Optional;
import java.util.Properties;
import java.util.Set;
import java.util.concurrent.atomic.AtomicReference;
import java.util.stream.Collectors;
+import java.util.stream.IntStream;
+
+import scala.jdk.javaapi.CollectionConverters;
import static org.apache.kafka.clients.consumer.GroupProtocol.CLASSIC;
import static org.apache.kafka.clients.consumer.GroupProtocol.CONSUMER;
@@ -188,8 +198,9 @@ public interface ClusterInstance {
default void waitForTopic(String topic, int partitions) throws InterruptedException {
// wait for metadata
+ Collection<KafkaBroker> brokers = aliveBrokers().values();
TestUtils.waitForCondition(
- () -> aliveBrokers().values().stream().allMatch(broker -> partitions == 0 ?
+ () -> brokers.stream().allMatch(broker -> partitions == 0 ?
broker.metadataCache().numPartitions(topic).isEmpty() :
broker.metadataCache().numPartitions(topic).contains(partitions)
), 60000L, topic + " metadata not propagated after 60000 ms");
@@ -197,9 +208,58 @@ public interface ClusterInstance {
for (ControllerServer controller : controllers().values()) {
long controllerOffset = controller.raftManager().replicatedLog().endOffset().offset() - 1;
TestUtils.waitForCondition(
- () -> aliveBrokers().values().stream().allMatch(broker -> ((BrokerServer) broker).sharedServer().loader().lastAppliedOffset() >= controllerOffset),
+ () -> brokers.stream().allMatch(broker -> ((BrokerServer) broker).sharedServer().loader().lastAppliedOffset() >= controllerOffset),
60000L, "Timeout waiting for controller metadata propagating to brokers");
}
+
+ if (partitions == 0) {
+ List<TopicPartition> topicPartitions = IntStream.range(0, 1)
+ .mapToObj(partition -> new TopicPartition(topic, partition))
+ .collect(Collectors.toList());
+
+ // Ensure that the topic-partition has been deleted from all brokers' replica managers
+ TestUtils.waitForCondition(() -> brokers.stream().allMatch(broker ->
+ topicPartitions.stream().allMatch(tp -> broker.replicaManager().onlinePartition(tp).isEmpty())),
+ "Replica manager's should have deleted all of this topic's partitions");
+
+ // Ensure that logs from all replicas are deleted
+ TestUtils.waitForCondition(() -> brokers.stream().allMatch(broker ->
+ topicPartitions.stream().allMatch(tp -> broker.logManager().getLog(tp, false).isEmpty())),
+ "Replica logs not deleted after delete topic is complete");
+
+ // Ensure that the topic is removed from all cleaner offsets
+ TestUtils.waitForCondition(() -> brokers.stream().allMatch(broker ->
+ topicPartitions.stream().allMatch(tp -> {
+ List<File> liveLogDirs = CollectionConverters.asJava(broker.logManager().liveLogDirs());
+ return liveLogDirs.stream().allMatch(logDir -> {
+ OffsetCheckpointFile checkpointFile;
+ try {
+ checkpointFile = new OffsetCheckpointFile(new File(logDir, "cleaner-offset-checkpoint"), null);
+ } catch (IOException e) {
+ throw new RuntimeException(e);
+ }
+ return !checkpointFile.read().containsKey(tp);
+ });
+ })),
+ "Cleaner offset for deleted partition should have been removed");
+
+ // Ensure that the topic directories are soft-deleted
+ TestUtils.waitForCondition(() -> brokers.stream().allMatch(broker ->
+ CollectionConverters.asJava(broker.config().logDirs()).stream().allMatch(logDir ->
+ topicPartitions.stream().noneMatch(tp ->
+ new File(logDir, tp.topic() + "-" + tp.partition()).exists()))),
+ "Failed to soft-delete the data to a delete directory");
+
+ // Ensure that the topic directories are hard-deleted
+ TestUtils.waitForCondition(() -> brokers.stream().allMatch(broker ->
+ CollectionConverters.asJava(broker.config().logDirs()).stream().allMatch(logDir ->
+ topicPartitions.stream().allMatch(tp ->
+ Arrays.stream(Objects.requireNonNull(new File(logDir).list())).noneMatch(partitionDirectoryName ->
+ partitionDirectoryName.startsWith(tp.topic() + "-" + tp.partition()) &&
+ partitionDirectoryName.endsWith(UnifiedLog.DeleteDirSuffix())))
+ )
+ ), "Failed to hard-delete the delete directory");
+ }
}
default List<Authorizer> authorizers() {
@@ -224,5 +284,4 @@ public interface ClusterInstance {
}, "expected acls: " + entries + ", actual acls: " + actualEntries.get());
}
}
-
}
diff --git a/test-common/test-common-api/src/test/java/org/apache/kafka/common/test/api/ClusterTestExtensionsTest.java b/test-common/test-common-api/src/test/java/org/apache/kafka/common/test/api/ClusterTestExtensionsTest.java
index 208188dc4cf..8e02aa7bd98 100644
--- a/test-common/test-common-api/src/test/java/org/apache/kafka/common/test/api/ClusterTestExtensionsTest.java
+++ b/test-common/test-common-api/src/test/java/org/apache/kafka/common/test/api/ClusterTestExtensionsTest.java
@@ -224,7 +224,14 @@ public class ClusterTestExtensionsTest {
}
- @ClusterTest(types = {Type.CO_KRAFT, Type.KRAFT}, brokers = 4)
+ @ClusterTest(
+ types = {Type.CO_KRAFT, Type.KRAFT},
+ brokers = 4,
+ serverProperties = {
+ @ClusterConfigProperty(key = "log.initial.task.delay.ms", value = "100"),
+ @ClusterConfigProperty(key = "log.segment.delete.delay.ms", value = "1000")
+ }
+ )
public void testVerifyTopicDeletion(ClusterInstance clusterInstance) throws Exception {
try (Admin admin = clusterInstance.createAdminClient()) {
String testTopic = "testTopic";
diff --git a/tools/src/test/java/org/apache/kafka/tools/TopicCommandTest.java b/tools/src/test/java/org/apache/kafka/tools/TopicCommandTest.java
index d554d7e4162..b696ae613a2 100644
--- a/tools/src/test/java/org/apache/kafka/tools/TopicCommandTest.java
+++ b/tools/src/test/java/org/apache/kafka/tools/TopicCommandTest.java
@@ -340,6 +340,8 @@ public class TopicCommandTest {
static List<ClusterConfig> generate() {
Map<String, String> serverProp = new HashMap<>();
serverProp.put(REPLICA_FETCH_MAX_BYTES_CONFIG, "1"); // if config name error, no exception throw
+ serverProp.put("log.initial.task.delay.ms", "100");
+ serverProp.put("log.segment.delete.delay.ms", "1000");
Map<Integer, Map<String, String>> rackInfo = new HashMap<>();
Map<String, String> infoPerBroker1 = new HashMap<>();