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

Code before:

        assertEquals(testGroupId, testGroupDescription.groupId)
        assertEquals(consumerSet.size, testGroupDescription.members().size())

        // Describing a share group using describeConsumerGroups reports it as a non-existent group
        // but the error message is different
        val describeConsumerGroupResult = client.describeConsumerGroups(util.Set.of(testGroupId),
          new DescribeConsumerGroupsOptions().includeAuthorizedOperations(true))
        assertFutureThrows(classOf[GroupIdNotFoundException],
          describeConsumerGroupResult.all(),
          s"Group $testGroupId is not a consumer group.")
      } finally {
        consumerThreads.foreach {
          case consumerThread =>
            consumerThread.interrupt()
            consumerThread.join()
        }
      }
    } finally {
      consumerSet.foreach(consumer => Utils.closeQuietly(consumer, "consumer"))
      Utils.closeQuietly(producer, "producer")
      Utils.closeQuietly(client, "adminClient")
    }
  }

  def sleepMillisToPropagateMetadata(durationMs: Long, partition: TopicPartition): Unit = {
    // Sleep 1000ms to ensure the first broker's metadata has been updated
    TimeUnit.MILLISECONDS.sleep(1000)
    val prior = brokers.head.metadataCache.getPartitionLeaderEndpoint(partition.topic, partition.partition(), listenerName).get.id()
    TestUtils.waitUntilTrue(
      () => {
        val allSynced = brokers.forall { broker =>
          val leaderIdOpt = broker.metadataCache.getPartitionLeaderEndpoint(partition.topic, partition.partition(), listenerName).map(_.id())
          val leaderId = leaderIdOpt.toScala
          leaderId.contains(prior)
        }
        allSynced
      }, s"Waited more than $durationMs ms, but not all brokers' metadata had updated the leader for partition $partition", durationMs)
  }

  @Test
  def testElectPreferredLeaders(): Unit = {
    client = createAdminClient

    val prefer0 = Seq(0, 1, 2)
    val prefer1 = Seq(1, 2, 0)
    val prefer2 = Seq(2, 0, 1)

    val partition1 = new TopicPartition("elect-preferred-leaders-topic-1", 0)
    createTopicWithAssignment(partition1.topic, Map[Int, Seq[Int]](partition1.partition -> prefer0))


Review comment (bug):
From my understanding, this function ensures that the leader update has propagated to all brokers’ metadata caches.
Is there a reason why this check isn't integrated directly into changePreferredLeader?
Please correct me if I'm wrong or miss something .

Code after:
        assertEquals(testGroupId, testGroupDescription.groupId)
        assertEquals(consumerSet.size, testGroupDescription.members().size())

        // Describing a share group using describeConsumerGroups reports it as a non-existent group
        // but the error message is different
        val describeConsumerGroupResult = client.describeConsumerGroups(util.Set.of(testGroupId),
          new DescribeConsumerGroupsOptions().includeAuthorizedOperations(true))
        assertFutureThrows(classOf[GroupIdNotFoundException],
          describeConsumerGroupResult.all(),
          s"Group $testGroupId is not a consumer group.")
      } finally {
        consumerThreads.foreach {
          case consumerThread =>
            consumerThread.interrupt()
            consumerThread.join()
        }
      }
    } finally {
      consumerSet.foreach(consumer => Utils.closeQuietly(consumer, "consumer"))
      Utils.closeQuietly(client, "adminClient")
    }
  }

  @Test
  def testDeleteShareGroupOffsets(): Unit = {
    val config = createConfig
    client = Admin.create(config)
    val testTopicName = "test_topic"
    val testGroupId = "test_group_id"
    val testClientId = "test_client_id"
    val fakeGroupId = "fake_group_id"
    val fakeTopicName = "foo"

    try {
      prepareTopics(List(testTopicName), 1)
      prepareRecords(testTopicName)

      val newShareConsumerConfig = new Properties(consumerConfig)
      newShareConsumerConfig.setProperty(ConsumerConfig.GROUP_ID_CONFIG, testGroupId)
      newShareConsumerConfig.setProperty(ConsumerConfig.CLIENT_ID_CONFIG, testClientId)

      Using.resource(createShareConsumer(configOverrides = newShareConsumerConfig)) { consumer =>
        consumer.subscribe(util.List.of(testTopicName))
        consumer.poll(JDuration.ofMillis(DEFAULT_MAX_WAIT_MS))
        consumer.commitSync()

        // listGroups is used to list share groups
        // Test that we can list the new group.
        TestUtils.waitUntilTrue(() => {
          client.listGroups.all.get.stream().filter(group =>
            group.groupId == testGroupId &&