Browse Source

KAFKA-15566: Fix test FetchRequestTest.testLastFetchedEpochValidation for KRaft mode (#14563)

Fix test FetchRequestTest.testLastFetchedEpochValidation for KRaft mode
The test fails due to unexpected error (OFFSET_OUT_OF_RANGE) when enabled with KRaft mode.
The reason it takes longer to set the leader epoch in KRaft mode is because of the way the topic partitions are created differently than Zookeeper. In Zookeeper mode, we create the topic partitions directly with Zookeeper therefore seem to take less time to create the logs and set leader epoch on broker. In KRaft mode, we use Admin client to create topic partitions. Even though the test waits for topic partitions to get created and appear in metadata cache, it doesn’t seem to be sufficient time for leader epoch to get set on the brokers.

Reviewers: Luke Chen <showuon@gmail.com>, dengziming <dengziming1993@gmail.com>
pull/7111/head
Gantigmaa Selenge 1 year ago committed by GitHub
parent
commit
84a58d75bb
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
  1. 5
      core/src/test/scala/unit/kafka/server/FetchRequestTest.scala

5
core/src/test/scala/unit/kafka/server/FetchRequestTest.scala

@ -196,13 +196,13 @@ class FetchRequestTest extends BaseFetchRequestTest { @@ -196,13 +196,13 @@ class FetchRequestTest extends BaseFetchRequestTest {
}
@ParameterizedTest(name = TestInfoUtils.TestWithParameterizedQuorumName)
@ValueSource(strings = Array("zk"))
@ValueSource(strings = Array("zk", "kraft"))
def testLastFetchedEpochValidation(quorum: String): Unit = {
checkLastFetchedEpochValidation(ApiKeys.FETCH.latestVersion())
}
@ParameterizedTest(name = TestInfoUtils.TestWithParameterizedQuorumName)
@ValueSource(strings = Array("zk"))
@ValueSource(strings = Array("zk", "kraft"))
def testLastFetchedEpochValidationV12(quorum: String): Unit = {
checkLastFetchedEpochValidation(12)
}
@ -211,6 +211,7 @@ class FetchRequestTest extends BaseFetchRequestTest { @@ -211,6 +211,7 @@ class FetchRequestTest extends BaseFetchRequestTest {
val topic = "topic"
val topicPartition = new TopicPartition(topic, 0)
val partitionToLeader = createTopic(topic, replicationFactor = 3)
TestUtils.waitUntilLeaderIsKnown(brokers, topicPartition)
val firstLeaderId = partitionToLeader(topicPartition.partition)
val firstLeaderEpoch = TestUtils.findLeaderEpoch(firstLeaderId, topicPartition, brokers)

Loading…
Cancel
Save