chia7712 commented on code in PR #20496:
URL: https://github.com/apache/kafka/pull/20496#discussion_r2334938734
##########
core/src/test/scala/integration/kafka/api/PlaintextAdminIntegrationTest.scala:
##########
@@ -2628,20 +2628,22 @@ class PlaintextAdminIntegrationTest extends
BaseAdminIntegrationTest {
val consumerGroupListing = new GroupListing(consumerGroupId,
Optional.of(GroupType.CONSUMER), "consumer", Optional.of(GroupState.STABLE))
val shareGroupListing = new GroupListing(shareGroupId,
Optional.of(GroupType.SHARE), "share", Optional.of(GroupState.STABLE))
val simpleGroupListing = new GroupListing(simpleGroupId,
Optional.of(GroupType.CLASSIC), "", Optional.of(GroupState.EMPTY))
- // Streams group could either be in STABLE or NOT_READY state
- val streamsGroupListingStable = new GroupListing(streamsGroupId,
Optional.of(GroupType.STREAMS), "streams", Optional.of(GroupState.STABLE))
- val streamsGroupListingNotReady = new GroupListing(streamsGroupId,
Optional.of(GroupType.STREAMS), "streams", Optional.of(GroupState.NOT_READY))
+ val streamsGroupListing = new GroupListing(streamsGroupId,
Optional.of(GroupType.STREAMS), "streams", Optional.of(GroupState.STABLE))
var listGroupsResult = client.listGroups()
assertTrue(listGroupsResult.errors().get().isEmpty)
- val expectedStreamListings = Set(streamsGroupListingStable,
streamsGroupListingNotReady)
- val expectedListings = Set(classicGroupListing, simpleGroupListing,
consumerGroupListing, shareGroupListing)
- val actualListings = listGroupsResult.all().get().asScala.toSet
-
- // Check that actualListings contains all expectedListings and one of
the streams listings
- assertTrue(expectedListings.subsetOf(actualListings))
- assertTrue(actualListings.exists(expectedStreamListings.contains))
+ TestUtils.waitUntilTrue(() => {
+ val listGroupResultScala = client.listGroups().all().get().asScala
+ val filteredStreamsGroups = listGroupResultScala.filter(_.groupId() ==
streamsGroupId)
+ val filteredClassicGroups = listGroupResultScala.filter(_.groupId() ==
classicGroupId)
+ val filteredConsumerGroups = listGroupResultScala.filter(_.groupId()
== consumerGroupId)
+ val filteredShareGroups = listGroupResultScala.filter(_.groupId() ==
shareGroupId)
+ filteredClassicGroups.forall(_.groupState().orElse(null) ==
GroupState.STABLE) &&
+ filteredConsumerGroups.forall(_.groupState().orElse(null) ==
GroupState.STABLE) &&
+ filteredShareGroups.forall(_.groupState().orElse(null) ==
GroupState.STABLE) &&
+ filteredStreamsGroups.forall(_.groupState().orElse(null) ==
GroupState.STABLE)
Review Comment:
Should we also add the check for simple group?
--
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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
