root
/
kafka
mirror of https://github.com/apache/kafka.git
1
0
Fork 0
Code Issues Actions 2 Packages Projects Releases Wiki Activity

KAFKA-18123 Fix flaky DynamicBrokerReconfigurationTest#testThreadPoolResize (#17986) 

Reviewers: Chia-Ping Tsai <chia7712@gmail.com>
This commit is contained in:
PoAn Yang 2024-12-01 20:58:02 +08:00 committed by GitHub
parent 615f1a0bf9
commit 60ce83b2c6
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
1 changed files with 11 additions and 0 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

11
core/src/test/scala/integration/kafka/server/DynamicBrokerReconfigurationTest.scala
View File

@ -760,6 +760,17 @@ class DynamicBrokerReconfigurationTest extends QuorumTestHarness with SaslSetup
@ParameterizedTest(name = TestInfoUtils.TestWithParameterizedQuorumAndGroupProtocolNames) @ParameterizedTest(name = TestInfoUtils.TestWithParameterizedQuorumAndGroupProtocolNames)
@MethodSource(Array("getTestQuorumAndGroupProtocolParametersAll")) @MethodSource(Array("getTestQuorumAndGroupProtocolParametersAll"))
def testThreadPoolResize(quorum: String, groupProtocol: String): Unit = { def testThreadPoolResize(quorum: String, groupProtocol: String): Unit = {
// In kraft mode, the StripedReplicaPlacer#initialize includes some randomization,
// so the replica assignment is not deterministic.
// If a fetcher thread is not assigned any topic partition, it will not be created.
// Change the replica assignment to ensure that all fetcher threads are created.
TestUtils.deleteTopicWithAdmin(adminClients.head, topic, servers, controllerServers)
val replicaAssignment = Map(
0 -> Seq(0, 1, 2), 1 -> Seq(1, 2, 0), 2 -> Seq(2, 1, 0), 3 -> Seq(0, 1, 2), 4 -> Seq(1, 2, 0),
5 -> Seq(2, 1, 0), 6 -> Seq(0, 1, 2), 7 -> Seq(1, 2, 0), 8 -> Seq(2, 1, 0), 9 -> Seq(0, 1, 2))
TestUtils.createTopicWithAdmin(adminClients.head, topic, servers, controllerServers, replicaAssignment = replicaAssignment)
val requestHandlerPrefix = "data-plane-kafka-request-handler-" val requestHandlerPrefix = "data-plane-kafka-request-handler-"
val networkThreadPrefix = "data-plane-kafka-network-thread-" val networkThreadPrefix = "data-plane-kafka-network-thread-"
val fetcherThreadPrefix = "ReplicaFetcherThread-" val fetcherThreadPrefix = "ReplicaFetcherThread-"