aliehsaeedii commented on code in PR #20244:
URL: https://github.com/apache/kafka/pull/20244#discussion_r2251643576
##########
core/src/test/scala/integration/kafka/api/PlaintextAdminIntegrationTest.scala:
##########
@@ -4363,6 +4364,170 @@ class PlaintextAdminIntegrationTest extends
BaseAdminIntegrationTest {
}
}
}
+
+ @ParameterizedTest(name =
TestInfoUtils.TestWithParameterizedGroupProtocolNames)
+
@MethodSource(Array("getTestGroupProtocolParametersStreamsGroupProtocolOnly"))
+ def testDescribeStreamsGroups(groupProtocol: String): Unit = {
+ val streamsGroupId = "stream_group_id"
+ val testTopicName = "test_topic"
+ val testOutputTopicName = "test_output_topic"
+ val testNumPartitions = 1
+
+ val config = createConfig
+ client = Admin.create(config)
+
+ prepareTopics(List(testTopicName, testOutputTopicName), testNumPartitions)
+ prepareRecords(testTopicName)
+
+ val streamsConfig = new Properties()
+ streamsConfig.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest")
+ streamsConfig.put(StreamsConfig.COMMIT_INTERVAL_MS_CONFIG, 10 * 1000)
+ val streams = createStreamsGroup(
+ configOverrides = streamsConfig,
+ inputTopic = testTopicName,
+ outputTopic = testOutputTopicName,
+ streamsGroupId = streamsGroupId,
+ groupProtocol = groupProtocol
+ )
+
+ try {
+ streams.cleanUp()
+ streams.start()
+
+ TestUtils.waitUntilTrue(() => streams.state() ==
KafkaStreams.State.RUNNING, "Streams not in RUNNING state")
+
+ TestUtils.waitUntilTrue(() => {
+ client.listGroups().all().get().stream()
+ .anyMatch(g => g.groupId() == streamsGroupId)
+ }, "Streams group not ready to describe yet")
+
+ Thread.sleep(10000)
+
+ TestUtils.waitUntilTrue(() => {
+ try {
+ val describedGroups =
client.describeStreamsGroups(util.List.of(streamsGroupId)).all().get()
+ val group = describedGroups.get(streamsGroupId)
+ if (group != null) {
+ group.groupState() == GroupState.STABLE &&
!group.subtopologies().isEmpty
+ } else {
+ false
+ }
+ } catch {
+ case _: Exception => false
+ }
+ }, "Stream group not fully initialized with topology")
+
Review Comment:
Could you please test describing `non-existent` group as well as you did in
delete group test?
##########
core/src/test/scala/integration/kafka/api/PlaintextAdminIntegrationTest.scala:
##########
@@ -4363,6 +4401,158 @@ class PlaintextAdminIntegrationTest extends
BaseAdminIntegrationTest {
}
}
}
+
+ @ParameterizedTest(name =
TestInfoUtils.TestWithParameterizedGroupProtocolNames)
+
@MethodSource(Array("getTestGroupProtocolParametersStreamsGroupProtocolOnly"))
+ def testDescribeStreamsGroups(groupProtocol: String): Unit = {
+ val streamsGroupId = "stream_group_id"
+ val testTopicName = "test_topic"
+ val testOutputTopicName = "test_output_topic"
+ val testNumPartitions = 1
+
+ val config = createConfig
+ client = Admin.create(config)
+
+ prepareTopics(List(testTopicName, testOutputTopicName), testNumPartitions)
+ prepareRecords(testTopicName)
+
+ val streamsConfig = new Properties()
+ streamsConfig.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest")
+ streamsConfig.put(StreamsConfig.COMMIT_INTERVAL_MS_CONFIG, 10 * 1000)
+ val streams = createStreamsGroup(
+ configOverrides = streamsConfig,
+ inputTopic = testTopicName,
+ outputTopic = testOutputTopicName,
+ streamsGroupId = streamsGroupId,
+ groupProtocol = groupProtocol
+ )
+
+ try {
+ streams.cleanUp()
+ streams.start()
+
+ TestUtils.waitUntilTrue(() => streams.state() ==
KafkaStreams.State.RUNNING, "Streams not in RUNNING state")
+
+ TestUtils.waitUntilTrue(() => {
+ client.listGroups().all().get().stream()
+ .anyMatch(g => g.groupId() == streamsGroupId)
+ }, "Streams group not ready to describe yet")
Review Comment:
Could we remove this part since we are checking the state below?
##########
core/src/test/scala/integration/kafka/api/PlaintextAdminIntegrationTest.scala:
##########
@@ -4363,6 +4364,170 @@ class PlaintextAdminIntegrationTest extends
BaseAdminIntegrationTest {
}
}
}
+
+ @ParameterizedTest(name =
TestInfoUtils.TestWithParameterizedGroupProtocolNames)
+
@MethodSource(Array("getTestGroupProtocolParametersStreamsGroupProtocolOnly"))
+ def testDescribeStreamsGroups(groupProtocol: String): Unit = {
+ val streamsGroupId = "stream_group_id"
+ val testTopicName = "test_topic"
+ val testOutputTopicName = "test_output_topic"
+ val testNumPartitions = 1
+
+ val config = createConfig
+ client = Admin.create(config)
+
+ prepareTopics(List(testTopicName, testOutputTopicName), testNumPartitions)
+ prepareRecords(testTopicName)
+
+ val streamsConfig = new Properties()
+ streamsConfig.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest")
+ streamsConfig.put(StreamsConfig.COMMIT_INTERVAL_MS_CONFIG, 10 * 1000)
+ val streams = createStreamsGroup(
+ configOverrides = streamsConfig,
+ inputTopic = testTopicName,
+ outputTopic = testOutputTopicName,
+ streamsGroupId = streamsGroupId,
+ groupProtocol = groupProtocol
+ )
+
+ try {
+ streams.cleanUp()
+ streams.start()
+
+ TestUtils.waitUntilTrue(() => streams.state() ==
KafkaStreams.State.RUNNING, "Streams not in RUNNING state")
+
+ TestUtils.waitUntilTrue(() => {
+ client.listGroups().all().get().stream()
+ .anyMatch(g => g.groupId() == streamsGroupId)
+ }, "Streams group not ready to describe yet")
+
+ Thread.sleep(10000)
+
+ TestUtils.waitUntilTrue(() => {
+ try {
+ val describedGroups =
client.describeStreamsGroups(util.List.of(streamsGroupId)).all().get()
+ val group = describedGroups.get(streamsGroupId)
+ if (group != null) {
+ group.groupState() == GroupState.STABLE &&
!group.subtopologies().isEmpty
+ } else {
+ false
+ }
+ } catch {
+ case _: Exception => false
+ }
+ }, "Stream group not fully initialized with topology")
+
+ // Verify the describe call works correctly
+ val describedGroups =
client.describeStreamsGroups(util.List.of(streamsGroupId)).all().get()
+ val group = describedGroups.get(streamsGroupId)
+
+ assertNotNull(group)
+ assertEquals(streamsGroupId, group.groupId())
+ assertFalse(group.members().isEmpty)
+ assertNotNull(group.subtopologies())
+ assertFalse(group.subtopologies().isEmpty)
+
+ // Verify the topology contains the expected source and sink topics
+ val subtopologies = group.subtopologies().asScala
+ assertTrue(subtopologies.exists(subtopology =>
+ subtopology.sourceTopics().contains(testTopicName)))
+
+ } finally {
+ Utils.closeQuietly(streams, "streams")
+ Utils.closeQuietly(client, "adminClient")
+ }
+ }
+
+ @ParameterizedTest(name =
TestInfoUtils.TestWithParameterizedGroupProtocolNames)
+
@MethodSource(Array("getTestGroupProtocolParametersStreamsGroupProtocolOnly"))
+ def testDeleteStreamsGroups(groupProtocol: String): Unit = {
+ val testTopicName = "test_topic"
+ val testOutputTopicName = "test_output_topic"
+ val testNumPartitions = 3
+ val testNumStreamsGroup = 3
+
+ val targetDeletedGroups = util.List.of("stream_group_id_2",
"stream_group_id_3")
+ val targetRemainingGroups = util.List.of("stream_group_id_1")
+
+ val config = createConfig
+ client = Admin.create(config)
+
+ prepareTopics(List(testTopicName, testOutputTopicName), testNumPartitions)
+ prepareRecords(testTopicName)
+
+ val streamsList = scala.collection.mutable.ListBuffer[(String,
KafkaStreams)]()
+
+ try {
+ for (i <- 1 to testNumStreamsGroup) {
+ val streamsGroupId = s"stream_group_id_$i"
+ val streamsConfig = new Properties()
+ streamsConfig.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest")
+ streamsConfig.put(StreamsConfig.COMMIT_INTERVAL_MS_CONFIG, 10 * 1000)
+ val streams = createStreamsGroup(
+ configOverrides = streamsConfig,
+ inputTopic = testTopicName,
+ outputTopic = testOutputTopicName,
+ streamsGroupId = streamsGroupId,
+ groupProtocol = groupProtocol
+ )
+ streams.cleanUp()
+ streams.start()
+ streamsList += ((streamsGroupId, streams))
+ }
+
+ TestUtils.waitUntilTrue(() => {
+ client.listGroups().all().get().stream()
+ .anyMatch(g => g.groupId().startsWith("stream_group_id_"))
+ }, "Streams groups not ready to delete yet")
+
+ // Verify that there are 3 groups created
+ val groups = client.listGroups().all().get()
Review Comment:
Could we check it up there so that `client.listGroups()` is called one time?
##########
core/src/test/scala/integration/kafka/api/PlaintextAdminIntegrationTest.scala:
##########
@@ -4363,6 +4364,170 @@ class PlaintextAdminIntegrationTest extends
BaseAdminIntegrationTest {
}
}
}
+
+ @ParameterizedTest(name =
TestInfoUtils.TestWithParameterizedGroupProtocolNames)
+
@MethodSource(Array("getTestGroupProtocolParametersStreamsGroupProtocolOnly"))
+ def testDescribeStreamsGroups(groupProtocol: String): Unit = {
+ val streamsGroupId = "stream_group_id"
+ val testTopicName = "test_topic"
+ val testOutputTopicName = "test_output_topic"
+ val testNumPartitions = 1
+
+ val config = createConfig
+ client = Admin.create(config)
+
+ prepareTopics(List(testTopicName, testOutputTopicName), testNumPartitions)
+ prepareRecords(testTopicName)
+
+ val streamsConfig = new Properties()
+ streamsConfig.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest")
+ streamsConfig.put(StreamsConfig.COMMIT_INTERVAL_MS_CONFIG, 10 * 1000)
+ val streams = createStreamsGroup(
+ configOverrides = streamsConfig,
+ inputTopic = testTopicName,
+ outputTopic = testOutputTopicName,
+ streamsGroupId = streamsGroupId,
+ groupProtocol = groupProtocol
+ )
+
+ try {
+ streams.cleanUp()
+ streams.start()
+
+ TestUtils.waitUntilTrue(() => streams.state() ==
KafkaStreams.State.RUNNING, "Streams not in RUNNING state")
+
+ TestUtils.waitUntilTrue(() => {
+ client.listGroups().all().get().stream()
+ .anyMatch(g => g.groupId() == streamsGroupId)
+ }, "Streams group not ready to describe yet")
+
+ Thread.sleep(10000)
+
+ TestUtils.waitUntilTrue(() => {
+ try {
+ val describedGroups =
client.describeStreamsGroups(util.List.of(streamsGroupId)).all().get()
+ val group = describedGroups.get(streamsGroupId)
+ if (group != null) {
+ group.groupState() == GroupState.STABLE &&
!group.subtopologies().isEmpty
+ } else {
+ false
+ }
+ } catch {
+ case _: Exception => false
+ }
+ }, "Stream group not fully initialized with topology")
+
+ // Verify the describe call works correctly
+ val describedGroups =
client.describeStreamsGroups(util.List.of(streamsGroupId)).all().get()
+ val group = describedGroups.get(streamsGroupId)
+
+ assertNotNull(group)
+ assertEquals(streamsGroupId, group.groupId())
+ assertFalse(group.members().isEmpty)
+ assertNotNull(group.subtopologies())
+ assertFalse(group.subtopologies().isEmpty)
+
+ // Verify the topology contains the expected source and sink topics
+ val subtopologies = group.subtopologies().asScala
+ assertTrue(subtopologies.exists(subtopology =>
+ subtopology.sourceTopics().contains(testTopicName)))
+
+ } finally {
+ Utils.closeQuietly(streams, "streams")
+ Utils.closeQuietly(client, "adminClient")
+ }
+ }
+
+ @ParameterizedTest(name =
TestInfoUtils.TestWithParameterizedGroupProtocolNames)
+
@MethodSource(Array("getTestGroupProtocolParametersStreamsGroupProtocolOnly"))
+ def testDeleteStreamsGroups(groupProtocol: String): Unit = {
+ val testTopicName = "test_topic"
+ val testOutputTopicName = "test_output_topic"
+ val testNumPartitions = 3
+ val testNumStreamsGroup = 3
+
+ val targetDeletedGroups = util.List.of("stream_group_id_2",
"stream_group_id_3")
+ val targetRemainingGroups = util.List.of("stream_group_id_1")
+
+ val config = createConfig
+ client = Admin.create(config)
+
+ prepareTopics(List(testTopicName, testOutputTopicName), testNumPartitions)
+ prepareRecords(testTopicName)
+
+ val streamsList = scala.collection.mutable.ListBuffer[(String,
KafkaStreams)]()
+
+ try {
+ for (i <- 1 to testNumStreamsGroup) {
+ val streamsGroupId = s"stream_group_id_$i"
+ val streamsConfig = new Properties()
+ streamsConfig.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest")
+ streamsConfig.put(StreamsConfig.COMMIT_INTERVAL_MS_CONFIG, 10 * 1000)
+ val streams = createStreamsGroup(
+ configOverrides = streamsConfig,
+ inputTopic = testTopicName,
+ outputTopic = testOutputTopicName,
+ streamsGroupId = streamsGroupId,
+ groupProtocol = groupProtocol
+ )
+ streams.cleanUp()
+ streams.start()
+ streamsList += ((streamsGroupId, streams))
+ }
+
+ TestUtils.waitUntilTrue(() => {
+ client.listGroups().all().get().stream()
+ .anyMatch(g => g.groupId().startsWith("stream_group_id_"))
+ }, "Streams groups not ready to delete yet")
+
+ // Verify that there are 3 groups created
+ val groups = client.listGroups().all().get()
Review Comment:
@lucliu1108 did you address this comment?
##########
core/src/test/scala/integration/kafka/api/PlaintextAdminIntegrationTest.scala:
##########
@@ -4363,6 +4364,170 @@ class PlaintextAdminIntegrationTest extends
BaseAdminIntegrationTest {
}
}
}
+
+ @ParameterizedTest(name =
TestInfoUtils.TestWithParameterizedGroupProtocolNames)
+
@MethodSource(Array("getTestGroupProtocolParametersStreamsGroupProtocolOnly"))
+ def testDescribeStreamsGroups(groupProtocol: String): Unit = {
+ val streamsGroupId = "stream_group_id"
+ val testTopicName = "test_topic"
+ val testOutputTopicName = "test_output_topic"
+ val testNumPartitions = 1
+
+ val config = createConfig
+ client = Admin.create(config)
+
+ prepareTopics(List(testTopicName, testOutputTopicName), testNumPartitions)
+ prepareRecords(testTopicName)
+
+ val streamsConfig = new Properties()
+ streamsConfig.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest")
+ streamsConfig.put(StreamsConfig.COMMIT_INTERVAL_MS_CONFIG, 10 * 1000)
+ val streams = createStreamsGroup(
+ configOverrides = streamsConfig,
+ inputTopic = testTopicName,
+ outputTopic = testOutputTopicName,
+ streamsGroupId = streamsGroupId,
+ groupProtocol = groupProtocol
+ )
+
+ try {
+ streams.cleanUp()
+ streams.start()
+
+ TestUtils.waitUntilTrue(() => streams.state() ==
KafkaStreams.State.RUNNING, "Streams not in RUNNING state")
+
+ TestUtils.waitUntilTrue(() => {
+ client.listGroups().all().get().stream()
+ .anyMatch(g => g.groupId() == streamsGroupId)
+ }, "Streams group not ready to describe yet")
+
+ Thread.sleep(10000)
+
+ TestUtils.waitUntilTrue(() => {
+ try {
+ val describedGroups =
client.describeStreamsGroups(util.List.of(streamsGroupId)).all().get()
Review Comment:
Could we check the group state and subtopologies up there when streams group
is returned by the `client.listGroups()`?
##########
core/src/test/scala/integration/kafka/api/PlaintextAdminIntegrationTest.scala:
##########
@@ -4365,6 +4365,71 @@ class PlaintextAdminIntegrationTest extends
BaseAdminIntegrationTest {
}
}
+ @ParameterizedTest(name =
TestInfoUtils.TestWithParameterizedGroupProtocolNames)
+
@MethodSource(Array("getTestGroupProtocolParametersStreamsGroupProtocolOnly"))
+ def testListStreamsGroups(groupProtocol: String): Unit = {
Review Comment:
I thought `list` is going to be tested in the next pr!
##########
core/src/test/scala/integration/kafka/api/PlaintextAdminIntegrationTest.scala:
##########
@@ -4363,6 +4364,170 @@ class PlaintextAdminIntegrationTest extends
BaseAdminIntegrationTest {
}
}
}
+
+ @ParameterizedTest(name =
TestInfoUtils.TestWithParameterizedGroupProtocolNames)
+
@MethodSource(Array("getTestGroupProtocolParametersStreamsGroupProtocolOnly"))
+ def testDescribeStreamsGroups(groupProtocol: String): Unit = {
+ val streamsGroupId = "stream_group_id"
+ val testTopicName = "test_topic"
+ val testOutputTopicName = "test_output_topic"
+ val testNumPartitions = 1
+
+ val config = createConfig
+ client = Admin.create(config)
+
+ prepareTopics(List(testTopicName, testOutputTopicName), testNumPartitions)
+ prepareRecords(testTopicName)
+
+ val streamsConfig = new Properties()
+ streamsConfig.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest")
+ streamsConfig.put(StreamsConfig.COMMIT_INTERVAL_MS_CONFIG, 10 * 1000)
+ val streams = createStreamsGroup(
+ configOverrides = streamsConfig,
+ inputTopic = testTopicName,
+ outputTopic = testOutputTopicName,
+ streamsGroupId = streamsGroupId,
+ groupProtocol = groupProtocol
+ )
+
+ try {
+ streams.cleanUp()
+ streams.start()
+
+ TestUtils.waitUntilTrue(() => streams.state() ==
KafkaStreams.State.RUNNING, "Streams not in RUNNING state")
+
+ TestUtils.waitUntilTrue(() => {
+ client.listGroups().all().get().stream()
+ .anyMatch(g => g.groupId() == streamsGroupId)
+ }, "Streams group not ready to describe yet")
+
+ Thread.sleep(10000)
+
+ TestUtils.waitUntilTrue(() => {
+ try {
+ val describedGroups =
client.describeStreamsGroups(util.List.of(streamsGroupId)).all().get()
+ val group = describedGroups.get(streamsGroupId)
+ if (group != null) {
+ group.groupState() == GroupState.STABLE &&
!group.subtopologies().isEmpty
+ } else {
+ false
+ }
+ } catch {
+ case _: Exception => false
+ }
+ }, "Stream group not fully initialized with topology")
+
+ // Verify the describe call works correctly
+ val describedGroups =
client.describeStreamsGroups(util.List.of(streamsGroupId)).all().get()
+ val group = describedGroups.get(streamsGroupId)
+
+ assertNotNull(group)
+ assertEquals(streamsGroupId, group.groupId())
+ assertFalse(group.members().isEmpty)
+ assertNotNull(group.subtopologies())
+ assertFalse(group.subtopologies().isEmpty)
+
+ // Verify the topology contains the expected source and sink topics
+ val subtopologies = group.subtopologies().asScala
+ assertTrue(subtopologies.exists(subtopology =>
+ subtopology.sourceTopics().contains(testTopicName)))
+
+ } finally {
+ Utils.closeQuietly(streams, "streams")
+ Utils.closeQuietly(client, "adminClient")
+ }
+ }
+
+ @ParameterizedTest(name =
TestInfoUtils.TestWithParameterizedGroupProtocolNames)
+
@MethodSource(Array("getTestGroupProtocolParametersStreamsGroupProtocolOnly"))
+ def testDeleteStreamsGroups(groupProtocol: String): Unit = {
+ val testTopicName = "test_topic"
+ val testOutputTopicName = "test_output_topic"
+ val testNumPartitions = 3
+ val testNumStreamsGroup = 3
+
+ val targetDeletedGroups = util.List.of("stream_group_id_2",
"stream_group_id_3")
+ val targetRemainingGroups = util.List.of("stream_group_id_1")
+
+ val config = createConfig
+ client = Admin.create(config)
+
+ prepareTopics(List(testTopicName, testOutputTopicName), testNumPartitions)
+ prepareRecords(testTopicName)
+
+ val streamsList = scala.collection.mutable.ListBuffer[(String,
KafkaStreams)]()
+
+ try {
+ for (i <- 1 to testNumStreamsGroup) {
+ val streamsGroupId = s"stream_group_id_$i"
+ val streamsConfig = new Properties()
+ streamsConfig.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest")
+ streamsConfig.put(StreamsConfig.COMMIT_INTERVAL_MS_CONFIG, 10 * 1000)
+ val streams = createStreamsGroup(
+ configOverrides = streamsConfig,
+ inputTopic = testTopicName,
+ outputTopic = testOutputTopicName,
+ streamsGroupId = streamsGroupId,
+ groupProtocol = groupProtocol
+ )
+ streams.cleanUp()
+ streams.start()
+ streamsList += ((streamsGroupId, streams))
+ }
+
+ TestUtils.waitUntilTrue(() => {
+ client.listGroups().all().get().stream()
+ .anyMatch(g => g.groupId().startsWith("stream_group_id_"))
+ }, "Streams groups not ready to delete yet")
+
+ // Verify that there are 3 groups created
+ val groups = client.listGroups().all().get()
+ assertEquals(testNumStreamsGroup, groups.size())
+
+ // Test deletion of non-empty existing groups
+ var deleteStreamsGroupResult =
client.deleteStreamsGroups(targetDeletedGroups)
+ assertFutureThrows(classOf[GroupNotEmptyException],
deleteStreamsGroupResult.all())
+ assertEquals(deleteStreamsGroupResult.deletedGroups().size(),2)
+
+ // Stop and clean up the streams for the groups that are going to be
deleted
+ streamsList
+ .filter { case (groupId, _) => targetDeletedGroups.contains(groupId) }
+ .foreach { case (_, streams) =>
+ streams.close(java.time.Duration.ofSeconds(10))
+ streams.cleanUp()
+ }
+
+ // Test deletion of emptied existing streams groups
+ deleteStreamsGroupResult =
client.deleteStreamsGroups(targetDeletedGroups)
+ assertEquals(deleteStreamsGroupResult.deletedGroups().size(),2)
Review Comment:
Could we also list topics and check whether deleting groups impacted
deleting topics as well? Do we have internal topics as well? Would be nice to
check them as well.
##########
core/src/test/scala/integration/kafka/api/PlaintextAdminIntegrationTest.scala:
##########
@@ -4363,6 +4364,170 @@ class PlaintextAdminIntegrationTest extends
BaseAdminIntegrationTest {
}
}
}
+
+ @ParameterizedTest(name =
TestInfoUtils.TestWithParameterizedGroupProtocolNames)
+
@MethodSource(Array("getTestGroupProtocolParametersStreamsGroupProtocolOnly"))
+ def testDescribeStreamsGroups(groupProtocol: String): Unit = {
+ val streamsGroupId = "stream_group_id"
+ val testTopicName = "test_topic"
+ val testOutputTopicName = "test_output_topic"
+ val testNumPartitions = 1
+
+ val config = createConfig
+ client = Admin.create(config)
+
+ prepareTopics(List(testTopicName, testOutputTopicName), testNumPartitions)
+ prepareRecords(testTopicName)
+
+ val streamsConfig = new Properties()
+ streamsConfig.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest")
+ streamsConfig.put(StreamsConfig.COMMIT_INTERVAL_MS_CONFIG, 10 * 1000)
+ val streams = createStreamsGroup(
+ configOverrides = streamsConfig,
+ inputTopic = testTopicName,
+ outputTopic = testOutputTopicName,
+ streamsGroupId = streamsGroupId,
+ groupProtocol = groupProtocol
+ )
+
+ try {
+ streams.cleanUp()
+ streams.start()
+
+ TestUtils.waitUntilTrue(() => streams.state() ==
KafkaStreams.State.RUNNING, "Streams not in RUNNING state")
+
+ TestUtils.waitUntilTrue(() => {
+ client.listGroups().all().get().stream()
+ .anyMatch(g => g.groupId() == streamsGroupId)
+ }, "Streams group not ready to describe yet")
+
+ Thread.sleep(10000)
+
+ TestUtils.waitUntilTrue(() => {
+ try {
+ val describedGroups =
client.describeStreamsGroups(util.List.of(streamsGroupId)).all().get()
+ val group = describedGroups.get(streamsGroupId)
+ if (group != null) {
+ group.groupState() == GroupState.STABLE &&
!group.subtopologies().isEmpty
+ } else {
+ false
+ }
+ } catch {
+ case _: Exception => false
+ }
+ }, "Stream group not fully initialized with topology")
+
Review Comment:
@lucliu1108 ?? this comment is not addressed!
##########
core/src/test/scala/integration/kafka/api/PlaintextAdminIntegrationTest.scala:
##########
@@ -4363,6 +4364,170 @@ class PlaintextAdminIntegrationTest extends
BaseAdminIntegrationTest {
}
}
}
+
+ @ParameterizedTest(name =
TestInfoUtils.TestWithParameterizedGroupProtocolNames)
+
@MethodSource(Array("getTestGroupProtocolParametersStreamsGroupProtocolOnly"))
+ def testDescribeStreamsGroups(groupProtocol: String): Unit = {
+ val streamsGroupId = "stream_group_id"
+ val testTopicName = "test_topic"
+ val testOutputTopicName = "test_output_topic"
+ val testNumPartitions = 1
+
+ val config = createConfig
+ client = Admin.create(config)
+
+ prepareTopics(List(testTopicName, testOutputTopicName), testNumPartitions)
+ prepareRecords(testTopicName)
+
+ val streamsConfig = new Properties()
+ streamsConfig.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest")
+ streamsConfig.put(StreamsConfig.COMMIT_INTERVAL_MS_CONFIG, 10 * 1000)
+ val streams = createStreamsGroup(
+ configOverrides = streamsConfig,
+ inputTopic = testTopicName,
+ outputTopic = testOutputTopicName,
+ streamsGroupId = streamsGroupId,
+ groupProtocol = groupProtocol
+ )
+
+ try {
+ streams.cleanUp()
+ streams.start()
+
+ TestUtils.waitUntilTrue(() => streams.state() ==
KafkaStreams.State.RUNNING, "Streams not in RUNNING state")
+
+ TestUtils.waitUntilTrue(() => {
+ client.listGroups().all().get().stream()
+ .anyMatch(g => g.groupId() == streamsGroupId)
+ }, "Streams group not ready to describe yet")
+
+ Thread.sleep(10000)
+
+ TestUtils.waitUntilTrue(() => {
+ try {
+ val describedGroups =
client.describeStreamsGroups(util.List.of(streamsGroupId)).all().get()
+ val group = describedGroups.get(streamsGroupId)
+ if (group != null) {
+ group.groupState() == GroupState.STABLE &&
!group.subtopologies().isEmpty
+ } else {
+ false
+ }
+ } catch {
+ case _: Exception => false
+ }
+ }, "Stream group not fully initialized with topology")
+
+ // Verify the describe call works correctly
+ val describedGroups =
client.describeStreamsGroups(util.List.of(streamsGroupId)).all().get()
+ val group = describedGroups.get(streamsGroupId)
+
+ assertNotNull(group)
+ assertEquals(streamsGroupId, group.groupId())
+ assertFalse(group.members().isEmpty)
+ assertNotNull(group.subtopologies())
+ assertFalse(group.subtopologies().isEmpty)
+
+ // Verify the topology contains the expected source and sink topics
+ val subtopologies = group.subtopologies().asScala
+ assertTrue(subtopologies.exists(subtopology =>
+ subtopology.sourceTopics().contains(testTopicName)))
+
+ } finally {
+ Utils.closeQuietly(streams, "streams")
+ Utils.closeQuietly(client, "adminClient")
+ }
+ }
+
+ @ParameterizedTest(name =
TestInfoUtils.TestWithParameterizedGroupProtocolNames)
+
@MethodSource(Array("getTestGroupProtocolParametersStreamsGroupProtocolOnly"))
+ def testDeleteStreamsGroups(groupProtocol: String): Unit = {
+ val testTopicName = "test_topic"
+ val testOutputTopicName = "test_output_topic"
+ val testNumPartitions = 3
+ val testNumStreamsGroup = 3
+
+ val targetDeletedGroups = util.List.of("stream_group_id_2",
"stream_group_id_3")
+ val targetRemainingGroups = util.List.of("stream_group_id_1")
+
+ val config = createConfig
+ client = Admin.create(config)
+
+ prepareTopics(List(testTopicName, testOutputTopicName), testNumPartitions)
+ prepareRecords(testTopicName)
+
+ val streamsList = scala.collection.mutable.ListBuffer[(String,
KafkaStreams)]()
+
+ try {
+ for (i <- 1 to testNumStreamsGroup) {
+ val streamsGroupId = s"stream_group_id_$i"
+ val streamsConfig = new Properties()
+ streamsConfig.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest")
+ streamsConfig.put(StreamsConfig.COMMIT_INTERVAL_MS_CONFIG, 10 * 1000)
+ val streams = createStreamsGroup(
+ configOverrides = streamsConfig,
+ inputTopic = testTopicName,
+ outputTopic = testOutputTopicName,
+ streamsGroupId = streamsGroupId,
+ groupProtocol = groupProtocol
+ )
+ streams.cleanUp()
+ streams.start()
+ streamsList += ((streamsGroupId, streams))
+ }
+
+ TestUtils.waitUntilTrue(() => {
+ client.listGroups().all().get().stream()
+ .anyMatch(g => g.groupId().startsWith("stream_group_id_"))
+ }, "Streams groups not ready to delete yet")
+
+ // Verify that there are 3 groups created
+ val groups = client.listGroups().all().get()
+ assertEquals(testNumStreamsGroup, groups.size())
+
+ // Test deletion of non-empty existing groups
+ var deleteStreamsGroupResult =
client.deleteStreamsGroups(targetDeletedGroups)
+ assertFutureThrows(classOf[GroupNotEmptyException],
deleteStreamsGroupResult.all())
+ assertEquals(deleteStreamsGroupResult.deletedGroups().size(),2)
+
+ // Stop and clean up the streams for the groups that are going to be
deleted
+ streamsList
+ .filter { case (groupId, _) => targetDeletedGroups.contains(groupId) }
+ .foreach { case (_, streams) =>
+ streams.close(java.time.Duration.ofSeconds(10))
+ streams.cleanUp()
+ }
+
+ // Test deletion of emptied existing streams groups
+ deleteStreamsGroupResult =
client.deleteStreamsGroups(targetDeletedGroups)
+ assertEquals(deleteStreamsGroupResult.deletedGroups().size(),2)
Review Comment:
@lucliu1108 this comment?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscr...@kafka.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
