You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@kafka.apache.org by so...@apache.org on 2024/09/06 20:57:20 UTC
(kafka) branch trunk updated: KAFKA-15793 Fix ZkMigrationIntegrationTest#testMigrateTopicDeletions (#17004)
This is an automated email from the ASF dual-hosted git repository.
soarez 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 e4d108d4df3 KAFKA-15793 Fix ZkMigrationIntegrationTest#testMigrateTopicDeletions (#17004)
e4d108d4df3 is described below
commit e4d108d4df3762843deb568498d5b29c93488e17
Author: David Arthur <mu...@gmail.com>
AuthorDate: Fri Sep 6 16:57:13 2024 -0400
KAFKA-15793 Fix ZkMigrationIntegrationTest#testMigrateTopicDeletions (#17004)
Reviewers: Igor Soarez <so...@apple.com>, Ajit Singh <>
---
.github/workflows/deflake.yml | 2 +-
core/src/test/resources/log4j.properties | 4 +-
.../kafka/zk/ZkMigrationIntegrationTest.scala | 62 ++++++++++++++--------
3 files changed, 43 insertions(+), 25 deletions(-)
diff --git a/.github/workflows/deflake.yml b/.github/workflows/deflake.yml
index 8cb89e8be8a..a87b9fe3ed8 100644
--- a/.github/workflows/deflake.yml
+++ b/.github/workflows/deflake.yml
@@ -54,7 +54,7 @@ jobs:
- name: Test
timeout-minutes: 60
run: |
- ./gradlew --build-cache --scan --continue \
+ ./gradlew --info --build-cache --scan --continue \
-PtestLoggingEvents=started,passed,skipped,failed \
-PignoreFailures=true -PmaxParallelForks=2 \
-Pkafka.cluster.test.repeat=${{ inputs.test-repeat }} \
diff --git a/core/src/test/resources/log4j.properties b/core/src/test/resources/log4j.properties
index b265ee9cdaa..4c238cdc59c 100644
--- a/core/src/test/resources/log4j.properties
+++ b/core/src/test/resources/log4j.properties
@@ -18,8 +18,8 @@ log4j.appender.stdout=org.apache.log4j.ConsoleAppender
log4j.appender.stdout.layout=org.apache.log4j.PatternLayout
log4j.appender.stdout.layout.ConversionPattern=[%d] %p %m (%c:%L)%n
-log4j.logger.kafka=WARN
-log4j.logger.org.apache.kafka=WARN
+log4j.logger.kafka=INFO
+log4j.logger.org.apache.kafka=INFO
# zkclient can be verbose, during debugging it is common to adjust it separately
log4j.logger.org.apache.zookeeper=WARN
diff --git a/core/src/test/scala/integration/kafka/zk/ZkMigrationIntegrationTest.scala b/core/src/test/scala/integration/kafka/zk/ZkMigrationIntegrationTest.scala
index cbea0d4ac71..e8946bbd6f3 100644
--- a/core/src/test/scala/integration/kafka/zk/ZkMigrationIntegrationTest.scala
+++ b/core/src/test/scala/integration/kafka/zk/ZkMigrationIntegrationTest.scala
@@ -65,7 +65,6 @@ import scala.jdk.CollectionConverters._
object ZkMigrationIntegrationTest {
def zkClustersForAllMigrationVersions(): java.util.List[ClusterConfig] = {
Seq(
- MetadataVersion.IBP_3_4_IV0,
MetadataVersion.IBP_3_5_IV2,
MetadataVersion.IBP_3_6_IV2,
MetadataVersion.IBP_3_7_IV0,
@@ -92,7 +91,7 @@ object ZkMigrationIntegrationTest {
}
@ExtendWith(value = Array(classOf[ClusterTestExtensions]))
-@Timeout(300)
+@Timeout(600)
class ZkMigrationIntegrationTest {
val log: Logger = LoggerFactory.getLogger(classOf[ZkMigrationIntegrationTest])
@@ -296,13 +295,29 @@ class ZkMigrationIntegrationTest {
def testMigrateTopicDeletions(zkCluster: ClusterInstance): Unit = {
// Create some topics in ZK mode
var admin = zkCluster.createAdminClient()
- val newTopics = new util.ArrayList[NewTopic]()
- newTopics.add(new NewTopic("test-topic-1", 10, 3.toShort))
- newTopics.add(new NewTopic("test-topic-2", 10, 3.toShort))
- newTopics.add(new NewTopic("test-topic-3", 10, 3.toShort))
- val createTopicResult = admin.createTopics(newTopics)
- createTopicResult.all().get(300, TimeUnit.SECONDS)
- admin.close()
+ try {
+ val newTopics = new util.ArrayList[NewTopic]()
+ newTopics.add(new NewTopic("test-topic-1", 10, 3.toShort))
+ newTopics.add(new NewTopic("test-topic-2", 10, 3.toShort))
+ newTopics.add(new NewTopic("test-topic-3", 10, 3.toShort))
+ val createTopicResult = admin.createTopics(newTopics)
+ createTopicResult.all().get(61, TimeUnit.SECONDS)
+ TestUtils.waitUntilTrue(() => {
+ val topicDescribe = admin.describeTopics(Seq("test-topic-1", "test-topic-2", "test-topic-3").asJava)
+ if (topicDescribe.topicNameValues() == null || topicDescribe.topicNameValues().size() < 3) {
+ false
+ } else {
+ topicDescribe.topicNameValues().values().stream().allMatch {
+ topic => topic.get(62, TimeUnit.SECONDS).partitions().stream().allMatch(part => {
+ part.leader() != null && part.isr().size() == 3
+ })
+ }
+ }
+ }, msg="waiting for topics to be available", waitTimeMs=303000)
+ } finally {
+ admin.close()
+ }
+
val zkClient = zkCluster.asInstanceOf[ZkClusterInstance].getUnderlying().zkClient
// Bootstrap the ZK cluster ID into KRaft
@@ -339,7 +354,7 @@ class ZkMigrationIntegrationTest {
zkClient.createDeleteTopicPath("test-topic-3")
zkCluster.waitForReadyBrokers()
- readyFuture.get(60, TimeUnit.SECONDS)
+ readyFuture.get(64, TimeUnit.SECONDS)
// Only continue with the test if there are some pending deletions to verify. If there are not any pending
// deletions, this will mark the test as "skipped" instead of failed.
@@ -352,12 +367,13 @@ class ZkMigrationIntegrationTest {
TestUtils.waitUntilTrue(
() => zkClient.getOrCreateMigrationState(ZkMigrationLeadershipState.EMPTY).initialZkMigrationComplete(),
"Timed out waiting for migration to complete",
- 30000)
+ 65000)
// At this point, some of the topics may have been deleted by ZK controller and the rest will be
// implicitly deleted by the KRaft controller and remove from the ZK brokers as stray partitions
def topicsAllDeleted(admin: Admin): Boolean = {
- val topics = admin.listTopics().names().get(60, TimeUnit.SECONDS)
+ val topics = admin.listTopics().names().get(66, TimeUnit.SECONDS)
+ log.info("Topics are {}", topics)
topics.retainAll(util.Arrays.asList(
"test-topic-1", "test-topic-2", "test-topic-3"
))
@@ -369,18 +385,19 @@ class ZkMigrationIntegrationTest {
TestUtils.waitUntilTrue(
() => topicsAllDeleted(admin),
"Timed out waiting for topics to be deleted",
- 30000,
+ 307000,
1000)
+ log.info("Topics were deleted. Now re-creating them.")
val newTopics = new util.ArrayList[NewTopic]()
newTopics.add(new NewTopic("test-topic-1", 2, 3.toShort))
newTopics.add(new NewTopic("test-topic-2", 1, 3.toShort))
newTopics.add(new NewTopic("test-topic-3", 10, 3.toShort))
val createTopicResult = admin.createTopics(newTopics)
- createTopicResult.all().get(60, TimeUnit.SECONDS)
+ createTopicResult.all().get(68, TimeUnit.SECONDS)
def topicsAllRecreated(admin: Admin): Boolean = {
- val topics = admin.listTopics().names().get(60, TimeUnit.SECONDS)
+ val topics = admin.listTopics().names().get(69, TimeUnit.SECONDS)
topics.retainAll(util.Arrays.asList(
"test-topic-1", "test-topic-2", "test-topic-3"
))
@@ -391,21 +408,22 @@ class ZkMigrationIntegrationTest {
TestUtils.waitUntilTrue(
() => topicsAllRecreated(admin),
"Timed out waiting for topics to be created",
- 30000,
+ 70000,
1000)
-
- TestUtils.retry(300000) {
+ log.info("Topics were re-created. Now waiting for consistent topic state.")
+ TestUtils.retry(311000) {
// Need a retry here since topic metadata may be inconsistent between brokers
val topicDescriptions = try {
admin.describeTopics(util.Arrays.asList(
"test-topic-1", "test-topic-2", "test-topic-3"
)).topicNameValues().asScala.map { case (name, description) =>
- name -> description.get(60, TimeUnit.SECONDS)
+ name -> description.get(72, TimeUnit.SECONDS)
}.toMap
} catch {
case e: ExecutionException if e.getCause.isInstanceOf[UnknownTopicOrPartitionException] => Map.empty[String, TopicDescription]
case t: Throwable => fail("Error describing topics", t.getCause)
}
+ log.debug("Topic describe: {}", topicDescriptions);
assertEquals(2, topicDescriptions("test-topic-1").partitions().size())
assertEquals(1, topicDescriptions("test-topic-2").partitions().size())
@@ -417,14 +435,13 @@ class ZkMigrationIntegrationTest {
})
}
- val absentTopics = admin.listTopics().names().get(60, TimeUnit.SECONDS).asScala
+ val absentTopics = admin.listTopics().names().get(73, TimeUnit.SECONDS).asScala
assertTrue(absentTopics.contains("test-topic-1"))
assertTrue(absentTopics.contains("test-topic-2"))
assertTrue(absentTopics.contains("test-topic-3"))
}
-
- admin.close()
} finally {
+ admin.close()
shutdownInSequence(zkCluster, kraftCluster)
}
}
@@ -1153,6 +1170,7 @@ class ZkMigrationIntegrationTest {
def shutdownInSequence(zkCluster: ClusterInstance, kraftCluster: KafkaClusterTestKit): Unit = {
zkCluster.brokerIds().forEach(zkCluster.shutdownBroker(_))
+ kraftCluster.nonFatalFaultHandler().setIgnore(true)
kraftCluster.close()
zkCluster.stop()
}