Repository: apache/kafka
File: core/src/test/scala/integration/kafka/api/PlaintextAdminIntegrationTest.scala (Scala)

Code before:
      assertNull(alterConsumerGroupOffsetsResult.partitionResult(topicPartition).get())

      TestUtils.waitUntilTrue(() => {
        val groups = client.listGroups().all().get()
        groups.size() == 5
      }, "Expected to find all groups")

      val classicGroupListing = new GroupListing(classicGroupId, Optional.of(GroupType.CLASSIC), "consumer", Optional.of(GroupState.STABLE))
      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))
      val streamsGroupListing = new GroupListing(streamsGroupId, Optional.of(GroupType.STREAMS), "streams", Optional.of(GroupState.STABLE))

      var listGroupsResult = client.listGroups()
      assertTrue(listGroupsResult.errors().get().isEmpty)

      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) &&
          filteredClassicGroups.forall(_.groupState().orElse(null) == GroupState.STABLE) &&
          filteredStreamsGroups.forall(_.groupState().orElse(null) == GroupState.STABLE)
      }, "Groups not stable yet")

      listGroupsResult = client.listGroups(new ListGroupsOptions().withTypes(util.Set.of(GroupType.CLASSIC)))
      assertTrue(listGroupsResult.errors().get().isEmpty)
      assertEquals(Set(classicGroupListing, simpleGroupListing), listGroupsResult.all().get().asScala.toSet)
      assertEquals(Set(classicGroupListing, simpleGroupListing), listGroupsResult.valid().get().asScala.toSet)

      listGroupsResult = client.listGroups(new ListGroupsOptions().withTypes(util.Set.of(GroupType.CONSUMER)))
      assertTrue(listGroupsResult.errors().get().isEmpty)
      assertEquals(Set(consumerGroupListing), listGroupsResult.all().get().asScala.toSet)
      assertEquals(Set(consumerGroupListing), listGroupsResult.valid().get().asScala.toSet)

      listGroupsResult = client.listGroups(new ListGroupsOptions().withTypes(util.Set.of(GroupType.SHARE)))
      assertTrue(listGroupsResult.errors().get().isEmpty)
      assertEquals(Set(shareGroupListing), listGroupsResult.all().get().asScala.toSet)
      assertEquals(Set(shareGroupListing), listGroupsResult.valid().get().asScala.toSet)

      listGroupsResult = client.listGroups(new ListGroupsOptions().withTypes(util.Set.of(GroupType.STREAMS)))
      assertTrue(listGroupsResult.errors().get().isEmpty)
      assertEquals(Set(streamsGroupListing), listGroupsResult.all().get().asScala.toSet)
      assertEquals(Set(streamsGroupListing), listGroupsResult.valid().get().asScala.toSet)

    } finally {
      Utils.closeQuietly(classicGroup, "classicGroup")

Review comment (refactor):
`filteredClassicGroups.forall(_.groupState().orElse(null) == GroupState.STABLE) &&`

It looks like this line is duplicated.


Code after:
      assertNull(alterConsumerGroupOffsetsResult.partitionResult(topicPartition).get())

      TestUtils.waitUntilTrue(() => {
        val groups = client.listGroups().all().get()
        groups.size() == 5
      }, "Expected to find all groups")

      val classicGroupListing = new GroupListing(classicGroupId, Optional.of(GroupType.CLASSIC), "consumer", Optional.of(GroupState.STABLE))
      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))
      val streamsGroupListing = new GroupListing(streamsGroupId, Optional.of(GroupType.STREAMS), "streams", Optional.of(GroupState.STABLE))

      var listGroupsResult = client.listGroups()
      assertTrue(listGroupsResult.errors().get().isEmpty)

      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)
      }, "Groups not stable yet")

      listGroupsResult = client.listGroups(new ListGroupsOptions().withTypes(util.Set.of(GroupType.CLASSIC)))
      assertTrue(listGroupsResult.errors().get().isEmpty)
      assertEquals(Set(classicGroupListing, simpleGroupListing), listGroupsResult.all().get().asScala.toSet)
      assertEquals(Set(classicGroupListing, simpleGroupListing), listGroupsResult.valid().get().asScala.toSet)

      listGroupsResult = client.listGroups(new ListGroupsOptions().withTypes(util.Set.of(GroupType.CONSUMER)))
      assertTrue(listGroupsResult.errors().get().isEmpty)
      assertEquals(Set(consumerGroupListing), listGroupsResult.all().get().asScala.toSet)
      assertEquals(Set(consumerGroupListing), listGroupsResult.valid().get().asScala.toSet)

      listGroupsResult = client.listGroups(new ListGroupsOptions().withTypes(util.Set.of(GroupType.SHARE)))
      assertTrue(listGroupsResult.errors().get().isEmpty)
      assertEquals(Set(shareGroupListing), listGroupsResult.all().get().asScala.toSet)
      assertEquals(Set(shareGroupListing), listGroupsResult.valid().get().asScala.toSet)

      listGroupsResult = client.listGroups(new ListGroupsOptions().withTypes(util.Set.of(GroupType.STREAMS)))
      assertTrue(listGroupsResult.errors().get().isEmpty)
      assertEquals(Set(streamsGroupListing), listGroupsResult.all().get().asScala.toSet)
      assertEquals(Set(streamsGroupListing), listGroupsResult.valid().get().asScala.toSet)

    } finally {
      Utils.closeQuietly(classicGroup, "classicGroup")
      Utils.closeQuietly(consumerGroup, "consumerGroup")