You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@kafka.apache.org by ch...@apache.org on 2025/02/13 23:28:51 UTC
(kafka) branch trunk updated: KAFKA-18298 Fix flaky testConsumerGroupsDeprecatedConsumerGroupState and testConsumerGroups in PlaintextAdminIntegrationTest (#18513)
This is an automated email from the ASF dual-hosted git repository.
chia7712 pushed a commit to branch trunk
in repository https://gitbox.apache.org/repos/asf/kafka.git
The following commit(s) were added to refs/heads/trunk by this push:
new 2bbd25841e5 KAFKA-18298 Fix flaky testConsumerGroupsDeprecatedConsumerGroupState and testConsumerGroups in PlaintextAdminIntegrationTest (#18513)
2bbd25841e5 is described below
commit 2bbd25841e5b76d99e731286899b0c5aa07fd850
Author: 陳昱霖(Yu-Lin Chen) <ch...@apache.org>
AuthorDate: Fri Feb 14 07:28:45 2025 +0800
KAFKA-18298 Fix flaky testConsumerGroupsDeprecatedConsumerGroupState and testConsumerGroups in PlaintextAdminIntegrationTest (#18513)
It's related to KAFKA-18298 and KAFKA-18297. The root cause of the flaky tests is member rejoin after member removal. To prevent members from rejoining after being removed, before removing group members, calling `consumers.close` in ConsumerThread . This fix also extract the flaky member removal test to new test `testConsumerGroupWithMemberRemoval`.
Flow of member removal test:
1. Set 2 static consumer + 1 dynamic consumer
2. Close all consumers.
3. remove one static member
4. remove remaining members
Before KIP-1092, the member count is different between ClassicConsumer/AsyncConsumer. (AsyncConsumer will remove dynamic member after consumer closed.)
To get more details, please refer to the discussion under KAFKA-18297 and this PR:
- discussion : [Link](https://issues.apache.org/jira/browse/KAFKA-18297?focusedCommentId=17912537&page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel#comment-17912537)
- review: https://github.com/apache/kafka/pull/18513#pullrequestreview-2589110367
This PR fixed below flaky errors:
1. **PlaintextAdminIntegrationTest#testConsumerGroups**
a. `org.opentest4j.AssertionFailedError: expected: <2> but was: <3>` ([Report](https://ge.apache.org/s/lt3lpviv45cns/tests/task/:core:test/details/kafka.api.PlaintextAdminIntegrationTest/testConsumerGroups(String%2C%20String)%5B1%5D?top-execution=1))
b. `org.opentest4j.AssertionFailedError: expected: <true> but was: <false>` ([Report](https://ge.apache.org/s/jlxo446xalpoa/tests/task/:core:test/details/kafka.api.PlaintextAdminIntegrationTest/testConsumerGroups(String%2C%20String)%5B1%5D?top-execution=1))
2. **PlaintextAdminIntegrationTest#testConsumerGroupsDeprecatedConsumerGroupState**
a. `org.opentest4j.AssertionFailedError: expected: <2> but was: <3>` ([Report](https://ge.apache.org/s/ndoj6s2stb446/tests/task/:core:test/details/kafka.api.PlaintextAdminIntegrationTest/testConsumerGroupsDeprecatedConsumerGroupState(String%2C%20String)%5B1%5D?top-execution=1))
b. `org.opentest4j.AssertionFailedError: expected: <true> but was: <false>` ([Report](https://ge.apache.org/s/kh3jze2tc5qeu/tests/task/:core:test/details/kafka.api.PlaintextAdminIntegrationTest/testConsumerGroupsDeprecatedConsumerGroupState(String%2C%20String)%5B1%5D?top-execution=1))
Reviewers: David Jacot <dj...@confluent.io>, TaiJuWu <tj...@gmail.com>, Chia-Ping Tsai <ch...@gmail.com>
---
.../kafka/api/PlaintextAdminIntegrationTest.scala | 964 +++++++++++----------
1 file changed, 493 insertions(+), 471 deletions(-)
diff --git a/core/src/test/scala/integration/kafka/api/PlaintextAdminIntegrationTest.scala b/core/src/test/scala/integration/kafka/api/PlaintextAdminIntegrationTest.scala
index 9df78d60f9a..6c836cf479c 100644
--- a/core/src/test/scala/integration/kafka/api/PlaintextAdminIntegrationTest.scala
+++ b/core/src/test/scala/integration/kafka/api/PlaintextAdminIntegrationTest.scala
@@ -25,7 +25,7 @@ import java.time.{Duration => JDuration}
import java.util.Arrays.asList
import java.util.concurrent.atomic.{AtomicBoolean, AtomicInteger}
import java.util.concurrent.{CountDownLatch, ExecutionException, TimeUnit}
-import java.util.{Collections, Optional, Properties}
+import java.util.{Collections, Locale, Optional, Properties}
import java.{time, util}
import kafka.integration.KafkaServerTestHarness
import kafka.server.KafkaConfig
@@ -1838,250 +1838,147 @@ class PlaintextAdminIntegrationTest extends BaseAdminIntegrationTest {
Utils.closeQuietly(producer, "producer")
}
- val EMPTY_GROUP_INSTANCE_ID = ""
val testGroupId = "test_group_id"
val testClientId = "test_client_id"
val testInstanceId1 = "test_instance_id_1"
val testInstanceId2 = "test_instance_id_2"
val fakeGroupId = "fake_group_id"
- def createProperties(groupInstanceId: String): Properties = {
- val newConsumerConfig = new Properties(consumerConfig)
- // We need to disable the auto commit because after the members got removed from group, the offset commit
- // will cause the member rejoining and the test will be flaky (check ConsumerCoordinator#OffsetCommitResponseHandler)
- newConsumerConfig.setProperty(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "false")
- newConsumerConfig.setProperty(ConsumerConfig.GROUP_ID_CONFIG, testGroupId)
- newConsumerConfig.setProperty(ConsumerConfig.CLIENT_ID_CONFIG, testClientId)
- if (groupInstanceId != EMPTY_GROUP_INSTANCE_ID) {
- newConsumerConfig.setProperty(ConsumerConfig.GROUP_INSTANCE_ID_CONFIG, groupInstanceId)
- }
- newConsumerConfig
- }
-
// contains two static members and one dynamic member
- val groupInstanceSet = Set(testInstanceId1, testInstanceId2, EMPTY_GROUP_INSTANCE_ID)
- val consumerSet = groupInstanceSet.map { groupInstanceId => createConsumer(configOverrides = createProperties(groupInstanceId))}
+ val groupInstanceSet = Set(testInstanceId1, testInstanceId2, "")
val topicSet = Set(testTopicName, testTopicName1, testTopicName2)
- val latch = new CountDownLatch(consumerSet.size)
- try {
- def createConsumerThread[K,V](consumer: Consumer[K,V], topic: String): Thread = {
- new Thread {
- override def run : Unit = {
- consumer.subscribe(Collections.singleton(topic))
- try {
- while (true) {
- consumer.poll(JDuration.ofSeconds(5))
- if (!consumer.assignment.isEmpty && latch.getCount > 0L)
- latch.countDown()
- try {
- consumer.commitSync()
- } catch {
- case _: CommitFailedException => // Ignore and retry on next iteration.
- }
- }
- } catch {
- case _: InterruptException => // Suppress the output to stderr
- }
- }
- }
+ // We need to disable the auto commit because after the members got removed from group, the offset commit
+ // will cause the member rejoining and the test will be flaky (check ConsumerCoordinator#OffsetCommitResponseHandler)
+ val defaultConsumerConfig = new Properties(consumerConfig)
+ defaultConsumerConfig.setProperty(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "false")
+ defaultConsumerConfig.setProperty(ConsumerConfig.GROUP_ID_CONFIG, testGroupId)
+ defaultConsumerConfig.setProperty(ConsumerConfig.CLIENT_ID_CONFIG, testClientId)
+
+ val backgroundConsumerSet = new BackgroundConsumerSet(defaultConsumerConfig)
+ groupInstanceSet.zip(topicSet).foreach { case (groupInstanceId, topic) =>
+ val configOverrides = new Properties()
+ if (groupInstanceId != "") {
+ // static member
+ configOverrides.setProperty(ConsumerConfig.GROUP_INSTANCE_ID_CONFIG, groupInstanceId)
}
+ backgroundConsumerSet.addConsumer(topic, configOverrides)
+ }
- // Start consumers in a thread that will subscribe to a new group.
- val consumerThreads = consumerSet.zip(topicSet).map(zipped => createConsumerThread(zipped._1, zipped._2))
+ try {
val groupType = if (groupProtocol.equalsIgnoreCase(GroupProtocol.CONSUMER.name)) GroupType.CONSUMER else GroupType.CLASSIC
+ // Start consumer polling threads in the background
+ backgroundConsumerSet.start()
- try {
- consumerThreads.foreach(_.start())
- assertTrue(latch.await(30000, TimeUnit.MILLISECONDS))
- // Test that we can list the new group.
- TestUtils.waitUntilTrue(() => {
- val matching = client.listConsumerGroups.all.get.asScala.filter(group =>
- group.groupId == testGroupId &&
- group.groupState.get == GroupState.STABLE)
- matching.size == 1
- }, s"Expected to be able to list $testGroupId")
-
- TestUtils.waitUntilTrue(() => {
- val options = new ListConsumerGroupsOptions().withTypes(Set(groupType).asJava)
- val matching = client.listConsumerGroups(options).all.get.asScala.filter(group =>
- group.groupId == testGroupId &&
- group.groupState.get == GroupState.STABLE)
- matching.size == 1
- }, s"Expected to be able to list $testGroupId in group type $groupType")
-
- TestUtils.waitUntilTrue(() => {
- val options = new ListConsumerGroupsOptions().withTypes(Set(groupType).asJava)
- .inGroupStates(Set(GroupState.STABLE).asJava)
- val matching = client.listConsumerGroups(options).all.get.asScala.filter(group =>
+ // Test that we can list the new group.
+ TestUtils.waitUntilTrue(() => {
+ val matching = client.listConsumerGroups.all.get.asScala.filter(group =>
group.groupId == testGroupId &&
- group.groupState.get == GroupState.STABLE)
- matching.size == 1
- }, s"Expected to be able to list $testGroupId in group type $groupType and state Stable")
-
- TestUtils.waitUntilTrue(() => {
- val options = new ListConsumerGroupsOptions().inGroupStates(Set(GroupState.STABLE).asJava)
- val matching = client.listConsumerGroups(options).all.get.asScala.filter(group =>
- group.groupId == testGroupId &&
- group.groupState.get == GroupState.STABLE)
- matching.size == 1
- }, s"Expected to be able to list $testGroupId in state Stable")
-
- TestUtils.waitUntilTrue(() => {
- val options = new ListConsumerGroupsOptions().inGroupStates(Set(GroupState.EMPTY).asJava)
- val matching = client.listConsumerGroups(options).all.get.asScala.filter(
- _.groupId == testGroupId)
- matching.isEmpty
- }, s"Expected to find zero groups")
-
- val describeWithFakeGroupResult = client.describeConsumerGroups(Seq(testGroupId, fakeGroupId).asJava,
- new DescribeConsumerGroupsOptions().includeAuthorizedOperations(true))
- assertEquals(2, describeWithFakeGroupResult.describedGroups().size())
-
- // Test that we can get information about the test consumer group.
- assertTrue(describeWithFakeGroupResult.describedGroups().containsKey(testGroupId))
- var testGroupDescription = describeWithFakeGroupResult.describedGroups().get(testGroupId).get()
- if (groupType == GroupType.CLASSIC) {
- assertTrue(testGroupDescription.groupEpoch.isEmpty)
- assertTrue(testGroupDescription.targetAssignmentEpoch.isEmpty)
- } else {
- assertEquals(Optional.of(3), testGroupDescription.groupEpoch)
- assertEquals(Optional.of(3), testGroupDescription.targetAssignmentEpoch)
- }
+ group.groupState.get == GroupState.STABLE)
+ matching.size == 1
+ }, s"Expected to be able to list $testGroupId")
- assertEquals(testGroupId, testGroupDescription.groupId())
- assertFalse(testGroupDescription.isSimpleConsumerGroup)
- assertEquals(groupInstanceSet.size, testGroupDescription.members().size())
- val members = testGroupDescription.members()
- members.asScala.foreach { member =>
- assertEquals(testClientId, member.clientId)
- assertEquals(if (groupType == GroupType.CLASSIC) Optional.empty else Optional.of(true), member.upgraded)
- }
- val topicPartitionsByTopic = members.asScala.flatMap(_.assignment().topicPartitions().asScala).groupBy(_.topic())
- topicSet.foreach { topic =>
- val topicPartitions = topicPartitionsByTopic.getOrElse(topic, List.empty)
- assertEquals(testNumPartitions, topicPartitions.size)
- }
+ TestUtils.waitUntilTrue(() => {
+ val options = new ListConsumerGroupsOptions().withTypes(Set(groupType).asJava)
+ val matching = client.listConsumerGroups(options).all.get.asScala.filter(group =>
+ group.groupId == testGroupId &&
+ group.groupState.get == GroupState.STABLE)
+ matching.size == 1
+ }, s"Expected to be able to list $testGroupId in group type $groupType")
- val expectedOperations = AclEntry.supportedOperations(ResourceType.GROUP)
- assertEquals(expectedOperations, testGroupDescription.authorizedOperations())
-
- // Test that the fake group throws GroupIdNotFoundException
- assertTrue(describeWithFakeGroupResult.describedGroups().containsKey(fakeGroupId))
- assertFutureThrows(describeWithFakeGroupResult.describedGroups().get(fakeGroupId), classOf[GroupIdNotFoundException],
- s"Group $fakeGroupId not found.")
-
- // Test that all() also throws GroupIdNotFoundException
- assertFutureThrows(describeWithFakeGroupResult.all(), classOf[GroupIdNotFoundException],
- s"Group $fakeGroupId not found.")
-
- val testTopicPart0 = new TopicPartition(testTopicName, 0)
-
- // Test listConsumerGroupOffsets
- TestUtils.waitUntilTrue(() => {
- val parts = client.listConsumerGroupOffsets(testGroupId).partitionsToOffsetAndMetadata().get()
- parts.containsKey(testTopicPart0) && (parts.get(testTopicPart0).offset() == 1)
- }, s"Expected the offset for partition 0 to eventually become 1.")
-
- // Test listConsumerGroupOffsets with requireStable true
- val options = new ListConsumerGroupOffsetsOptions().requireStable(true)
- var parts = client.listConsumerGroupOffsets(testGroupId, options)
- .partitionsToOffsetAndMetadata().get()
- assertTrue(parts.containsKey(testTopicPart0))
- assertEquals(1, parts.get(testTopicPart0).offset())
-
- // Test listConsumerGroupOffsets with listConsumerGroupOffsetsSpec
- val groupSpecs = Collections.singletonMap(testGroupId,
- new ListConsumerGroupOffsetsSpec().topicPartitions(Collections.singleton(new TopicPartition(testTopicName, 0))))
- parts = client.listConsumerGroupOffsets(groupSpecs).partitionsToOffsetAndMetadata().get()
- assertTrue(parts.containsKey(testTopicPart0))
- assertEquals(1, parts.get(testTopicPart0).offset())
-
- // Test listConsumerGroupOffsets with listConsumerGroupOffsetsSpec and requireStable option
- parts = client.listConsumerGroupOffsets(groupSpecs, options).partitionsToOffsetAndMetadata().get()
- assertTrue(parts.containsKey(testTopicPart0))
- assertEquals(1, parts.get(testTopicPart0).offset())
-
- // Test delete non-exist consumer instance
- val invalidInstanceId = "invalid-instance-id"
- var removeMembersResult = client.removeMembersFromConsumerGroup(testGroupId, new RemoveMembersFromConsumerGroupOptions(
- Collections.singleton(new MemberToRemove(invalidInstanceId))
- ))
-
- assertFutureThrows(removeMembersResult.all, classOf[UnknownMemberIdException])
- val firstMemberFuture = removeMembersResult.memberResult(new MemberToRemove(invalidInstanceId))
- assertFutureThrows(firstMemberFuture, classOf[UnknownMemberIdException])
-
- // Test consumer group deletion
- var deleteResult = client.deleteConsumerGroups(Seq(testGroupId, fakeGroupId).asJava)
- assertEquals(2, deleteResult.deletedGroups().size())
-
- // Deleting the fake group ID should get GroupIdNotFoundException.
- assertTrue(deleteResult.deletedGroups().containsKey(fakeGroupId))
- assertFutureThrows(deleteResult.deletedGroups().get(fakeGroupId),
- classOf[GroupIdNotFoundException])
-
- // Deleting the real group ID should get GroupNotEmptyException
- assertTrue(deleteResult.deletedGroups().containsKey(testGroupId))
- assertFutureThrows(deleteResult.deletedGroups().get(testGroupId),
- classOf[GroupNotEmptyException])
-
- // Test delete one correct static member
- val removeOptions = new RemoveMembersFromConsumerGroupOptions(Collections.singleton(new MemberToRemove(testInstanceId1)))
- removeOptions.reason("test remove")
- removeMembersResult = client.removeMembersFromConsumerGroup(testGroupId, removeOptions)
-
- assertNull(removeMembersResult.all().get())
- val validMemberFuture = removeMembersResult.memberResult(new MemberToRemove(testInstanceId1))
- assertNull(validMemberFuture.get())
-
- val describeTestGroupResult = client.describeConsumerGroups(Seq(testGroupId).asJava,
- new DescribeConsumerGroupsOptions().includeAuthorizedOperations(true))
- assertEquals(1, describeTestGroupResult.describedGroups().size())
-
- testGroupDescription = describeTestGroupResult.describedGroups().get(testGroupId).get()
-
- assertEquals(testGroupId, testGroupDescription.groupId)
- assertFalse(testGroupDescription.isSimpleConsumerGroup)
- assertEquals(consumerSet.size - 1, testGroupDescription.members().size())
-
- // Delete all active members remaining (a static member + a dynamic member)
- removeMembersResult = client.removeMembersFromConsumerGroup(testGroupId, new RemoveMembersFromConsumerGroupOptions())
- assertNull(removeMembersResult.all().get())
-
- // The group should contain no members now.
- testGroupDescription = client.describeConsumerGroups(Seq(testGroupId).asJava,
- new DescribeConsumerGroupsOptions().includeAuthorizedOperations(true))
- .describedGroups().get(testGroupId).get()
- assertTrue(testGroupDescription.members().isEmpty)
-
- // Consumer group deletion on empty group should succeed
- deleteResult = client.deleteConsumerGroups(Seq(testGroupId).asJava)
- assertEquals(1, deleteResult.deletedGroups().size())
-
- assertTrue(deleteResult.deletedGroups().containsKey(testGroupId))
- assertNull(deleteResult.deletedGroups().get(testGroupId).get())
-
- // Test alterConsumerGroupOffsets
- val alterConsumerGroupOffsetsResult = client.alterConsumerGroupOffsets(testGroupId,
- Collections.singletonMap(testTopicPart0, new OffsetAndMetadata(0L)))
- assertNull(alterConsumerGroupOffsetsResult.all().get())
- assertNull(alterConsumerGroupOffsetsResult.partitionResult(testTopicPart0).get())
-
- // Verify alterConsumerGroupOffsets success
- TestUtils.waitUntilTrue(() => {
- val parts = client.listConsumerGroupOffsets(testGroupId).partitionsToOffsetAndMetadata().get()
- parts.containsKey(testTopicPart0) && (parts.get(testTopicPart0).offset() == 0)
- }, s"Expected the offset for partition 0 to eventually become 0.")
- } finally {
- consumerThreads.foreach {
- case consumerThread =>
- consumerThread.interrupt()
- consumerThread.join()
+ TestUtils.waitUntilTrue(() => {
+ val options = new ListConsumerGroupsOptions().withTypes(Set(groupType).asJava)
+ .inGroupStates(Set(GroupState.STABLE).asJava)
+ val matching = client.listConsumerGroups(options).all.get.asScala.filter(group =>
+ group.groupId == testGroupId &&
+ group.groupState.get == GroupState.STABLE)
+ matching.size == 1
+ }, s"Expected to be able to list $testGroupId in group type $groupType and state Stable")
+
+ TestUtils.waitUntilTrue(() => {
+ val options = new ListConsumerGroupsOptions().inGroupStates(Set(GroupState.STABLE).asJava)
+ val matching = client.listConsumerGroups(options).all.get.asScala.filter(group =>
+ group.groupId == testGroupId &&
+ group.groupState.get == GroupState.STABLE)
+ matching.size == 1
+ }, s"Expected to be able to list $testGroupId in state Stable")
+
+ TestUtils.waitUntilTrue(() => {
+ val options = new ListConsumerGroupsOptions().inGroupStates(Set(GroupState.EMPTY).asJava)
+ val matching = client.listConsumerGroups(options).all.get.asScala.filter(
+ _.groupId == testGroupId)
+ matching.isEmpty
+ }, s"Expected to find zero groups")
+
+ val describeWithFakeGroupResult = client.describeConsumerGroups(Seq(testGroupId, fakeGroupId).asJava,
+ new DescribeConsumerGroupsOptions().includeAuthorizedOperations(true))
+ assertEquals(2, describeWithFakeGroupResult.describedGroups().size())
+
+ // Test that we can get information about the test consumer group.
+ assertTrue(describeWithFakeGroupResult.describedGroups().containsKey(testGroupId))
+ val testGroupDescription = describeWithFakeGroupResult.describedGroups().get(testGroupId).get()
+ if (groupType == GroupType.CLASSIC) {
+ assertTrue(testGroupDescription.groupEpoch.isEmpty)
+ assertTrue(testGroupDescription.targetAssignmentEpoch.isEmpty)
+ } else {
+ assertEquals(Optional.of(3), testGroupDescription.groupEpoch)
+ assertEquals(Optional.of(3), testGroupDescription.targetAssignmentEpoch)
}
- }
+
+ assertEquals(testGroupId, testGroupDescription.groupId())
+ assertFalse(testGroupDescription.isSimpleConsumerGroup)
+ assertEquals(groupInstanceSet.size, testGroupDescription.members().size())
+ val members = testGroupDescription.members()
+ members.asScala.foreach { member =>
+ assertEquals(testClientId, member.clientId)
+ assertEquals(if (groupType == GroupType.CLASSIC) Optional.empty else Optional.of(true), member.upgraded)
+ }
+ val topicPartitionsByTopic = members.asScala.flatMap(_.assignment().topicPartitions().asScala).groupBy(_.topic())
+ topicSet.foreach { topic =>
+ val topicPartitions = topicPartitionsByTopic.getOrElse(topic, List.empty)
+ assertEquals(testNumPartitions, topicPartitions.size)
+ }
+
+ val expectedOperations = AclEntry.supportedOperations(ResourceType.GROUP)
+ assertEquals(expectedOperations, testGroupDescription.authorizedOperations())
+
+ // Test that the fake group throws GroupIdNotFoundException
+ assertTrue(describeWithFakeGroupResult.describedGroups().containsKey(fakeGroupId))
+ assertFutureThrows(describeWithFakeGroupResult.describedGroups().get(fakeGroupId), classOf[GroupIdNotFoundException],
+ s"Group $fakeGroupId not found.")
+
+ // Test that all() also throws GroupIdNotFoundException
+ assertFutureThrows(describeWithFakeGroupResult.all(), classOf[GroupIdNotFoundException],
+ s"Group $fakeGroupId not found.")
+
+ val testTopicPart0 = new TopicPartition(testTopicName, 0)
+
+ // Test listConsumerGroupOffsets
+ TestUtils.waitUntilTrue(() => {
+ val parts = client.listConsumerGroupOffsets(testGroupId).partitionsToOffsetAndMetadata().get()
+ parts.containsKey(testTopicPart0) && (parts.get(testTopicPart0).offset() == 1)
+ }, s"Expected the offset for partition 0 to eventually become 1.")
+
+ // Test listConsumerGroupOffsets with requireStable true
+ val options = new ListConsumerGroupOffsetsOptions().requireStable(true)
+ var parts = client.listConsumerGroupOffsets(testGroupId, options)
+ .partitionsToOffsetAndMetadata().get()
+ assertTrue(parts.containsKey(testTopicPart0))
+ assertEquals(1, parts.get(testTopicPart0).offset())
+
+ // Test listConsumerGroupOffsets with listConsumerGroupOffsetsSpec
+ val groupSpecs = Collections.singletonMap(testGroupId,
+ new ListConsumerGroupOffsetsSpec().topicPartitions(Collections.singleton(new TopicPartition(testTopicName, 0))))
+ parts = client.listConsumerGroupOffsets(groupSpecs).partitionsToOffsetAndMetadata().get()
+ assertTrue(parts.containsKey(testTopicPart0))
+ assertEquals(1, parts.get(testTopicPart0).offset())
+
+ // Test listConsumerGroupOffsets with listConsumerGroupOffsetsSpec and requireStable option
+ parts = client.listConsumerGroupOffsets(groupSpecs, options).partitionsToOffsetAndMetadata().get()
+ assertTrue(parts.containsKey(testTopicPart0))
+ assertEquals(1, parts.get(testTopicPart0).offset())
} finally {
- consumerSet.zip(groupInstanceSet).foreach(zipped => Utils.closeQuietly(zipped._1, zipped._2))
+ backgroundConsumerSet.close()
}
} finally {
Utils.closeQuietly(client, "adminClient")
@@ -2204,270 +2101,331 @@ class PlaintextAdminIntegrationTest extends BaseAdminIntegrationTest {
Utils.closeQuietly(producer, "producer")
}
- val EMPTY_GROUP_INSTANCE_ID = ""
val testGroupId = "test_group_id"
val testClientId = "test_client_id"
val testInstanceId1 = "test_instance_id_1"
val testInstanceId2 = "test_instance_id_2"
val fakeGroupId = "fake_group_id"
- def createProperties(groupInstanceId: String): Properties = {
- val newConsumerConfig = new Properties(consumerConfig)
- // We need to disable the auto commit because after the members got removed from group, the offset commit
- // will cause the member rejoining and the test will be flaky (check ConsumerCoordinator#OffsetCommitResponseHandler)
- newConsumerConfig.setProperty(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "false")
- newConsumerConfig.setProperty(ConsumerConfig.GROUP_ID_CONFIG, testGroupId)
- newConsumerConfig.setProperty(ConsumerConfig.CLIENT_ID_CONFIG, testClientId)
- if (groupInstanceId != EMPTY_GROUP_INSTANCE_ID) {
- newConsumerConfig.setProperty(ConsumerConfig.GROUP_INSTANCE_ID_CONFIG, groupInstanceId)
+ // contains two static members and one dynamic member
+ val groupInstanceSet = Set(testInstanceId1, testInstanceId2, "")
+ val topicSet = Set(testTopicName, testTopicName1, testTopicName2)
+
+ // We need to disable the auto commit because after the members got removed from group, the offset commit
+ // will cause the member rejoining and the test will be flaky (check ConsumerCoordinator#OffsetCommitResponseHandler)
+ val defaultConsumerConfig = new Properties(consumerConfig)
+ defaultConsumerConfig.setProperty(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "false")
+ defaultConsumerConfig.setProperty(ConsumerConfig.GROUP_ID_CONFIG, testGroupId)
+ defaultConsumerConfig.setProperty(ConsumerConfig.CLIENT_ID_CONFIG, testClientId)
+
+ val backgroundConsumerSet = new BackgroundConsumerSet(defaultConsumerConfig)
+ groupInstanceSet.zip(topicSet).foreach { case (groupInstanceId, topic) =>
+ val configOverrides = new Properties()
+ if (groupInstanceId != "") {
+ // static member
+ configOverrides.setProperty(ConsumerConfig.GROUP_INSTANCE_ID_CONFIG, groupInstanceId)
+ }
+ backgroundConsumerSet.addConsumer(topic, configOverrides)
+ }
+
+ try {
+ val groupType = if (groupProtocol.equalsIgnoreCase(GroupProtocol.CONSUMER.name)) GroupType.CONSUMER else GroupType.CLASSIC
+ // Start consumer polling threads in the background
+ backgroundConsumerSet.start()
+
+ // Test that we can list the new group.
+ TestUtils.waitUntilTrue(() => {
+ val matching = client.listConsumerGroups.all.get.asScala.filter(group =>
+ group.groupId == testGroupId &&
+ group.state.get == ConsumerGroupState.STABLE &&
+ group.groupState.get == GroupState.STABLE)
+ matching.size == 1
+ }, s"Expected to be able to list $testGroupId")
+
+ TestUtils.waitUntilTrue(() => {
+ val options = new ListConsumerGroupsOptions().withTypes(Set(groupType).asJava)
+ val matching = client.listConsumerGroups(options).all.get.asScala.filter(group =>
+ group.groupId == testGroupId &&
+ group.state.get == ConsumerGroupState.STABLE &&
+ group.groupState.get == GroupState.STABLE)
+ matching.size == 1
+ }, s"Expected to be able to list $testGroupId in group type $groupType")
+
+ TestUtils.waitUntilTrue(() => {
+ val options = new ListConsumerGroupsOptions().withTypes(Set(groupType).asJava)
+ .inStates(Set(ConsumerGroupState.STABLE).asJava)
+ val matching = client.listConsumerGroups(options).all.get.asScala.filter(group =>
+ group.groupId == testGroupId &&
+ group.state.get == ConsumerGroupState.STABLE &&
+ group.groupState.get == GroupState.STABLE)
+ matching.size == 1
+ }, s"Expected to be able to list $testGroupId in group type $groupType and state Stable")
+
+ TestUtils.waitUntilTrue(() => {
+ val options = new ListConsumerGroupsOptions().withTypes(Set(groupType).asJava)
+ .inGroupStates(Set(GroupState.STABLE).asJava)
+ val matching = client.listConsumerGroups(options).all.get.asScala.filter(group =>
+ group.groupId == testGroupId &&
+ group.state.get == ConsumerGroupState.STABLE &&
+ group.groupState.get == GroupState.STABLE)
+ matching.size == 1
+ }, s"Expected to be able to list $testGroupId in group type $groupType and state Stable")
+
+ TestUtils.waitUntilTrue(() => {
+ val options = new ListConsumerGroupsOptions().inStates(Set(ConsumerGroupState.STABLE).asJava)
+ val matching = client.listConsumerGroups(options).all.get.asScala.filter(group =>
+ group.groupId == testGroupId &&
+ group.state.get == ConsumerGroupState.STABLE &&
+ group.groupState.get == GroupState.STABLE)
+ matching.size == 1
+ }, s"Expected to be able to list $testGroupId in state Stable")
+
+ TestUtils.waitUntilTrue(() => {
+ val options = new ListConsumerGroupsOptions().inGroupStates(Set(GroupState.STABLE).asJava)
+ val matching = client.listConsumerGroups(options).all.get.asScala.filter(group =>
+ group.groupId == testGroupId &&
+ group.state.get == ConsumerGroupState.STABLE &&
+ group.groupState.get == GroupState.STABLE)
+ matching.size == 1
+ }, s"Expected to be able to list $testGroupId in state Stable")
+
+ TestUtils.waitUntilTrue(() => {
+ val options = new ListConsumerGroupsOptions().inStates(Set(ConsumerGroupState.EMPTY).asJava)
+ val matching = client.listConsumerGroups(options).all.get.asScala.filter(
+ _.groupId == testGroupId)
+ matching.isEmpty
+ }, s"Expected to find zero groups")
+
+ TestUtils.waitUntilTrue(() => {
+ val options = new ListConsumerGroupsOptions().inGroupStates(Set(GroupState.EMPTY).asJava)
+ val matching = client.listConsumerGroups(options).all.get.asScala.filter(
+ _.groupId == testGroupId)
+ matching.isEmpty
+ }, s"Expected to find zero groups")
+
+ val describeWithFakeGroupResult = client.describeConsumerGroups(Seq(testGroupId, fakeGroupId).asJava,
+ new DescribeConsumerGroupsOptions().includeAuthorizedOperations(true))
+ assertEquals(2, describeWithFakeGroupResult.describedGroups().size())
+
+ // Test that we can get information about the test consumer group.
+ assertTrue(describeWithFakeGroupResult.describedGroups().containsKey(testGroupId))
+ val testGroupDescription = describeWithFakeGroupResult.describedGroups().get(testGroupId).get()
+
+ assertEquals(testGroupId, testGroupDescription.groupId())
+ assertFalse(testGroupDescription.isSimpleConsumerGroup)
+ assertEquals(groupInstanceSet.size, testGroupDescription.members().size())
+ val members = testGroupDescription.members()
+ members.asScala.foreach(member => assertEquals(testClientId, member.clientId()))
+ val topicPartitionsByTopic = members.asScala.flatMap(_.assignment().topicPartitions().asScala).groupBy(_.topic())
+ topicSet.foreach { topic =>
+ val topicPartitions = topicPartitionsByTopic.getOrElse(topic, List.empty)
+ assertEquals(testNumPartitions, topicPartitions.size)
}
- newConsumerConfig
+
+ val expectedOperations = AclEntry.supportedOperations(ResourceType.GROUP)
+ assertEquals(expectedOperations, testGroupDescription.authorizedOperations())
+
+ // Test that the fake group throws GroupIdNotFoundException
+ assertTrue(describeWithFakeGroupResult.describedGroups().containsKey(fakeGroupId))
+ assertFutureThrows(describeWithFakeGroupResult.describedGroups().get(fakeGroupId),
+ classOf[GroupIdNotFoundException], s"Group $fakeGroupId not found.")
+
+ // Test that all() also throws GroupIdNotFoundException
+ assertFutureThrows(describeWithFakeGroupResult.all(),
+ classOf[GroupIdNotFoundException], s"Group $fakeGroupId not found.")
+
+ val testTopicPart0 = new TopicPartition(testTopicName, 0)
+
+ // Test listConsumerGroupOffsets
+ TestUtils.waitUntilTrue(() => {
+ val parts = client.listConsumerGroupOffsets(testGroupId).partitionsToOffsetAndMetadata().get()
+ parts.containsKey(testTopicPart0) && (parts.get(testTopicPart0).offset() == 1)
+ }, s"Expected the offset for partition 0 to eventually become 1.")
+
+ // Test listConsumerGroupOffsets with requireStable true
+ val options = new ListConsumerGroupOffsetsOptions().requireStable(true)
+ var parts = client.listConsumerGroupOffsets(testGroupId, options)
+ .partitionsToOffsetAndMetadata().get()
+ assertTrue(parts.containsKey(testTopicPart0))
+ assertEquals(1, parts.get(testTopicPart0).offset())
+
+ // Test listConsumerGroupOffsets with listConsumerGroupOffsetsSpec
+ val groupSpecs = Collections.singletonMap(testGroupId,
+ new ListConsumerGroupOffsetsSpec().topicPartitions(Collections.singleton(new TopicPartition(testTopicName, 0))))
+ parts = client.listConsumerGroupOffsets(groupSpecs).partitionsToOffsetAndMetadata().get()
+ assertTrue(parts.containsKey(testTopicPart0))
+ assertEquals(1, parts.get(testTopicPart0).offset())
+
+ // Test listConsumerGroupOffsets with listConsumerGroupOffsetsSpec and requireStable option
+ parts = client.listConsumerGroupOffsets(groupSpecs, options).partitionsToOffsetAndMetadata().get()
+ assertTrue(parts.containsKey(testTopicPart0))
+ assertEquals(1, parts.get(testTopicPart0).offset())
+ } finally {
+ backgroundConsumerSet.close()
}
+ } finally {
+ Utils.closeQuietly(client, "adminClient")
+ }
+ }
+
+
+ /**
+ * Test the consumer group APIs for member removal.
+ */
+ @ParameterizedTest(name = TestInfoUtils.TestWithParameterizedQuorumAndGroupProtocolNames)
+ @MethodSource(Array("getTestQuorumAndGroupProtocolParametersAll"))
+ def testConsumerGroupWithMemberRemoval(quorum: String, groupProtocol: String): Unit = {
+ val config = createConfig
+ client = Admin.create(config)
+ try {
+ // Verify that initially there are no consumer groups to list.
+ val list1 = client.listConsumerGroups()
+ assertEquals(0, list1.all().get().size())
+ assertEquals(0, list1.errors().get().size())
+ assertEquals(0, list1.valid().get().size())
+ val testTopicName = "test_topic"
+ val testTopicName1 = testTopicName + "1"
+ val testTopicName2 = testTopicName + "2"
+ val testNumPartitions = 2
+
+ client.createTopics(util.Arrays.asList(
+ new NewTopic(testTopicName, testNumPartitions, 1.toShort),
+ new NewTopic(testTopicName1, testNumPartitions, 1.toShort),
+ new NewTopic(testTopicName2, testNumPartitions, 1.toShort)
+ )).all().get()
+ waitForTopics(client, List(testTopicName, testTopicName1, testTopicName2), List())
+
+ val producer = createProducer()
+ try {
+ producer.send(new ProducerRecord(testTopicName, 0, null, null)).get()
+ } finally {
+ Utils.closeQuietly(producer, "producer")
+ }
+
+ val testGroupId = "test_group_id"
+ val testClientId = "test_client_id"
+ val testInstanceId1 = "test_instance_id_1"
+ val testInstanceId2 = "test_instance_id_2"
+ val fakeGroupId = "fake_group_id"
// contains two static members and one dynamic member
- val groupInstanceSet = Set(testInstanceId1, testInstanceId2, EMPTY_GROUP_INSTANCE_ID)
- val consumerSet = groupInstanceSet.map { groupInstanceId => createConsumer(configOverrides = createProperties(groupInstanceId))}
+ val groupInstanceSet = Set(testInstanceId1, testInstanceId2, "")
val topicSet = Set(testTopicName, testTopicName1, testTopicName2)
- val latch = new CountDownLatch(consumerSet.size)
+ // We need to disable the auto commit because after the members got removed from group, the offset commit
+ // will cause the member rejoining and the test will be flaky (check ConsumerCoordinator#OffsetCommitResponseHandler)
+ val defaultConsumerConfig = new Properties(consumerConfig)
+ defaultConsumerConfig.setProperty(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "false")
+ defaultConsumerConfig.setProperty(ConsumerConfig.GROUP_ID_CONFIG, testGroupId)
+ defaultConsumerConfig.setProperty(ConsumerConfig.CLIENT_ID_CONFIG, testClientId)
+ // We need to set internal.leave.group.on.close to validate dynamic member removal, but it only works for ClassicConsumer
+ // After KIP-1092, we can control dynamic member removal for both ClassicConsumer and AsyncConsumer
+ defaultConsumerConfig.setProperty("internal.leave.group.on.close", "false")
+
+ val backgroundConsumerSet = new BackgroundConsumerSet(defaultConsumerConfig)
+ groupInstanceSet.zip(topicSet).foreach { case (groupInstanceId, topic) =>
+ val configOverrides = new Properties()
+ if (groupInstanceId != "") {
+ // static member
+ configOverrides.setProperty(ConsumerConfig.GROUP_INSTANCE_ID_CONFIG, groupInstanceId)
+ }
+ backgroundConsumerSet.addConsumer(topic, configOverrides)
+ }
+
try {
- def createConsumerThread[K,V](consumer: Consumer[K,V], topic: String): Thread = {
- new Thread {
- override def run : Unit = {
- consumer.subscribe(Collections.singleton(topic))
- try {
- while (true) {
- consumer.poll(JDuration.ofSeconds(5))
- if (!consumer.assignment.isEmpty && latch.getCount > 0L)
- latch.countDown()
- try {
- consumer.commitSync()
- } catch {
- case _: CommitFailedException => // Ignore and retry on next iteration.
- }
- }
- } catch {
- case _: InterruptException => // Suppress the output to stderr
- }
- }
- }
+ // Start consumer polling threads in the background
+ backgroundConsumerSet.start()
+
+ // Test delete non-exist consumer instance
+ val invalidInstanceId = "invalid-instance-id"
+ var removeMembersResult = client.removeMembersFromConsumerGroup(testGroupId, new RemoveMembersFromConsumerGroupOptions(
+ Collections.singleton(new MemberToRemove(invalidInstanceId))
+ ))
+ assertFutureThrows(removeMembersResult.all, classOf[UnknownMemberIdException])
+ val firstMemberFuture = removeMembersResult.memberResult(new MemberToRemove(invalidInstanceId))
+ assertFutureThrows(firstMemberFuture, classOf[UnknownMemberIdException])
+
+ // Test consumer group deletion
+ var deleteResult = client.deleteConsumerGroups(Seq(testGroupId, fakeGroupId).asJava)
+ assertEquals(2, deleteResult.deletedGroups().size())
+
+ // Deleting the fake group ID should get GroupIdNotFoundException.
+ assertTrue(deleteResult.deletedGroups().containsKey(fakeGroupId))
+ assertFutureThrows(deleteResult.deletedGroups().get(fakeGroupId),
+ classOf[GroupIdNotFoundException])
+
+ // Deleting the real group ID should get GroupNotEmptyException
+ assertTrue(deleteResult.deletedGroups().containsKey(testGroupId))
+ assertFutureThrows(deleteResult.deletedGroups().get(testGroupId),
+ classOf[GroupNotEmptyException])
+
+ // Stop the consumer threads and close consumers to prevent member rejoining.
+ backgroundConsumerSet.stop()
+
+ // Check the members in the group after consumers have stopped
+ var describeTestGroupResult = client.describeConsumerGroups(Seq(testGroupId).asJava,
+ new DescribeConsumerGroupsOptions().includeAuthorizedOperations(true))
+ assertEquals(1, describeTestGroupResult.describedGroups().size())
+
+ var testGroupDescription = describeTestGroupResult.describedGroups().get(testGroupId).get()
+ assertEquals(testGroupId, testGroupDescription.groupId)
+ assertFalse(testGroupDescription.isSimpleConsumerGroup)
+
+ // Although we set `internal.leave.group.on.close` in the consumer, it only works for ClassicConsumer.
+ // After KIP-1092, we can control dynamic member removal in consumer.close()
+ if (groupProtocol == GroupProtocol.CLASSIC.name.toLowerCase(Locale.ROOT)) {
+ assertEquals(3, testGroupDescription.members().size())
+ } else if (groupProtocol == GroupProtocol.CONSUMER.name.toLowerCase(Locale.ROOT)) {
+ assertEquals(2, testGroupDescription.members().size())
}
- // Start consumers in a thread that will subscribe to a new group.
- val consumerThreads = consumerSet.zip(topicSet).map(zipped => createConsumerThread(zipped._1, zipped._2))
- val groupType = if (groupProtocol.equalsIgnoreCase(GroupProtocol.CONSUMER.name)) GroupType.CONSUMER else GroupType.CLASSIC
+ // Test delete one static member
+ removeMembersResult = client.removeMembersFromConsumerGroup(testGroupId,
+ new RemoveMembersFromConsumerGroupOptions(Collections.singleton(new MemberToRemove(testInstanceId1))))
- try {
- consumerThreads.foreach(_.start())
- assertTrue(latch.await(30000, TimeUnit.MILLISECONDS))
- // Test that we can list the new group.
- TestUtils.waitUntilTrue(() => {
- val matching = client.listConsumerGroups.all.get.asScala.filter(group =>
- group.groupId == testGroupId &&
- group.state.get == ConsumerGroupState.STABLE &&
- group.groupState.get == GroupState.STABLE)
- matching.size == 1
- }, s"Expected to be able to list $testGroupId")
-
- TestUtils.waitUntilTrue(() => {
- val options = new ListConsumerGroupsOptions().withTypes(Set(groupType).asJava)
- val matching = client.listConsumerGroups(options).all.get.asScala.filter(group =>
- group.groupId == testGroupId &&
- group.state.get == ConsumerGroupState.STABLE &&
- group.groupState.get == GroupState.STABLE)
- matching.size == 1
- }, s"Expected to be able to list $testGroupId in group type $groupType")
-
- TestUtils.waitUntilTrue(() => {
- val options = new ListConsumerGroupsOptions().withTypes(Set(groupType).asJava)
- .inStates(Set(ConsumerGroupState.STABLE).asJava)
- val matching = client.listConsumerGroups(options).all.get.asScala.filter(group =>
- group.groupId == testGroupId &&
- group.state.get == ConsumerGroupState.STABLE &&
- group.groupState.get == GroupState.STABLE)
- matching.size == 1
- }, s"Expected to be able to list $testGroupId in group type $groupType and state Stable")
-
- TestUtils.waitUntilTrue(() => {
- val options = new ListConsumerGroupsOptions().withTypes(Set(groupType).asJava)
- .inGroupStates(Set(GroupState.STABLE).asJava)
- val matching = client.listConsumerGroups(options).all.get.asScala.filter(group =>
- group.groupId == testGroupId &&
- group.state.get == ConsumerGroupState.STABLE &&
- group.groupState.get == GroupState.STABLE)
- matching.size == 1
- }, s"Expected to be able to list $testGroupId in group type $groupType and state Stable")
-
- TestUtils.waitUntilTrue(() => {
- val options = new ListConsumerGroupsOptions().inStates(Set(ConsumerGroupState.STABLE).asJava)
- val matching = client.listConsumerGroups(options).all.get.asScala.filter(group =>
- group.groupId == testGroupId &&
- group.state.get == ConsumerGroupState.STABLE &&
- group.groupState.get == GroupState.STABLE)
- matching.size == 1
- }, s"Expected to be able to list $testGroupId in state Stable")
-
- TestUtils.waitUntilTrue(() => {
- val options = new ListConsumerGroupsOptions().inGroupStates(Set(GroupState.STABLE).asJava)
- val matching = client.listConsumerGroups(options).all.get.asScala.filter(group =>
- group.groupId == testGroupId &&
- group.state.get == ConsumerGroupState.STABLE &&
- group.groupState.get == GroupState.STABLE)
- matching.size == 1
- }, s"Expected to be able to list $testGroupId in state Stable")
-
- TestUtils.waitUntilTrue(() => {
- val options = new ListConsumerGroupsOptions().inStates(Set(ConsumerGroupState.EMPTY).asJava)
- val matching = client.listConsumerGroups(options).all.get.asScala.filter(
- _.groupId == testGroupId)
- matching.isEmpty
- }, s"Expected to find zero groups")
+ assertNull(removeMembersResult.all().get())
+ assertNull(removeMembersResult.memberResult(new MemberToRemove(testInstanceId1)).get())
- TestUtils.waitUntilTrue(() => {
- val options = new ListConsumerGroupsOptions().inGroupStates(Set(GroupState.EMPTY).asJava)
- val matching = client.listConsumerGroups(options).all.get.asScala.filter(
- _.groupId == testGroupId)
- matching.isEmpty
- }, s"Expected to find zero groups")
-
- val describeWithFakeGroupResult = client.describeConsumerGroups(Seq(testGroupId, fakeGroupId).asJava,
- new DescribeConsumerGroupsOptions().includeAuthorizedOperations(true))
- assertEquals(2, describeWithFakeGroupResult.describedGroups().size())
-
- // Test that we can get information about the test consumer group.
- assertTrue(describeWithFakeGroupResult.describedGroups().containsKey(testGroupId))
- var testGroupDescription = describeWithFakeGroupResult.describedGroups().get(testGroupId).get()
-
- assertEquals(testGroupId, testGroupDescription.groupId())
- assertFalse(testGroupDescription.isSimpleConsumerGroup)
- assertEquals(groupInstanceSet.size, testGroupDescription.members().size())
- val members = testGroupDescription.members()
- members.asScala.foreach(member => assertEquals(testClientId, member.clientId()))
- val topicPartitionsByTopic = members.asScala.flatMap(_.assignment().topicPartitions().asScala).groupBy(_.topic())
- topicSet.foreach { topic =>
- val topicPartitions = topicPartitionsByTopic.getOrElse(topic, List.empty)
- assertEquals(testNumPartitions, topicPartitions.size)
- }
+ describeTestGroupResult = client.describeConsumerGroups(Seq(testGroupId).asJava,
+ new DescribeConsumerGroupsOptions().includeAuthorizedOperations(true))
+ testGroupDescription = describeTestGroupResult.describedGroups().get(testGroupId).get()
- val expectedOperations = AclEntry.supportedOperations(ResourceType.GROUP)
- assertEquals(expectedOperations, testGroupDescription.authorizedOperations())
-
- // Test that the fake group throws GroupIdNotFoundException
- assertTrue(describeWithFakeGroupResult.describedGroups().containsKey(fakeGroupId))
- assertFutureThrows(describeWithFakeGroupResult.describedGroups().get(fakeGroupId),
- classOf[GroupIdNotFoundException], s"Group $fakeGroupId not found.")
-
- // Test that all() also throws GroupIdNotFoundException
- assertFutureThrows(describeWithFakeGroupResult.all(),
- classOf[GroupIdNotFoundException], s"Group $fakeGroupId not found.")
-
- val testTopicPart0 = new TopicPartition(testTopicName, 0)
-
- // Test listConsumerGroupOffsets
- TestUtils.waitUntilTrue(() => {
- val parts = client.listConsumerGroupOffsets(testGroupId).partitionsToOffsetAndMetadata().get()
- parts.containsKey(testTopicPart0) && (parts.get(testTopicPart0).offset() == 1)
- }, s"Expected the offset for partition 0 to eventually become 1.")
-
- // Test listConsumerGroupOffsets with requireStable true
- val options = new ListConsumerGroupOffsetsOptions().requireStable(true)
- var parts = client.listConsumerGroupOffsets(testGroupId, options)
- .partitionsToOffsetAndMetadata().get()
- assertTrue(parts.containsKey(testTopicPart0))
- assertEquals(1, parts.get(testTopicPart0).offset())
-
- // Test listConsumerGroupOffsets with listConsumerGroupOffsetsSpec
- val groupSpecs = Collections.singletonMap(testGroupId,
- new ListConsumerGroupOffsetsSpec().topicPartitions(Collections.singleton(new TopicPartition(testTopicName, 0))))
- parts = client.listConsumerGroupOffsets(groupSpecs).partitionsToOffsetAndMetadata().get()
- assertTrue(parts.containsKey(testTopicPart0))
- assertEquals(1, parts.get(testTopicPart0).offset())
-
- // Test listConsumerGroupOffsets with listConsumerGroupOffsetsSpec and requireStable option
- parts = client.listConsumerGroupOffsets(groupSpecs, options).partitionsToOffsetAndMetadata().get()
- assertTrue(parts.containsKey(testTopicPart0))
- assertEquals(1, parts.get(testTopicPart0).offset())
-
- // Test delete non-exist consumer instance
- val invalidInstanceId = "invalid-instance-id"
- var removeMembersResult = client.removeMembersFromConsumerGroup(testGroupId, new RemoveMembersFromConsumerGroupOptions(
- Collections.singleton(new MemberToRemove(invalidInstanceId))
- ))
-
- assertFutureThrows(removeMembersResult.all, classOf[UnknownMemberIdException])
- val firstMemberFuture = removeMembersResult.memberResult(new MemberToRemove(invalidInstanceId))
- assertFutureThrows(firstMemberFuture, classOf[UnknownMemberIdException])
-
- // Test consumer group deletion
- var deleteResult = client.deleteConsumerGroups(Seq(testGroupId, fakeGroupId).asJava)
- assertEquals(2, deleteResult.deletedGroups().size())
-
- // Deleting the fake group ID should get GroupIdNotFoundException.
- assertTrue(deleteResult.deletedGroups().containsKey(fakeGroupId))
- assertFutureThrows(deleteResult.deletedGroups().get(fakeGroupId),
- classOf[GroupIdNotFoundException])
-
- // Deleting the real group ID should get GroupNotEmptyException
- assertTrue(deleteResult.deletedGroups().containsKey(testGroupId))
- assertFutureThrows(deleteResult.deletedGroups().get(testGroupId),
- classOf[GroupNotEmptyException])
-
- // Test delete one correct static member
- val removeOptions = new RemoveMembersFromConsumerGroupOptions(Collections.singleton(new MemberToRemove(testInstanceId1)))
- removeOptions.reason("test remove")
- removeMembersResult = client.removeMembersFromConsumerGroup(testGroupId, removeOptions)
-
- assertNull(removeMembersResult.all().get())
- val validMemberFuture = removeMembersResult.memberResult(new MemberToRemove(testInstanceId1))
- assertNull(validMemberFuture.get())
-
- val describeTestGroupResult = client.describeConsumerGroups(Seq(testGroupId).asJava,
- new DescribeConsumerGroupsOptions().includeAuthorizedOperations(true))
- assertEquals(1, describeTestGroupResult.describedGroups().size())
-
- testGroupDescription = describeTestGroupResult.describedGroups().get(testGroupId).get()
-
- assertEquals(testGroupId, testGroupDescription.groupId)
- assertFalse(testGroupDescription.isSimpleConsumerGroup)
- assertEquals(consumerSet.size - 1, testGroupDescription.members().size())
-
- // Delete all active members remaining (a static member + a dynamic member)
- removeMembersResult = client.removeMembersFromConsumerGroup(testGroupId, new RemoveMembersFromConsumerGroupOptions())
- assertNull(removeMembersResult.all().get())
-
- // The group should contain no members now.
- testGroupDescription = client.describeConsumerGroups(Seq(testGroupId).asJava,
- new DescribeConsumerGroupsOptions().includeAuthorizedOperations(true))
- .describedGroups().get(testGroupId).get()
- assertTrue(testGroupDescription.members().isEmpty)
-
- // Consumer group deletion on empty group should succeed
- deleteResult = client.deleteConsumerGroups(Seq(testGroupId).asJava)
- assertEquals(1, deleteResult.deletedGroups().size())
-
- assertTrue(deleteResult.deletedGroups().containsKey(testGroupId))
- assertNull(deleteResult.deletedGroups().get(testGroupId).get())
-
- // Test alterConsumerGroupOffsets
- val alterConsumerGroupOffsetsResult = client.alterConsumerGroupOffsets(testGroupId,
- Collections.singletonMap(testTopicPart0, new OffsetAndMetadata(0L)))
- assertNull(alterConsumerGroupOffsetsResult.all().get())
- assertNull(alterConsumerGroupOffsetsResult.partitionResult(testTopicPart0).get())
-
- // Verify alterConsumerGroupOffsets success
- TestUtils.waitUntilTrue(() => {
- val parts = client.listConsumerGroupOffsets(testGroupId).partitionsToOffsetAndMetadata().get()
- parts.containsKey(testTopicPart0) && (parts.get(testTopicPart0).offset() == 0)
- }, s"Expected the offset for partition 0 to eventually become 0.")
- } finally {
- consumerThreads.foreach {
- case consumerThread =>
- consumerThread.interrupt()
- consumerThread.join()
- }
+ if (groupProtocol == GroupProtocol.CLASSIC.name.toLowerCase(Locale.ROOT)) {
+ assertEquals(2, testGroupDescription.members().size())
+ } else if (groupProtocol == GroupProtocol.CONSUMER.name.toLowerCase(Locale.ROOT)) {
+ assertEquals(1, testGroupDescription.members().size())
}
+
+ // Delete all active members remaining
+ removeMembersResult = client.removeMembersFromConsumerGroup(testGroupId, new RemoveMembersFromConsumerGroupOptions())
+ assertNull(removeMembersResult.all().get())
+
+ // The group should contain no members now.
+ testGroupDescription = client.describeConsumerGroups(Seq(testGroupId).asJava,
+ new DescribeConsumerGroupsOptions().includeAuthorizedOperations(true)).describedGroups().get(testGroupId).get()
+ assertTrue(testGroupDescription.members().isEmpty)
+
+ // Consumer group deletion on empty group should succeed
+ deleteResult = client.deleteConsumerGroups(Seq(testGroupId).asJava)
+ assertEquals(1, deleteResult.deletedGroups().size())
+
+ assertTrue(deleteResult.deletedGroups().containsKey(testGroupId))
+ assertNull(deleteResult.deletedGroups().get(testGroupId).get())
+
+ // Test alterConsumerGroupOffsets when group is empty
+ val testTopicPart0 = new TopicPartition(testTopicName, 0)
+ val alterConsumerGroupOffsetsResult = client.alterConsumerGroupOffsets(testGroupId,
+ Collections.singletonMap(testTopicPart0, new OffsetAndMetadata(0L)))
+ assertNull(alterConsumerGroupOffsetsResult.all().get())
+ assertNull(alterConsumerGroupOffsetsResult.partitionResult(testTopicPart0).get())
+
+ // Verify alterConsumerGroupOffsets success
+ TestUtils.waitUntilTrue(() => {
+ val parts = client.listConsumerGroupOffsets(testGroupId).partitionsToOffsetAndMetadata().get()
+ parts.containsKey(testTopicPart0) && (parts.get(testTopicPart0).offset() == 0)
+ }, s"Expected the offset for partition 0 to eventually become 0.")
} finally {
- consumerSet.zip(groupInstanceSet).foreach(zipped => Utils.closeQuietly(zipped._1, zipped._2))
+ backgroundConsumerSet.close()
}
} finally {
Utils.closeQuietly(client, "adminClient")
@@ -4003,6 +3961,70 @@ class PlaintextAdminIntegrationTest extends BaseAdminIntegrationTest {
ConfigSource.DYNAMIC_TOPIC_CONFIG, false, false, Collections.emptyList(), null, null),
topicConfigs.get(TopicConfig.INDEX_INTERVAL_BYTES_CONFIG))
}
+
+ class BackgroundConsumerSet(defaultConsumerConfig: Properties) {
+ private val consumerSet: scala.collection.mutable.Set[Consumer[Array[Byte], Array[Byte]]] = scala.collection.mutable.Set.empty
+ private val consumerThreads: scala.collection.mutable.Set[Thread] = scala.collection.mutable.Set.empty
+ private var startLatch: CountDownLatch = new CountDownLatch(0)
+ private var stopLatch: CountDownLatch = new CountDownLatch(0)
+ private var consumerThreadRunning = new AtomicBoolean(false)
+
+ def addConsumer(topic: String, configOverrides: Properties = new Properties()): Unit = {
+ val newConsumerConfig = defaultConsumerConfig.clone().asInstanceOf[Properties]
+ newConsumerConfig.putAll(configOverrides)
+
+ val consumer = createConsumer(configOverrides = newConsumerConfig)
+ val consumerThread = createConsumerThread(consumer, topic)
+ consumerSet.add(consumer)
+ consumerThreads.add(consumerThread)
+ }
+
+ def start(): Unit = {
+ startLatch = new CountDownLatch(consumerSet.size)
+ stopLatch = new CountDownLatch(consumerSet.size)
+ consumerThreadRunning = new AtomicBoolean(true)
+ consumerThreads.foreach(_.start())
+ assertTrue(startLatch.await(30000, TimeUnit.MILLISECONDS), "Failed to start consumer threads in time")
+ }
+
+ def stop(): Unit = {
+ consumerSet.foreach(_.wakeup())
+ consumerThreadRunning.set(false)
+ assertTrue(stopLatch.await(30000, TimeUnit.MILLISECONDS), "Failed to stop consumer threads in time")
+ }
+
+ def close(): Unit = {
+ // stop the consumers and wait for consumer threads stopped
+ stop()
+ consumerThreads.foreach(_.join())
+ }
+
+ private def createConsumerThread[K,V](consumer: Consumer[K,V], topic: String): Thread = {
+ new Thread {
+ override def run : Unit = {
+ consumer.subscribe(Collections.singleton(topic))
+ try {
+ while (consumerThreadRunning.get()) {
+ consumer.poll(JDuration.ofSeconds(5))
+ if (!consumer.assignment.isEmpty && startLatch.getCount > 0L)
+ startLatch.countDown()
+ try {
+ consumer.commitSync()
+ } catch {
+ case _: CommitFailedException => // Ignore and retry on next iteration.
+ }
+ }
+ } catch {
+ case _: WakeupException => // ignore
+ case _: InterruptException => // ignore
+ } finally {
+ consumer.close()
+ stopLatch.countDown()
+ }
+ }
+ }
+ }
+ }
}
object PlaintextAdminIntegrationTest {