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

Code before:
    val simpleGroupId = "simple_group_id"
    val streamsGroupId = "streams_group_id"
    val testTopicName = "test_topic"
    val testStreamsOutputTopicName = "test_streams_output_topic"

    consumerConfig.put(ConsumerConfig.GROUP_PROTOCOL_CONFIG, GroupProtocol.CLASSIC.name)
    val classicGroupConfig = new Properties(consumerConfig)
    classicGroupConfig.put(ConsumerConfig.GROUP_ID_CONFIG, classicGroupId)
    val classicGroup = createConsumer(configOverrides = classicGroupConfig)

    consumerConfig.put(ConsumerConfig.GROUP_PROTOCOL_CONFIG, GroupProtocol.CONSUMER.name)
    val consumerGroupConfig = new Properties(consumerConfig)
    consumerGroupConfig.put(ConsumerConfig.GROUP_ID_CONFIG, consumerGroupId)
    val consumerGroup = createConsumer(configOverrides = consumerGroupConfig)

    val shareGroupConfig = new Properties(consumerConfig)
    shareGroupConfig.put(ConsumerConfig.GROUP_ID_CONFIG, shareGroupId)
    val shareGroup = createShareConsumer(configOverrides = shareGroupConfig)

    val streamsConfig = new Properties()
    streamsConfig.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest")

    val streamsGroup = createStreamsGroup(
      configOverrides = streamsConfig ,
      inputTopic = testTopicName ,
      outputTopic = testStreamsOutputTopicName ,
      streamsGroupId = streamsGroupId
    )

    val config = createConfig
    client = Admin.create(config)
    try {
      //      client.createTopics(util.Set.of(
      //        new NewTopic(testTopicName, 1, 1.toShort),
      //        new NewTopic(testStreamsOutputTopicName, 1, 1.toShort)
      //      )).all().get()
      //
      //      waitForTopics(client, List(testTopicName), List())
      val topicPartition = new TopicPartition(testTopicName, 0)

      classicGroup.subscribe(util.Set.of(testTopicName))
      classicGroup.poll(JDuration.ofMillis(1000))
      consumerGroup.subscribe(util.Set.of(testTopicName))
      consumerGroup.poll(JDuration.ofMillis(1000))
      shareGroup.subscribe(util.Set.of(testTopicName))
      shareGroup.poll(JDuration.ofMillis(1000))

      val alterConsumerGroupOffsetsResult = client.alterConsumerGroupOffsets(simpleGroupId,
        util.Map.of(topicPartition, new OffsetAndMetadata(0L)))
      assertNull(alterConsumerGroupOffsetsResult.all().get())
      assertNull(alterConsumerGroupOffsetsResult.partitionResult(topicPartition).get())

Review comment (suggestion):
Remove trailing spaces after commas on lines 2599-2601 to maintain consistent code formatting.
```suggestion
      configOverrides = streamsConfig,
      inputTopic = testTopicName,
      outputTopic = testStreamsOutputTopicName,
```

Code after:
    val simpleGroupId = "simple_group_id"
    val streamsGroupId = "streams_group_id"
    val testTopicName = "test_topic"

    val config = createConfig
    client = Admin.create(config)

    consumerConfig.put(ConsumerConfig.GROUP_PROTOCOL_CONFIG, GroupProtocol.CLASSIC.name)
    val classicGroupConfig = new Properties(consumerConfig)
    classicGroupConfig.put(ConsumerConfig.GROUP_ID_CONFIG, classicGroupId)
    val classicGroup = createConsumer(configOverrides = classicGroupConfig)

    consumerConfig.put(ConsumerConfig.GROUP_PROTOCOL_CONFIG, GroupProtocol.CONSUMER.name)
    val consumerGroupConfig = new Properties(consumerConfig)
    consumerGroupConfig.put(ConsumerConfig.GROUP_ID_CONFIG, consumerGroupId)
    val consumerGroup = createConsumer(configOverrides = consumerGroupConfig)

    val shareGroupConfig = new Properties(consumerConfig)
    shareGroupConfig.put(ConsumerConfig.GROUP_ID_CONFIG, shareGroupId)
    val shareGroup = createShareConsumer(configOverrides = shareGroupConfig)

    val streamsGroup = createStreamsGroup(
      inputTopic = testTopicName,
      streamsGroupId = streamsGroupId
    )

    try {
      client.createTopics(util.Set.of(
        new NewTopic(testTopicName, 1, 1.toShort)
      )).all().get()
      waitForTopics(client, List(testTopicName), List())
      val topicPartition = new TopicPartition(testTopicName, 0)

      classicGroup.subscribe(util.Set.of(testTopicName))
      classicGroup.poll(JDuration.ofMillis(1000))
      consumerGroup.subscribe(util.Set.of(testTopicName))
      consumerGroup.poll(JDuration.ofMillis(1000))
      shareGroup.subscribe(util.Set.of(testTopicName))
      shareGroup.poll(JDuration.ofMillis(1000))
      streamsGroup.poll(JDuration.ofMillis(1000))

      val alterConsumerGroupOffsetsResult = client.alterConsumerGroupOffsets(simpleGroupId,
        util.Map.of(topicPartition, new OffsetAndMetadata(0L)))
      assertNull(alterConsumerGroupOffsetsResult.all().get())
      assertNull(alterConsumerGroupOffsetsResult.partitionResult(topicPartition).get())

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