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

MINOR: Address flakiness in `KRaftClusterTest::testDescribeQuorumRequestToBrokers` (#12738) 

We have seen some errors such as the following:
```
org.opentest4j.AssertionFailedError: expected: not equal but was: <OptionalLong.empty>
Stacktrace
org.opentest4j.AssertionFailedError: expected: not equal but was: <OptionalLong.empty>
	at org.junit.jupiter.api.AssertionUtils.fail(AssertionUtils.java:39)
	at org.junit.jupiter.api.AssertNotEquals.failEqual(AssertNotEquals.java:276)
	at org.junit.jupiter.api.AssertNotEquals.assertNotEquals(AssertNotEquals.java:265)
	at org.junit.jupiter.api.AssertNotEquals.assertNotEquals(AssertNotEquals.java:260)
	at org.junit.jupiter.api.Assertions.assertNotEquals(Assertions.java:2815)
	at kafka.server.KRaftClusterTest.$anonfun$testDescribeQuorumRequestToBrokers$5(KRaftClusterTest.scala:818)
	at java.util.ArrayList.forEach(ArrayList.java:1259)
	at kafka.server.KRaftClusterTest.testDescribeQuorumRequestToBrokers(KRaftClusterTest.scala:814)
```
The patch changes some of the assertions to wait longer for the condition to be satisfied.

Reviewers: Jason Gustafson <jason@confluent.io>
This commit is contained in:
Niket 2022-10-17 16:16:26 -07:00 committed by GitHub
parent e8e815b1b6
commit 3e7eddecd6
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
1 changed files with 26 additions and 13 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

39
core/src/test/scala/integration/kafka/server/KRaftClusterTest.scala
View File

@ -811,20 +811,33 @@ class KRaftClusterTest {
assertTrue(cluster.controllers.asScala.keySet.contains(quorumInfo.leaderId),
s"Leader ID ${quorumInfo.leaderId} was not a controller ID.")
quorumInfo.voters.forEach { voter =>
assertTrue(0 < voter.logEndOffset,
s"logEndOffset for voter with ID ${voter.replicaId} was ${voter.logEndOffset}")
assertNotEquals(OptionalLong.empty(), voter.lastFetchTimestamp)
assertNotEquals(OptionalLong.empty(), voter.lastCaughtUpTimestamp)
}
val (voters, voterResponseValid) =
TestUtils.computeUntilTrue(
admin.describeMetadataQuorum(new DescribeMetadataQuorumOptions)
.quorumInfo().get().voters()
) { voters => voters.stream
.allMatch(voter => (voter.logEndOffset > 0
&& voter.lastFetchTimestamp() != OptionalLong.empty()
&& voter.lastCaughtUpTimestamp() != OptionalLong.empty()))
}
assertEquals(cluster.brokers.asScala.keySet, quorumInfo.observers.asScala.map(_.replicaId).toSet)
quorumInfo.observers.forEach { observer =>
assertTrue(0 < observer.logEndOffset,
s"logEndOffset for observer with ID ${observer.replicaId} was ${observer.logEndOffset}")
assertNotEquals(OptionalLong.empty(), observer.lastFetchTimestamp)
assertNotEquals(OptionalLong.empty(), observer.lastCaughtUpTimestamp)
}
assertTrue(voterResponseValid, s"At least one voter did not return the expected state within timeout." +
s"The responses gathered for all the voters: ${voters.toString}")
val (observers, observerResponseValid) =
TestUtils.computeUntilTrue(
admin.describeMetadataQuorum(new DescribeMetadataQuorumOptions)
.quorumInfo().get().observers()
) { observers =>
(
cluster.brokers.asScala.keySet == observers.asScala.map(_.replicaId).toSet
&& observers.stream.allMatch(observer => (observer.logEndOffset > 0
&& observer.lastFetchTimestamp() != OptionalLong.empty()
&& observer.lastCaughtUpTimestamp() != OptionalLong.empty())))
}
assertTrue(observerResponseValid, s"At least one observer did not return the expected state within timeout." +
s"The responses gathered for all the observers: ${observers.toString}")
} finally {
admin.close()
}