You are viewing a plain text version of this content. The canonical link for it is here.
Posted to jira@kafka.apache.org by GitBox <gi...@apache.org> on 2022/10/13 01:18:38 UTC
[GitHub] [kafka] hachikuji commented on a diff in pull request #12738: minor: Trying to fix some flakiness in KRaftClusterTest::testDescribeQuorumRequestToBrokers()
hachikuji commented on code in PR #12738:
URL: https://github.com/apache/kafka/pull/12738#discussion_r994016187
##########
core/src/test/scala/integration/kafka/server/KRaftClusterTest.scala:
##########
@@ -811,20 +811,29 @@ 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"Atleast 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 => observers.stream.allMatch(observer => (observer.logEndOffset > 0
Review Comment:
Hmm, do these checks achieve anything? In what case would the leader report an observer which doesn't have these fields set? I'm considering alternatively if the condition we should be waiting for here is this one:
```scala
assertEquals(cluster.brokers.asScala.keySet, observers.asScala.map(_.replicaId).toSet)
```
Once we are sure we have responses from all expected observers, then we can assert the additional fields. Does that make sense?
##########
core/src/test/scala/integration/kafka/server/KRaftClusterTest.scala:
##########
@@ -811,20 +811,29 @@ 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(
Review Comment:
nit: space after `val`
##########
core/src/test/scala/integration/kafka/server/KRaftClusterTest.scala:
##########
@@ -811,20 +811,29 @@ 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"Atleast one voter did not return the expected state within timeout." +
Review Comment:
nit: "At least" (same typo below)
##########
core/src/test/scala/integration/kafka/server/KRaftClusterTest.scala:
##########
@@ -811,20 +811,29 @@ 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()){
Review Comment:
nit: feels like there should be a space before the brace? I would probably write it like this:
```scala
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()
}
}
```
Guess it still looks a bit awkward...
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org