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

Code before:
        }
      }
    } finally {
      consumerSet.foreach(consumer => Utils.closeQuietly(consumer, "consumer"))
      Utils.closeQuietly(producer, "producer")
      Utils.closeQuietly(client, "adminClient")
    }
  }

  /**
   * Waits until the metadata for the given partition has fully propagated
   * and become consistent across all brokers.
   *
   * This method repeatedly checks the leader information for the specified
   * TopicPartition in each broker's metadata cache. It compares all brokers'
   * views against the leader reported by the head broker. The loop continues
   * until all brokers agree on the same leader, ensuring metadata consistency.
   *
   * This is useful in integration tests where operations such as
   * preferred leader election or other metadata updates require
   * propagation time before assertions can be made reliably.
   */
  def sleepMillisToPropagateMetadata(partition: TopicPartition): Unit = {
    var allSynced: Boolean = false

    while (!allSynced) {
      val prior = brokers.head.metadataCache.getPartitionLeaderEndpoint(partition.topic, partition.partition(), listenerName).get.id()

      allSynced = brokers.forall { broker =>
        val leaderIdOpt = broker.metadataCache.getPartitionLeaderEndpoint(partition.topic, partition.partition(), listenerName).map(_.id())
        val leaderId = leaderIdOpt.toScala
        leaderId.contains(prior)
      }
    }
  }

  @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))

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

    def preferredLeader(topicPartition: TopicPartition): Int = {

Review comment (performance):
```scala
  def sleepMillisToPropagateMetadata(partition: TopicPartition): Unit = {
    while (true) {
      val prior = brokers.head.metadataCache.getPartitionLeaderEndpoint(partition.topic, partition.partition(), listenerName).get.id()
      if (brokers.forall { broker =>
        val leaderIdOpt = broker.metadataCache.getPartitionLeaderEndpoint(partition.topic, partition.partition(), listenerName).map(_.id())
        val leaderId = leaderIdOpt.toScala
        leaderId.contains(prior)
      }) return
    }
  }
```

Code after:
      }
    } 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 &&
              group.groupState.get == GroupState.STABLE).count() == 1
        }, s"Expected to be able to list $testGroupId")

        // Test offset deletion while consuming
        val offsetDeleteResult = client.deleteShareGroupOffsets(testGroupId, util.Set.of(testTopicName, fakeTopicName))

        // Deleting the offset with real group ID should get GroupNotEmptyException
        assertFutureThrows(classOf[GroupNotEmptyException], offsetDeleteResult.all())
        assertFutureThrows(classOf[GroupNotEmptyException], offsetDeleteResult.topicResult(testTopicName))
        assertFutureThrows(classOf[GroupNotEmptyException], offsetDeleteResult.topicResult(fakeTopicName))

        // Test the fake group ID
        val fakeDeleteResult = client.deleteShareGroupOffsets(fakeGroupId, util.Set.of(testTopicName, fakeTopicName))

        assertFutureThrows(classOf[GroupIdNotFoundException], fakeDeleteResult.all())
        assertFutureThrows(classOf[GroupIdNotFoundException], fakeDeleteResult.topicResult(testTopicName))