KAFKA-17986: Fix ConsumerRebootstrapTest and ProducerRebootstrapTest #18175

peterxcli · 2024-12-13T19:31:18Z

ProducerRebootstrapTest has many issues in using kraft.

it passes unclean.leader.election.enable as server config but it is not configured on controller node. We should append the config to topic in creating topic.
it assumes topic is distributed on [0, 1]. If the order is not correct, the test can't pass. We should create the topic manually to ensure the distribution.
in kraft mode, the time to trigger unclean election is 5 mins. it is too long to wait. We should configure controller to reduce the number.

More detailed description of your change,
if necessary. The PR title and PR message become
the squashed commit message, so use a separate
comment to ping reviewers.

Summary of testing strategy (including rationale)
for the feature or bug fix. Unit and/or integration
tests are expected for any behaviour change and
system tests should be considered for larger changes.

Committer Checklist (excluded from commit message)

Verify design and implementation
Verify test coverage and CI build status
Verify documentation (including upgrade notes)

- Add unclean.leader.election.enable=true to topic configuration when creating test topic to ensure controller node has the correct setting - Set unclean.leader.election.interval.ms to 1ms (down from default 5 minutes) to speed up test execution in KRaft mode

chia7712 · 2024-12-14T04:16:04Z

@peterxcli I update the topic to match the changes. it is ok to file two PRs :)

TaiJuWu · 2024-12-15T14:46:50Z

core/src/test/java/kafka/server/ProducerRebootstrapIntegrationTest.java

+            Properties topicProps = new Properties();
+            topicProps.put(TopicConfig.UNCLEAN_LEADER_ELECTION_ENABLE_CONFIG, "true");
+
+            admin.createTopics(List.of(new NewTopic(topicName, 2, (short) 2).configs(Utils.propsToStringMap(topicProps)))).all().get();


Should we need cluster.waitForTopic(topicName, 2); here?

I think that's only necessary when the topic metadata has changed?

Yes, but create topic will change metadata, right?

Oh thanks, you're right.
But seems the old test infra doesn't wait for the metadata propagation explicitly.
So I think it's ok the skip that.

kafka/core/src/test/scala/unit/kafka/utils/TestUtils.scala

Lines 380 to 406 in 772aa24

def createTopicWithAdminRaw[B <: KafkaBroker](

admin: Admin,

topic: String,

numPartitions: Int = 1,

replicationFactor: Int = 1,

replicaAssignment: collection.Map[Int, Seq[Int]] = Map.empty,

topicConfig: Properties = new Properties,

): Uuid = {

val configsMap = new util.HashMap[String, String]()

topicConfig.forEach((k, v) => configsMap.put(k.toString, v.toString))

val result = if (replicaAssignment.isEmpty) {

admin.createTopics(Collections.singletonList(new NewTopic(

topic, numPartitions, replicationFactor.toShort).configs(configsMap)))

} else {

val assignment = new util.HashMap[Integer, util.List[Integer]]()

replicaAssignment.foreachEntry { case (k, v) =>

val replicas = new util.ArrayList[Integer]

v.foreach(r => replicas.add(r.asInstanceOf[Integer]))

assignment.put(k.asInstanceOf[Integer], replicas)

}

admin.createTopics(Collections.singletonList(new NewTopic(

topic, assignment).configs(configsMap)))

}

result.topicId(topic).get()

}

Add ability to configure fixed broker ports in test framework instead of using ephemeral ports. This helps with: - Debugging by making broker ports predictable and consistent - Avoiding port conflicts in certain test scenarios - Better control over broker port assignment Changes include: - Add fixedBrokerPorts flag to TestKitNodes and ClusterConfig - Add port selection logic in KafkaClusterTestKit - Add choosePort() utility method to TestUtils - Update ClusterTest annotation to support fixed ports configuration The change is backward compatible as fixedBrokerPorts defaults to false, maintaining existing behavior.

This reverts commit 33e7c12.

This reverts commit 5580f3e.

@disabled

- Remove @disabled annotation from ConsumerRebootstrapTest - Add clarifying comment about leader shutdown in ProducerRebootstrapTest - Improve documentation in RebootstrapTest regarding unclean leader election - Clean up comment formatting

peterxcli · 2024-12-17T11:45:34Z

