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

Code before:
        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))
        assertFutureThrows(classOf[GroupIdNotFoundException], fakeDeleteResult.topicResult(fakeTopicName))
      }

      // Test offset deletion when group is empty
      val offsetDeleteResult = client.deleteShareGroupOffsets(testGroupId, util.Set.of(testTopicName, fakeTopicName))

      assertFutureThrows(classOf[UnknownTopicOrPartitionException], offsetDeleteResult.all())
      assertNull(offsetDeleteResult.topicResult(testTopicName).get())
      assertFutureThrows(classOf[UnknownTopicOrPartitionException], offsetDeleteResult.topicResult(fakeTopicName))
    } finally {
      Utils.closeQuietly(client, "adminClient")
    }
  }

  @Test
  def testAlterShareGroupOffsets(): 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"

    val tp1 = new TopicPartition(testTopicName, 0)
    val tp2 = new TopicPartition(fakeTopicName, 0)
    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(() => {

Review comment (suggestion):
We should be able to see whether offsets were deleted using `listShareGroupOffsets`.

Code after:
        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))
        assertFutureThrows(classOf[GroupIdNotFoundException], fakeDeleteResult.topicResult(fakeTopicName))
      }

      // Test offset deletion when group is empty
      val offsetDeleteResult = client.deleteShareGroupOffsets(testGroupId, util.Set.of(testTopicName, fakeTopicName))

      assertFutureThrows(classOf[UnknownTopicOrPartitionException], offsetDeleteResult.all())
      assertNull(offsetDeleteResult.topicResult(testTopicName).get())
      assertFutureThrows(classOf[UnknownTopicOrPartitionException], offsetDeleteResult.topicResult(fakeTopicName))

      val tp1 = new TopicPartition(testTopicName, 0)
      val parts = client.listShareGroupOffsets(util.Map.of(testGroupId, new ListShareGroupOffsetsSpec().topicPartitions(util.List.of(tp1))))
        .partitionsToOffsetAndMetadata(testGroupId)
        .get()
      assertTrue(parts.containsKey(tp1))
      assertNull(parts.get(tp1))
    } finally {
      Utils.closeQuietly(client, "adminClient")
    }
  }

  @Test
  def testAlterShareGroupOffsets(): 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"

    val tp1 = new TopicPartition(testTopicName, 0)
    val tp2 = new TopicPartition(fakeTopicName, 0)
    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 =>