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

MINOR: Deflake EligibleLeaderReplicasIntegrationTest (#18923) 

Make sure to give enough time for the partition ISR updates.

Reviewers: David Jacot <djacot@confluent.io>
This commit is contained in:
Calvin Liu 2025-02-20 05:14:15 -08:00 committed by GitHub
parent c89fd2bff6
commit 1eecd02ce8
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
1 changed files with 21 additions and 11 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

32
core/src/test/java/kafka/server/integration/EligibleLeaderReplicasIntegrationTest.java
View File

@ -70,6 +70,7 @@ import scala.collection.JavaConverters;
import scala.collection.Seq;
import scala.collection.mutable.HashMap;
import static org.apache.kafka.test.TestUtils.DEFAULT_MAX_WAIT_MS;
import static org.junit.jupiter.api.Assertions.assertDoesNotThrow;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertTrue;
@ -202,7 +203,7 @@ public class EligibleLeaderReplicasIntegrationTest extends KafkaServerTestHarnes
}
void waitUntilOneMessageIsConsumed(Consumer consumer) {
kafka.utils.TestUtils.waitUntilTrue(
TestUtils.waitUntilTrue(
() -> {
try {
ConsumerRecords record = consumer.poll(Duration.ofMillis(100L));
@ -212,7 +213,7 @@ public class EligibleLeaderReplicasIntegrationTest extends KafkaServerTestHarnes
}
},
() -> "fail to consume messages",
org.apache.kafka.test.TestUtils.DEFAULT_MAX_WAIT_MS, 100L
DEFAULT_MAX_WAIT_MS, 100L
);
}
@ -417,30 +418,39 @@ public class EligibleLeaderReplicasIntegrationTest extends KafkaServerTestHarnes
waitForIsrAndElr((isrSize, elrSize) -> {
return isrSize > 0 && elrSize == 0;
});
topicPartitionInfo = adminClient.describeTopics(Collections.singletonList(testTopicName))
.allTopicNames().get().get(testTopicName).partitions().get(0);
assertEquals(0, topicPartitionInfo.lastKnownElr().size());
assertEquals(0, topicPartitionInfo.elr().size());
assertEquals(lastKnownLeader, topicPartitionInfo.leader().id());
TestUtils.waitUntilTrue(
() -> {
try {
TopicPartitionInfo partition = adminClient.describeTopics(Collections.singletonList(testTopicName))
.allTopicNames().get().get(testTopicName).partitions().get(0);
if (partition.leader() == null) return false;
return partition.lastKnownElr().isEmpty() && partition.elr().isEmpty() && partition.leader().id() == lastKnownLeader;
} catch (Exception e) {
return false;
}
},
() -> String.format("Partition metadata for %s is not correct", testTopicName),
DEFAULT_MAX_WAIT_MS, 100L
);
} finally {
restartDeadBrokers(false);
}
}
void waitForIsrAndElr(BiFunction<Integer, Integer, Boolean> isIsrAndElrSizeSatisfied) {
kafka.utils.TestUtils.waitUntilTrue(
TestUtils.waitUntilTrue(
() -> {
try {
TopicDescription topicDescription = adminClient.describeTopics(Collections.singletonList(testTopicName))
.allTopicNames().get().get(testTopicName);
TopicPartitionInfo partition = topicDescription.partitions().get(0);
if (!isIsrAndElrSizeSatisfied.apply(partition.isr().size(), partition.elr().size())) return false;
return isIsrAndElrSizeSatisfied.apply(partition.isr().size(), partition.elr().size());
} catch (Exception e) {
return false;
}
return true;
},
() -> String.format("Partition metadata for %s is not propagated", testTopicName),
org.apache.kafka.test.TestUtils.DEFAULT_MAX_WAIT_MS, 100L);
DEFAULT_MAX_WAIT_MS, 100L);
}
}