This PR only fix these two tests, will migrate them to ClusteTest after I add the ability to use fixed broker port with it.

peterxcli · 2024-12-17T11:46:06Z

Hi @chia7712, this PR is ready for review, PTAL. Thanks!

peterxcli · 2024-12-17T11:47:39Z

@peterxcli I update the topic to match the changes. it is ok to file two PRs :)

As we discussed #18175 (comment), change this PR title back to "Fix ConsumerRebootstrapTest and ProducerRebootstrapTest".

github-actions bot added triage PRs from the community core Kafka Broker tests Test fixes (including flaky tests) small Small PRs labels Dec 13, 2024

chia7712 changed the title ~~KAFKA-17986: Fix ConsumerRebootstrapTest and ProducerRebootstrapTest~~ KAFKA-17986: Fix ProducerRebootstrapTest Dec 14, 2024

wip

5580f3e

github-actions bot removed the small Small PRs label Dec 15, 2024

TaiJuWu reviewed Dec 15, 2024

View reviewed changes

peterxcli added 3 commits December 16, 2024 23:57

Revert "Add support for fixed broker ports in ClusterTest"

8672a64

This reverts commit 33e7c12.

Revert "wip"

99a3a3f

This reverts commit 5580f3e.

github-actions bot added the small Small PRs label Dec 17, 2024

peterxcli marked this pull request as ready for review December 17, 2024 11:43

peterxcli changed the title ~~KAFKA-17986: Fix ProducerRebootstrapTest~~ KAFKA-17986: Fix ConsumerRebootstrapTest and ProducerRebootstrapTest Dec 17, 2024

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

KAFKA-17986: Fix ConsumerRebootstrapTest and ProducerRebootstrapTest #18175

KAFKA-17986: Fix ConsumerRebootstrapTest and ProducerRebootstrapTest #18175

peterxcli commented Dec 13, 2024 •

edited

Loading

chia7712 commented Dec 14, 2024

TaiJuWu Dec 15, 2024

peterxcli Dec 15, 2024

TaiJuWu Dec 16, 2024

peterxcli Dec 16, 2024

peterxcli commented Dec 17, 2024

peterxcli commented Dec 17, 2024

peterxcli commented Dec 17, 2024

	def createTopicWithAdminRaw[B <: KafkaBroker](
	admin: Admin,
	topic: String,
	numPartitions: Int = 1,
	replicationFactor: Int = 1,
	replicaAssignment: collection.Map[Int, Seq[Int]] = Map.empty,
	topicConfig: Properties = new Properties,
	): Uuid = {
	val configsMap = new util.HashMap[String, String]()
	topicConfig.forEach((k, v) => configsMap.put(k.toString, v.toString))

	val result = if (replicaAssignment.isEmpty) {
	admin.createTopics(Collections.singletonList(new NewTopic(
	topic, numPartitions, replicationFactor.toShort).configs(configsMap)))
	} else {
	val assignment = new util.HashMap[Integer, util.List[Integer]]()
	replicaAssignment.foreachEntry { case (k, v) =>
	val replicas = new util.ArrayList[Integer]
	v.foreach(r => replicas.add(r.asInstanceOf[Integer]))
	assignment.put(k.asInstanceOf[Integer], replicas)
	}
	admin.createTopics(Collections.singletonList(new NewTopic(
	topic, assignment).configs(configsMap)))
	}

	result.topicId(topic).get()
	}

KAFKA-17986: Fix ConsumerRebootstrapTest and ProducerRebootstrapTest #18175

Are you sure you want to change the base?

KAFKA-17986: Fix ConsumerRebootstrapTest and ProducerRebootstrapTest #18175

Conversation

peterxcli commented Dec 13, 2024 • edited Loading

Committer Checklist (excluded from commit message)

chia7712 commented Dec 14, 2024

TaiJuWu Dec 15, 2024

Choose a reason for hiding this comment

peterxcli Dec 15, 2024

Choose a reason for hiding this comment

TaiJuWu Dec 16, 2024

Choose a reason for hiding this comment

peterxcli Dec 16, 2024

Choose a reason for hiding this comment

peterxcli commented Dec 17, 2024

peterxcli commented Dec 17, 2024

peterxcli commented Dec 17, 2024

peterxcli commented Dec 13, 2024 •

edited

Loading