You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@kafka.apache.org by de...@apache.org on 2024/01/12 03:50:18 UTC
(kafka) branch trunk updated: MINOR: Enable kraft test in kafka.api (#14595)
This is an automated email from the ASF dual-hosted git repository.
dengziming 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 da6f05258fc MINOR: Enable kraft test in kafka.api (#14595)
da6f05258fc is described below
commit da6f05258fcd727e20f2beaaf136b799129ec3ef
Author: dengziming <de...@gmail.com>
AuthorDate: Fri Jan 12 11:50:12 2024 +0800
MINOR: Enable kraft test in kafka.api (#14595)
Reviewers: Mickael Maison <mi...@gmail.com>
---
.../kafka/api/ProducerFailureHandlingTest.scala | 6 +-
.../kafka/api/ProducerSendWhileDeletionTest.scala | 140 +++++++++++----------
.../kafka/api/TransactionsBounceTest.scala | 24 ++--
3 files changed, 88 insertions(+), 82 deletions(-)
diff --git a/core/src/test/scala/integration/kafka/api/ProducerFailureHandlingTest.scala b/core/src/test/scala/integration/kafka/api/ProducerFailureHandlingTest.scala
index 0b1628a762e..4fe7f5e87f9 100644
--- a/core/src/test/scala/integration/kafka/api/ProducerFailureHandlingTest.scala
+++ b/core/src/test/scala/integration/kafka/api/ProducerFailureHandlingTest.scala
@@ -182,9 +182,9 @@ class ProducerFailureHandlingTest extends KafkaServerTestHarness {
}
/**
- * Send with invalid partition id should return ExecutionException caused by TimeoutException
- * when partition is higher than the upper bound of partitions.
- */
+ * Send with invalid partition id should return ExecutionException caused by TimeoutException
+ * when partition is higher than the upper bound of partitions.
+ */
@ParameterizedTest(name = TestInfoUtils.TestWithParameterizedQuorumName)
@ValueSource(strings = Array("zk", "kraft"))
def testInvalidPartition(quorum: String): Unit = {
diff --git a/core/src/test/scala/integration/kafka/api/ProducerSendWhileDeletionTest.scala b/core/src/test/scala/integration/kafka/api/ProducerSendWhileDeletionTest.scala
index 9d9c5ab8081..0ad6384b115 100644
--- a/core/src/test/scala/integration/kafka/api/ProducerSendWhileDeletionTest.scala
+++ b/core/src/test/scala/integration/kafka/api/ProducerSendWhileDeletionTest.scala
@@ -1,83 +1,87 @@
/**
- * 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.
- */
+ * 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.api
import kafka.server.KafkaConfig
-import kafka.utils.TestUtils
+import kafka.utils.{TestInfoUtils, TestUtils}
+import org.apache.kafka.clients.admin.NewPartitionReassignment
import org.apache.kafka.clients.producer.{ProducerConfig, ProducerRecord}
-
import org.apache.kafka.common.TopicPartition
import org.junit.jupiter.api.Assertions.assertEquals
-import org.junit.jupiter.api.Test
+import org.junit.jupiter.params.ParameterizedTest
+import org.junit.jupiter.params.provider.ValueSource
import java.nio.charset.StandardCharsets
+import java.util
+import java.util.Optional
+import scala.jdk.CollectionConverters._
class ProducerSendWhileDeletionTest extends IntegrationTestHarness {
- val producerCount: Int = 1
- val brokerCount: Int = 2
-
- serverConfig.put(KafkaConfig.NumPartitionsProp, 2.toString)
- serverConfig.put(KafkaConfig.DefaultReplicationFactorProp, 2.toString)
- serverConfig.put(KafkaConfig.AutoLeaderRebalanceEnableProp, false.toString)
-
- producerConfig.put(ProducerConfig.MAX_BLOCK_MS_CONFIG, 5000L.toString)
- producerConfig.put(ProducerConfig.REQUEST_TIMEOUT_MS_CONFIG, 10000.toString)
- producerConfig.put(ProducerConfig.DELIVERY_TIMEOUT_MS_CONFIG, 10000.toString)
-
- /**
- * Tests that Producer gets self-recovered when a topic is deleted mid-way of produce.
- *
- * Producer will attempt to send messages to the partition specified in each record, and should
- * succeed as long as the partition is included in the metadata.
- */
- @Test
- def testSendWithTopicDeletionMidWay(): Unit = {
- val numRecords = 10
- val topic = "topic"
-
- // Create topic with leader as 0 for the 2 partitions.
- createTopicWithAssignment(topic, Map(0 -> Seq(0, 1), 1 -> Seq(0, 1)))
-
- val reassignment = Map(
- new TopicPartition(topic, 0) -> Seq(1, 0),
- new TopicPartition(topic, 1) -> Seq(1, 0)
- )
-
- // Change leader to 1 for both the partitions to increase leader epoch from 0 -> 1
- zkClient.createPartitionReassignment(reassignment)
- TestUtils.waitUntilTrue(() => !zkClient.reassignPartitionsInProgress,
- "failed to remove reassign partitions path after completion")
-
- val producer = createProducer()
-
- (1 to numRecords).foreach { i =>
- val resp = producer.send(new ProducerRecord(topic, null, ("value" + i).getBytes(StandardCharsets.UTF_8))).get
- assertEquals(topic, resp.topic())
- }
-
- // Start topic deletion
- adminZkClient.deleteTopic(topic)
-
- // Verify that the topic is deleted when no metadata request comes in
- TestUtils.verifyTopicDeletion(zkClient, topic, 2, servers)
-
- // Producer should be able to send messages even after topic gets deleted and auto-created
- assertEquals(topic, producer.send(new ProducerRecord(topic, null, "value".getBytes(StandardCharsets.UTF_8))).get.topic())
+ val producerCount: Int = 1
+ val brokerCount: Int = 2
+
+ serverConfig.put(KafkaConfig.NumPartitionsProp, 2.toString)
+ serverConfig.put(KafkaConfig.DefaultReplicationFactorProp, 2.toString)
+ serverConfig.put(KafkaConfig.AutoLeaderRebalanceEnableProp, false.toString)
+
+ producerConfig.put(ProducerConfig.MAX_BLOCK_MS_CONFIG, 5000L.toString)
+ producerConfig.put(ProducerConfig.REQUEST_TIMEOUT_MS_CONFIG, 10000.toString)
+ producerConfig.put(ProducerConfig.DELIVERY_TIMEOUT_MS_CONFIG, 10000.toString)
+
+ /**
+ * Tests that Producer gets self-recovered when a topic is deleted mid-way of produce.
+ *
+ * Producer will attempt to send messages to the partition specified in each record, and should
+ * succeed as long as the partition is included in the metadata.
+ */
+ @ParameterizedTest(name = TestInfoUtils.TestWithParameterizedQuorumName)
+ @ValueSource(strings = Array("zk", "kraft"))
+ def testSendWithTopicDeletionMidWay(quorum: String): Unit = {
+ val numRecords = 10
+ val topic = "topic"
+
+ // Create topic with leader as 0 for the 2 partitions.
+ createTopicWithAssignment(topic, Map(0 -> Seq(0, 1), 1 -> Seq(0, 1)))
+
+ val reassignment = Map(
+ new TopicPartition(topic, 0) -> Optional.of(new NewPartitionReassignment(util.Arrays.asList(1, 0))),
+ new TopicPartition(topic, 1) -> Optional.of(new NewPartitionReassignment(util.Arrays.asList(1, 0)))
+ )
+
+ // Change leader to 1 for both the partitions to increase leader epoch from 0 -> 1
+ val admin = createAdminClient()
+ admin.alterPartitionReassignments(reassignment.asJava).all().get()
+
+ val producer = createProducer()
+
+ (1 to numRecords).foreach { i =>
+ val resp = producer.send(new ProducerRecord(topic, null, ("value" + i).getBytes(StandardCharsets.UTF_8))).get
+ assertEquals(topic, resp.topic())
}
+ // Start topic deletion
+ deleteTopic(topic, listenerName)
+
+ // Verify that the topic is deleted when no metadata request comes in
+ TestUtils.verifyTopicDeletion(zkClientOrNull, topic, 2, brokers)
+
+ // Producer should be able to send messages even after topic gets deleted and auto-created
+ assertEquals(topic, producer.send(new ProducerRecord(topic, null, "value".getBytes(StandardCharsets.UTF_8))).get.topic())
+ }
+
}
diff --git a/core/src/test/scala/integration/kafka/api/TransactionsBounceTest.scala b/core/src/test/scala/integration/kafka/api/TransactionsBounceTest.scala
index 121e3d06d61..f236c935851 100644
--- a/core/src/test/scala/integration/kafka/api/TransactionsBounceTest.scala
+++ b/core/src/test/scala/integration/kafka/api/TransactionsBounceTest.scala
@@ -18,16 +18,16 @@
package kafka.api
import java.util.Properties
-
import kafka.server.KafkaConfig
-import kafka.utils.TestUtils
+import kafka.utils.{TestInfoUtils, TestUtils}
import org.apache.kafka.clients.consumer.{Consumer, ConsumerConfig}
import org.apache.kafka.clients.producer.{KafkaProducer, ProducerConfig}
import org.apache.kafka.clients.producer.internals.ErrorLoggingCallback
import org.apache.kafka.common.TopicPartition
import org.apache.kafka.server.util.ShutdownableThread
import org.junit.jupiter.api.Assertions._
-import org.junit.jupiter.api.Test
+import org.junit.jupiter.params.ParameterizedTest
+import org.junit.jupiter.params.provider.ValueSource
import scala.annotation.nowarn
import scala.jdk.CollectionConverters._
@@ -67,21 +67,23 @@ class TransactionsBounceTest extends IntegrationTestHarness {
// Since such quick rotation of servers is incredibly unrealistic, we allow this one test to preallocate ports, leaving
// a small risk of hitting errors due to port conflicts. Hopefully this is infrequent enough to not cause problems.
override def generateConfigs = {
- FixedPortTestUtils.createBrokerConfigs(brokerCount, zkConnect, enableControlledShutdown = true)
+ FixedPortTestUtils.createBrokerConfigs(brokerCount, zkConnectOrNull, enableControlledShutdown = true)
.map(KafkaConfig.fromProps(_, overridingProps))
}
override protected def brokerCount: Int = 4
@nowarn("cat=deprecation")
- @Test
- def testWithGroupId(): Unit = {
+ @ParameterizedTest(name = TestInfoUtils.TestWithParameterizedQuorumName)
+ @ValueSource(strings = Array("zk", "kraft"))
+ def testWithGroupId(quorum: String): Unit = {
testBrokerFailure((producer, groupId, consumer) =>
producer.sendOffsetsToTransaction(TestUtils.consumerPositions(consumer).asJava, groupId))
}
- @Test
- def testWithGroupMetadata(): Unit = {
+ @ParameterizedTest(name = TestInfoUtils.TestWithParameterizedQuorumName)
+ @ValueSource(strings = Array("zk", "kraft"))
+ def testWithGroupMetadata(quorum: String): Unit = {
testBrokerFailure((producer, _, consumer) =>
producer.sendOffsetsToTransaction(TestUtils.consumerPositions(consumer).asJava, consumer.groupMetadata()))
}
@@ -94,7 +96,7 @@ class TransactionsBounceTest extends IntegrationTestHarness {
val numInputRecords = 10000
createTopics()
- TestUtils.seedTopicWithNumberedRecords(inputTopic, numInputRecords, servers)
+ TestUtils.seedTopicWithNumberedRecords(inputTopic, numInputRecords, brokers)
val consumer = createConsumerAndSubscribe(consumerGroup, List(inputTopic))
val producer = createTransactionalProducer("test-txn")
@@ -189,7 +191,7 @@ class TransactionsBounceTest extends IntegrationTestHarness {
private class BounceScheduler extends ShutdownableThread("daemon-broker-bouncer", false) {
override def doWork(): Unit = {
- for (server <- servers) {
+ for (server <- brokers) {
trace("Shutting down server : %s".format(server.config.brokerId))
server.shutdown()
server.awaitShutdown()
@@ -200,7 +202,7 @@ class TransactionsBounceTest extends IntegrationTestHarness {
Thread.sleep(500)
}
- (0 until numPartitions).foreach(partition => TestUtils.waitUntilLeaderIsElectedOrChanged(zkClient, outputTopic, partition))
+ (0 until numPartitions).foreach(partition => TestUtils.waitUntilLeaderIsElectedOrChangedWithAdmin(createAdminClient(), outputTopic, partition))
}
override def shutdown(): Unit = {