You are viewing a plain text version of this content. The canonical link for it is here.
Posted to jira@kafka.apache.org by "FrankYang0529 (via GitHub)" <gi...@apache.org> on 2024/12/04 12:16:34 UTC
[PR] KAFKA-17750: Extend kafka-consumer-groups command line tool to support new consumer group (part 2) [kafka]
FrankYang0529 opened a new pull request, #18034:
URL: https://github.com/apache/kafka/pull/18034
* Add fields `groupEpoch` and `targetAssignmentEpoch` to `ConsumerGroupDescription.java`.
* Add fields `memberEpoch` and `isClassic` to `MemberDescription.java`.
* Add assertion to `PlaintextAdminIntegrationTest#testDescribeClassicGroups` to make sure member in classic group returns `isClassic` as `Optional.of(true)`.
* Add new case `testConsumerGroupWithMemberMigration` to `PlaintextAdminIntegrationTest` to make sure migration member has correct `isClassic` value. Add assertion for `groupEpoch`, `targetAssignmentEpoch`, `memberEpoch` as well.
### Committer Checklist (excluded from commit message)
- [ ] Verify design and implementation
- [ ] Verify test coverage and CI build status
- [ ] Verify documentation (including upgrade notes)
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-17750: Extend kafka-consumer-groups command line tool to support new consumer group (part 2) [kafka]
Posted by "dajac (via GitHub)" <gi...@apache.org>.
dajac commented on code in PR #18034:
URL: https://github.com/apache/kafka/pull/18034#discussion_r1870988775
##########
clients/src/main/java/org/apache/kafka/clients/admin/ConsumerGroupDescription.java:
##########
@@ -108,7 +113,8 @@ public ConsumerGroupDescription(String groupId,
GroupState groupState,
Node coordinator,
Set<AclOperation> authorizedOperations) {
- this(groupId, isSimpleConsumerGroup, members, partitionAssignor, GroupType.CLASSIC, groupState, coordinator, authorizedOperations);
+ this(groupId, isSimpleConsumerGroup, members, partitionAssignor, GroupType.CLASSIC, groupState, coordinator, authorizedOperations,
+ Optional.empty(), Optional.empty());
}
public ConsumerGroupDescription(String groupId,
Review Comment:
Was this constructor introduced in 4.0 only? It seems to be the case but we need to double check. If it is not, we would need to introduce a new one in order to not break existing usages.
##########
clients/src/main/java/org/apache/kafka/clients/admin/MemberDescription.java:
##########
@@ -30,21 +30,26 @@ public class MemberDescription {
private final String host;
private final MemberAssignment assignment;
private final Optional<MemberAssignment> targetAssignment;
+ private final Optional<Integer> memberEpoch;
+ private final Optional<Boolean> isClassic;
public MemberDescription(String memberId,
Optional<String> groupInstanceId,
String clientId,
String host,
MemberAssignment assignment,
- Optional<MemberAssignment> targetAssignment
- ) {
+ Optional<MemberAssignment> targetAssignment,
+ Optional<Integer> memberEpoch,
+ Optional<Boolean> isClassic) {
Review Comment:
nit: Let's keep the closing `)` on a new line as it was before.
##########
clients/src/main/java/org/apache/kafka/clients/admin/MemberDescription.java:
##########
@@ -30,21 +30,26 @@ public class MemberDescription {
private final String host;
private final MemberAssignment assignment;
private final Optional<MemberAssignment> targetAssignment;
+ private final Optional<Integer> memberEpoch;
+ private final Optional<Boolean> isClassic;
public MemberDescription(String memberId,
Review Comment:
Here, I suppose that we must introduce a new overload of the constructor with the new arguments in order to not break compatibility of existing usages.
##########
clients/src/main/java/org/apache/kafka/clients/admin/MemberDescription.java:
##########
@@ -131,13 +140,30 @@ public Optional<MemberAssignment> targetAssignment() {
return targetAssignment;
}
+ /**
+ * The epoch of the group member.
+ */
+ public Optional<Integer> memberEpoch() {
+ return memberEpoch;
+ }
+
+ /**
+ * The flag indicating whether a member is classic.
+ */
+ public Optional<Boolean> isClassic() {
Review Comment:
Sorry for coming back on this one but I just thought about an alternative. How about calling it `nonUpgraded`? We could set it to true when the group is a consumer group and the member uses classic protocol. We could also only set it when the group is a consumer group vs a classic group. This may be even more explicit for users. What do you think? @chia7712 Do you have any thoughts too?
##########
clients/src/main/java/org/apache/kafka/clients/admin/MemberDescription.java:
##########
@@ -131,13 +140,30 @@ public Optional<MemberAssignment> targetAssignment() {
return targetAssignment;
}
+ /**
+ * The epoch of the group member.
+ */
+ public Optional<Integer> memberEpoch() {
+ return memberEpoch;
+ }
+
+ /**
+ * The flag indicating whether a member is classic.
Review Comment:
nit: How about `The flag indicating whether a member with a consumer group uses the classic protocol. It indicates that the member was perhaps not upgraded`?
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-17750: Extend kafka-consumer-groups command line tool to support new consumer group (part 2) [kafka]
Posted by "FrankYang0529 (via GitHub)" <gi...@apache.org>.
FrankYang0529 commented on code in PR #18034:
URL: https://github.com/apache/kafka/pull/18034#discussion_r1873377917
##########
clients/src/main/java/org/apache/kafka/clients/admin/MemberDescription.java:
##########
@@ -30,21 +30,26 @@ public class MemberDescription {
private final String host;
private final MemberAssignment assignment;
private final Optional<MemberAssignment> targetAssignment;
+ private final Optional<Integer> memberEpoch;
+ private final Optional<Boolean> isClassic;
public MemberDescription(String memberId,
Review Comment:
Updated both PR and KIP. Thanks for the suggestion.
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-17750: Extend kafka-consumer-groups command line tool to support new consumer group (part 2) [kafka]
Posted by "FrankYang0529 (via GitHub)" <gi...@apache.org>.
FrankYang0529 commented on code in PR #18034:
URL: https://github.com/apache/kafka/pull/18034#discussion_r1871641626
##########
clients/src/main/java/org/apache/kafka/clients/admin/ConsumerGroupDescription.java:
##########
@@ -108,7 +113,8 @@ public ConsumerGroupDescription(String groupId,
GroupState groupState,
Node coordinator,
Set<AclOperation> authorizedOperations) {
- this(groupId, isSimpleConsumerGroup, members, partitionAssignor, GroupType.CLASSIC, groupState, coordinator, authorizedOperations);
+ this(groupId, isSimpleConsumerGroup, members, partitionAssignor, GroupType.CLASSIC, groupState, coordinator, authorizedOperations,
+ Optional.empty(), Optional.empty());
}
public ConsumerGroupDescription(String groupId,
Review Comment:
This constructor is introduced in 4.0. It is not existent in 3.9.
https://github.com/apache/kafka/blob/84caaa6e9da06435411510a81fa321d4f99c351f/clients/src/main/java/org/apache/kafka/clients/admin/ConsumerGroupDescription.java#L45-L81
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-17750: Extend kafka-consumer-groups command line tool to support new consumer group (part 2) [kafka]
Posted by "dajac (via GitHub)" <gi...@apache.org>.
dajac commented on code in PR #18034:
URL: https://github.com/apache/kafka/pull/18034#discussion_r1871335509
##########
clients/src/main/java/org/apache/kafka/clients/admin/MemberDescription.java:
##########
@@ -131,13 +140,30 @@ public Optional<MemberAssignment> targetAssignment() {
return targetAssignment;
}
+ /**
+ * The epoch of the group member.
+ */
+ public Optional<Integer> memberEpoch() {
+ return memberEpoch;
+ }
+
+ /**
+ * The flag indicating whether a member is classic.
+ */
+ public Optional<Boolean> isClassic() {
Review Comment:
That works too. Something as follow?
* Classic Group -> Optional.empty
* Consumer Group -> Optional.empty if unknown
* Consumer Group -> Optional.of(false) if classic
* Consumer Group -> Optional.of(true) if consumer
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-17750: Extend kafka-consumer-groups command line tool to support new consumer group (part 2) [kafka]
Posted by "FrankYang0529 (via GitHub)" <gi...@apache.org>.
FrankYang0529 commented on code in PR #18034:
URL: https://github.com/apache/kafka/pull/18034#discussion_r1876003093
##########
core/src/test/scala/integration/kafka/api/PlaintextAdminIntegrationTest.scala:
##########
@@ -2062,6 +2067,86 @@ class PlaintextAdminIntegrationTest extends BaseAdminIntegrationTest {
}
}
+ /**
+ * Test the consumer group APIs.
+ */
+ @ParameterizedTest
+ @ValueSource(strings = Array("kraft"))
+ def testConsumerGroupWithMemberMigration(quorum: String): Unit = {
+ val config = createConfig
+ client = Admin.create(config)
+ var classicConsumer: Consumer[Array[Byte], Array[Byte]] = null
+ var consumerConsumer: Consumer[Array[Byte], Array[Byte]] = null
+ 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 testNumPartitions = 2
+
+ client.createTopics(util.Arrays.asList(
+ new NewTopic(testTopicName, testNumPartitions, 1.toShort),
+ )).all.get
+ waitForTopics(client, List(testTopicName), List())
+
+ val producer = createProducer()
+ try {
+ producer.send(new ProducerRecord(testTopicName, 0, null, null))
+ producer.send(new ProducerRecord(testTopicName, 1, null, null))
+ producer.flush()
+ } finally {
+ Utils.closeQuietly(producer, "producer")
+ }
+
+ val testGroupId = "test_group_id"
+ val testClassicClientId = "test_classic_client_id"
+ val testConsumerClientId = "test_consumer_client_id"
+
+ val newConsumerConfig = new Properties(consumerConfig)
+ newConsumerConfig.put(ConsumerConfig.GROUP_ID_CONFIG, testGroupId)
+ newConsumerConfig.put(ConsumerConfig.CLIENT_ID_CONFIG, testClassicClientId)
+ consumerConfig.put(ConsumerConfig.GROUP_PROTOCOL_CONFIG, GroupProtocol.CLASSIC.name)
+
+ classicConsumer = createConsumer(configOverrides = newConsumerConfig)
+ classicConsumer.subscribe(List(testTopicName).asJava)
+ classicConsumer.poll(JDuration.ofMillis(1000))
+
+ newConsumerConfig.put(ConsumerConfig.CLIENT_ID_CONFIG, testConsumerClientId)
+ consumerConfig.put(ConsumerConfig.GROUP_PROTOCOL_CONFIG, GroupProtocol.CONSUMER.name)
+ consumerConsumer = createConsumer(configOverrides = newConsumerConfig)
+ consumerConsumer.subscribe(List(testTopicName).asJava)
+ consumerConsumer.poll(JDuration.ofMillis(1000))
+
+ TestUtils.waitUntilTrue(() => {
+ val describeConsumerGroupResult = client.describeConsumerGroups(Seq(testGroupId).asJava).all.get
+ describeConsumerGroupResult.containsKey(testGroupId) &&
+ describeConsumerGroupResult.get(testGroupId).members.size == 2
Review Comment:
Sorry, my mistake. I add `classicConsumer#poll` and `consumerConsumer#poll` to `TestUtils#waitUntilTure` function. Finally, the group can be STABLE. Thanks.
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-17750: Extend kafka-consumer-groups command line tool to support new consumer group (part 2) [kafka]
Posted by "dajac (via GitHub)" <gi...@apache.org>.
dajac commented on PR #18034:
URL: https://github.com/apache/kafka/pull/18034#issuecomment-2531578528
@FrankYang0529 Done :)
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-17750: Extend kafka-consumer-groups command line tool to support new consumer group (part 2) [kafka]
Posted by "chia7712 (via GitHub)" <gi...@apache.org>.
chia7712 commented on code in PR #18034:
URL: https://github.com/apache/kafka/pull/18034#discussion_r1871218090
##########
clients/src/main/java/org/apache/kafka/clients/admin/MemberDescription.java:
##########
@@ -131,13 +140,30 @@ public Optional<MemberAssignment> targetAssignment() {
return targetAssignment;
}
+ /**
+ * The epoch of the group member.
+ */
+ public Optional<Integer> memberEpoch() {
+ return memberEpoch;
+ }
+
+ /**
+ * The flag indicating whether a member is classic.
+ */
+ public Optional<Boolean> isClassic() {
Review Comment:
> How about calling it nonUpgraded? We could set it to true when the group is a consumer group and the member uses classic protocol.
How about using "upgraded"? it is true only if the group is a consumer group and the member uses the consumer protocol.
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-17750: Extend kafka-consumer-groups command line tool to support new consumer group (part 2) [kafka]
Posted by "FrankYang0529 (via GitHub)" <gi...@apache.org>.
FrankYang0529 commented on code in PR #18034:
URL: https://github.com/apache/kafka/pull/18034#discussion_r1871644232
##########
clients/src/main/java/org/apache/kafka/clients/admin/MemberDescription.java:
##########
@@ -30,21 +30,26 @@ public class MemberDescription {
private final String host;
private final MemberAssignment assignment;
private final Optional<MemberAssignment> targetAssignment;
+ private final Optional<Integer> memberEpoch;
+ private final Optional<Boolean> isClassic;
public MemberDescription(String memberId,
Review Comment:
Thanks for reminder. Updated it.
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-17750: Extend kafka-consumer-groups command line tool to support new consumer group (part 2) [kafka]
Posted by "dajac (via GitHub)" <gi...@apache.org>.
dajac commented on code in PR #18034:
URL: https://github.com/apache/kafka/pull/18034#discussion_r1872827341
##########
clients/src/main/java/org/apache/kafka/clients/admin/MemberDescription.java:
##########
@@ -45,8 +49,37 @@ public MemberDescription(String memberId,
this.assignment = assignment == null ?
new MemberAssignment(Collections.emptySet()) : assignment;
this.targetAssignment = targetAssignment;
+ this.memberEpoch = memberEpoch;
+ this.upgraded = upgraded;
}
+ /**
+ * @deprecated Since 4.0. Use {@link #MemberDescription(String, Optional, String, String, MemberAssignment, Optional, Optional, Optional)}.
+ */
+ @Deprecated
+ public MemberDescription(String memberId,
+ Optional<String> groupInstanceId,
+ String clientId,
+ String host,
+ MemberAssignment assignment,
+ Optional<MemberAssignment> targetAssignment
+ ) {
Review Comment:
nit: I think that we are mixing two code styles here. We would usually use:
```
public MemberDescription(
String memberId,
Optional<String> groupInstanceId,
String clientId,
String host,
MemberAssignment assignment,
Optional<MemberAssignment> targetAssignment
) {
```
or
```
public MemberDescription(String memberId,
Optional<String> groupInstanceId,
String clientId,
String host,
MemberAssignment assignment,
Optional<MemberAssignment> targetAssignment) {
```
I personally prefer the first one but it is up to you.
##########
core/src/test/scala/integration/kafka/api/PlaintextAdminIntegrationTest.scala:
##########
@@ -1921,12 +1921,17 @@ class PlaintextAdminIntegrationTest extends BaseAdminIntegrationTest {
// Test that we can get information about the test consumer group.
assertTrue(describeWithFakeGroupResult.describedGroups().containsKey(testGroupId))
var testGroupDescription = describeWithFakeGroupResult.describedGroups().get(testGroupId).get()
+ assertEquals(groupType == GroupType.CLASSIC, testGroupDescription.groupEpoch.isEmpty)
+ assertEquals(groupType == GroupType.CLASSIC, testGroupDescription.targetAssignmentEpoch.isEmpty)
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()))
+ members.asScala.foreach(member => {
+ assertEquals(testClientId, member.clientId)
+ assertEquals(if (groupType == GroupType.CLASSIC) Optional.empty else Optional.of(true), member.upgraded)
+ })
Review Comment:
nit: `foreach(member => {` -> `foreach { member => `
##########
clients/src/test/java/org/apache/kafka/clients/admin/internals/DescribeConsumerGroupsHandlerTest.java:
##########
@@ -164,17 +165,21 @@ public void testSuccessfulHandleConsumerGroupResponse() {
Optional.of(new MemberAssignment(Set.of(
new TopicPartition("foo", 1),
new TopicPartition("bar", 2)
- )))
+ ))),
+ Optional.of(10),
+ Optional.of(true)
Review Comment:
Should we cover the unknown and the classic cases in the tests too?
##########
core/src/test/scala/integration/kafka/api/PlaintextAdminIntegrationTest.scala:
##########
@@ -2062,6 +2067,86 @@ class PlaintextAdminIntegrationTest extends BaseAdminIntegrationTest {
}
}
+ /**
+ * Test the consumer group APIs.
+ */
+ @ParameterizedTest
+ @ValueSource(strings = Array("kraft"))
+ def testConsumerGroupWithMemberMigration(quorum: String): Unit = {
+ val config = createConfig
+ client = Admin.create(config)
+ var classicConsumer: Consumer[Array[Byte], Array[Byte]] = null
+ var consumerConsumer: Consumer[Array[Byte], Array[Byte]] = null
+ 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 testNumPartitions = 2
+
+ client.createTopics(util.Arrays.asList(
+ new NewTopic(testTopicName, testNumPartitions, 1.toShort),
+ )).all.get
+ waitForTopics(client, List(testTopicName), List())
+
+ val producer = createProducer()
+ try {
+ producer.send(new ProducerRecord(testTopicName, 0, null, null))
+ producer.send(new ProducerRecord(testTopicName, 1, null, null))
+ producer.flush()
+ } finally {
+ Utils.closeQuietly(producer, "producer")
+ }
+
+ val testGroupId = "test_group_id"
+ val testClassicClientId = "test_classic_client_id"
+ val testConsumerClientId = "test_consumer_client_id"
+
+ val newConsumerConfig = new Properties(consumerConfig)
+ newConsumerConfig.put(ConsumerConfig.GROUP_ID_CONFIG, testGroupId)
+ newConsumerConfig.put(ConsumerConfig.CLIENT_ID_CONFIG, testClassicClientId)
+ consumerConfig.put(ConsumerConfig.GROUP_PROTOCOL_CONFIG, GroupProtocol.CLASSIC.name)
+
+ classicConsumer = createConsumer(configOverrides = newConsumerConfig)
+ classicConsumer.subscribe(List(testTopicName).asJava)
+ classicConsumer.poll(JDuration.ofMillis(1000))
+
+ newConsumerConfig.put(ConsumerConfig.CLIENT_ID_CONFIG, testConsumerClientId)
+ consumerConfig.put(ConsumerConfig.GROUP_PROTOCOL_CONFIG, GroupProtocol.CONSUMER.name)
+ consumerConsumer = createConsumer(configOverrides = newConsumerConfig)
+ consumerConsumer.subscribe(List(testTopicName).asJava)
+ consumerConsumer.poll(JDuration.ofMillis(1000))
+
+ TestUtils.waitUntilTrue(() => {
+ val describeConsumerGroupResult = client.describeConsumerGroups(Seq(testGroupId).asJava).all.get
+ describeConsumerGroupResult.containsKey(testGroupId) &&
+ describeConsumerGroupResult.get(testGroupId).members.size == 2
+ }, s"Expected the offset for partition 0 to eventually become 1.")
+
+ val describeConsumerGroupResult = client.describeConsumerGroups(Seq(testGroupId).asJava).all.get
+ val group = describeConsumerGroupResult.get(testGroupId)
+ assertNotNull(group)
+ assertTrue(group.groupEpoch.isPresent && group.groupEpoch.get > 0)
+ assertTrue(group.targetAssignmentEpoch.isPresent && group.targetAssignmentEpoch.get > 0)
Review Comment:
ditto.
##########
core/src/test/scala/integration/kafka/api/PlaintextAdminIntegrationTest.scala:
##########
@@ -2062,6 +2067,86 @@ class PlaintextAdminIntegrationTest extends BaseAdminIntegrationTest {
}
}
+ /**
+ * Test the consumer group APIs.
+ */
+ @ParameterizedTest
+ @ValueSource(strings = Array("kraft"))
+ def testConsumerGroupWithMemberMigration(quorum: String): Unit = {
+ val config = createConfig
+ client = Admin.create(config)
+ var classicConsumer: Consumer[Array[Byte], Array[Byte]] = null
+ var consumerConsumer: Consumer[Array[Byte], Array[Byte]] = null
+ 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 testNumPartitions = 2
+
+ client.createTopics(util.Arrays.asList(
+ new NewTopic(testTopicName, testNumPartitions, 1.toShort),
+ )).all.get
+ waitForTopics(client, List(testTopicName), List())
+
+ val producer = createProducer()
+ try {
+ producer.send(new ProducerRecord(testTopicName, 0, null, null))
+ producer.send(new ProducerRecord(testTopicName, 1, null, null))
+ producer.flush()
+ } finally {
+ Utils.closeQuietly(producer, "producer")
+ }
+
+ val testGroupId = "test_group_id"
+ val testClassicClientId = "test_classic_client_id"
+ val testConsumerClientId = "test_consumer_client_id"
+
+ val newConsumerConfig = new Properties(consumerConfig)
+ newConsumerConfig.put(ConsumerConfig.GROUP_ID_CONFIG, testGroupId)
+ newConsumerConfig.put(ConsumerConfig.CLIENT_ID_CONFIG, testClassicClientId)
+ consumerConfig.put(ConsumerConfig.GROUP_PROTOCOL_CONFIG, GroupProtocol.CLASSIC.name)
+
+ classicConsumer = createConsumer(configOverrides = newConsumerConfig)
+ classicConsumer.subscribe(List(testTopicName).asJava)
+ classicConsumer.poll(JDuration.ofMillis(1000))
+
+ newConsumerConfig.put(ConsumerConfig.CLIENT_ID_CONFIG, testConsumerClientId)
+ consumerConfig.put(ConsumerConfig.GROUP_PROTOCOL_CONFIG, GroupProtocol.CONSUMER.name)
+ consumerConsumer = createConsumer(configOverrides = newConsumerConfig)
+ consumerConsumer.subscribe(List(testTopicName).asJava)
+ consumerConsumer.poll(JDuration.ofMillis(1000))
+
+ TestUtils.waitUntilTrue(() => {
+ val describeConsumerGroupResult = client.describeConsumerGroups(Seq(testGroupId).asJava).all.get
+ describeConsumerGroupResult.containsKey(testGroupId) &&
+ describeConsumerGroupResult.get(testGroupId).members.size == 2
Review Comment:
Here you could also wait until the group becomes STABLE. It will ease the assertions afterwards.
##########
core/src/test/scala/integration/kafka/api/PlaintextAdminIntegrationTest.scala:
##########
@@ -1921,12 +1921,17 @@ class PlaintextAdminIntegrationTest extends BaseAdminIntegrationTest {
// Test that we can get information about the test consumer group.
assertTrue(describeWithFakeGroupResult.describedGroups().containsKey(testGroupId))
var testGroupDescription = describeWithFakeGroupResult.describedGroups().get(testGroupId).get()
+ assertEquals(groupType == GroupType.CLASSIC, testGroupDescription.groupEpoch.isEmpty)
+ assertEquals(groupType == GroupType.CLASSIC, testGroupDescription.targetAssignmentEpoch.isEmpty)
Review Comment:
Would it be possible to actually verify the values too?
##########
core/src/test/scala/integration/kafka/api/PlaintextAdminIntegrationTest.scala:
##########
@@ -2062,6 +2067,86 @@ class PlaintextAdminIntegrationTest extends BaseAdminIntegrationTest {
}
}
+ /**
+ * Test the consumer group APIs.
+ */
+ @ParameterizedTest
+ @ValueSource(strings = Array("kraft"))
+ def testConsumerGroupWithMemberMigration(quorum: String): Unit = {
+ val config = createConfig
+ client = Admin.create(config)
+ var classicConsumer: Consumer[Array[Byte], Array[Byte]] = null
+ var consumerConsumer: Consumer[Array[Byte], Array[Byte]] = null
+ 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 testNumPartitions = 2
+
+ client.createTopics(util.Arrays.asList(
+ new NewTopic(testTopicName, testNumPartitions, 1.toShort),
+ )).all.get
+ waitForTopics(client, List(testTopicName), List())
+
+ val producer = createProducer()
+ try {
+ producer.send(new ProducerRecord(testTopicName, 0, null, null))
+ producer.send(new ProducerRecord(testTopicName, 1, null, null))
+ producer.flush()
+ } finally {
+ Utils.closeQuietly(producer, "producer")
+ }
+
+ val testGroupId = "test_group_id"
+ val testClassicClientId = "test_classic_client_id"
+ val testConsumerClientId = "test_consumer_client_id"
+
+ val newConsumerConfig = new Properties(consumerConfig)
+ newConsumerConfig.put(ConsumerConfig.GROUP_ID_CONFIG, testGroupId)
+ newConsumerConfig.put(ConsumerConfig.CLIENT_ID_CONFIG, testClassicClientId)
+ consumerConfig.put(ConsumerConfig.GROUP_PROTOCOL_CONFIG, GroupProtocol.CLASSIC.name)
+
+ classicConsumer = createConsumer(configOverrides = newConsumerConfig)
+ classicConsumer.subscribe(List(testTopicName).asJava)
+ classicConsumer.poll(JDuration.ofMillis(1000))
+
+ newConsumerConfig.put(ConsumerConfig.CLIENT_ID_CONFIG, testConsumerClientId)
+ consumerConfig.put(ConsumerConfig.GROUP_PROTOCOL_CONFIG, GroupProtocol.CONSUMER.name)
+ consumerConsumer = createConsumer(configOverrides = newConsumerConfig)
+ consumerConsumer.subscribe(List(testTopicName).asJava)
+ consumerConsumer.poll(JDuration.ofMillis(1000))
+
+ TestUtils.waitUntilTrue(() => {
+ val describeConsumerGroupResult = client.describeConsumerGroups(Seq(testGroupId).asJava).all.get
+ describeConsumerGroupResult.containsKey(testGroupId) &&
+ describeConsumerGroupResult.get(testGroupId).members.size == 2
+ }, s"Expected the offset for partition 0 to eventually become 1.")
Review Comment:
This comment seems to be off.
##########
clients/src/main/java/org/apache/kafka/clients/admin/MemberDescription.java:
##########
@@ -131,13 +170,33 @@ public Optional<MemberAssignment> targetAssignment() {
return targetAssignment;
}
+ /**
+ * The epoch of the group member.
+ */
+ public Optional<Integer> memberEpoch() {
+ return memberEpoch;
+ }
+
+ /**
+ * The flag indicating whether a member with a consumer group uses the consumer protocol.
+ * True for consumer member.
+ * False for classic member
+ * Empty for unknown or members in classic group.
Review Comment:
nit: I suggest to reference the group type in the java doc in order to be crystal clear. e.g.
```
The flag indicating whether a member within a {CONSUMER} group uses the {CONSUMER} protocol.
The optional is set to true if it does, to false if it does not, and to empty if it is unknown or if the group
is a {CLASSIC} group.
```
##########
clients/src/main/java/org/apache/kafka/clients/admin/ConsumerGroupDescription.java:
##########
@@ -215,6 +208,20 @@ public Set<AclOperation> authorizedOperations() {
return authorizedOperations;
}
+ /**
+ * The epoch of the consumer group.
+ */
+ public Optional<Integer> groupEpoch() {
+ return groupEpoch;
+ }
+
+ /**
+ * The epoch of the target assignment.
+ */
+ public Optional<Integer> targetAssignmentEpoch() {
+ return targetAssignmentEpoch;
+ }
Review Comment:
Let's also mention here that they are only provided when the type is consumer.
##########
core/src/test/scala/integration/kafka/api/PlaintextAdminIntegrationTest.scala:
##########
@@ -2062,6 +2067,86 @@ class PlaintextAdminIntegrationTest extends BaseAdminIntegrationTest {
}
}
+ /**
+ * Test the consumer group APIs.
+ */
+ @ParameterizedTest
+ @ValueSource(strings = Array("kraft"))
+ def testConsumerGroupWithMemberMigration(quorum: String): Unit = {
+ val config = createConfig
+ client = Admin.create(config)
+ var classicConsumer: Consumer[Array[Byte], Array[Byte]] = null
+ var consumerConsumer: Consumer[Array[Byte], Array[Byte]] = null
+ 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 testNumPartitions = 2
+
+ client.createTopics(util.Arrays.asList(
+ new NewTopic(testTopicName, testNumPartitions, 1.toShort),
+ )).all.get
+ waitForTopics(client, List(testTopicName), List())
+
+ val producer = createProducer()
+ try {
+ producer.send(new ProducerRecord(testTopicName, 0, null, null))
+ producer.send(new ProducerRecord(testTopicName, 1, null, null))
+ producer.flush()
+ } finally {
+ Utils.closeQuietly(producer, "producer")
+ }
+
+ val testGroupId = "test_group_id"
+ val testClassicClientId = "test_classic_client_id"
+ val testConsumerClientId = "test_consumer_client_id"
+
+ val newConsumerConfig = new Properties(consumerConfig)
+ newConsumerConfig.put(ConsumerConfig.GROUP_ID_CONFIG, testGroupId)
+ newConsumerConfig.put(ConsumerConfig.CLIENT_ID_CONFIG, testClassicClientId)
+ consumerConfig.put(ConsumerConfig.GROUP_PROTOCOL_CONFIG, GroupProtocol.CLASSIC.name)
+
+ classicConsumer = createConsumer(configOverrides = newConsumerConfig)
+ classicConsumer.subscribe(List(testTopicName).asJava)
+ classicConsumer.poll(JDuration.ofMillis(1000))
+
+ newConsumerConfig.put(ConsumerConfig.CLIENT_ID_CONFIG, testConsumerClientId)
+ consumerConfig.put(ConsumerConfig.GROUP_PROTOCOL_CONFIG, GroupProtocol.CONSUMER.name)
+ consumerConsumer = createConsumer(configOverrides = newConsumerConfig)
+ consumerConsumer.subscribe(List(testTopicName).asJava)
+ consumerConsumer.poll(JDuration.ofMillis(1000))
+
+ TestUtils.waitUntilTrue(() => {
+ val describeConsumerGroupResult = client.describeConsumerGroups(Seq(testGroupId).asJava).all.get
+ describeConsumerGroupResult.containsKey(testGroupId) &&
+ describeConsumerGroupResult.get(testGroupId).members.size == 2
+ }, s"Expected the offset for partition 0 to eventually become 1.")
+
+ val describeConsumerGroupResult = client.describeConsumerGroups(Seq(testGroupId).asJava).all.get
+ val group = describeConsumerGroupResult.get(testGroupId)
+ assertNotNull(group)
+ assertTrue(group.groupEpoch.isPresent && group.groupEpoch.get > 0)
+ assertTrue(group.targetAssignmentEpoch.isPresent && group.targetAssignmentEpoch.get > 0)
+
+ val classicMember = group.members.asScala.find(_.clientId == testClassicClientId)
+ assertTrue(classicMember.isDefined)
+ assertTrue(classicMember.get.memberEpoch.isPresent && classicMember.get.memberEpoch.get > 0)
+ assertTrue(classicMember.get.upgraded.isPresent && !classicMember.get.upgraded.get)
Review Comment:
`assertEquals(Optional.of(false), classicMember.get.upgraded)`.
##########
core/src/test/scala/integration/kafka/api/PlaintextAdminIntegrationTest.scala:
##########
@@ -2062,6 +2067,86 @@ class PlaintextAdminIntegrationTest extends BaseAdminIntegrationTest {
}
}
+ /**
+ * Test the consumer group APIs.
+ */
+ @ParameterizedTest
+ @ValueSource(strings = Array("kraft"))
+ def testConsumerGroupWithMemberMigration(quorum: String): Unit = {
+ val config = createConfig
+ client = Admin.create(config)
+ var classicConsumer: Consumer[Array[Byte], Array[Byte]] = null
+ var consumerConsumer: Consumer[Array[Byte], Array[Byte]] = null
+ 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 testNumPartitions = 2
+
+ client.createTopics(util.Arrays.asList(
+ new NewTopic(testTopicName, testNumPartitions, 1.toShort),
+ )).all.get
+ waitForTopics(client, List(testTopicName), List())
+
+ val producer = createProducer()
+ try {
+ producer.send(new ProducerRecord(testTopicName, 0, null, null))
+ producer.send(new ProducerRecord(testTopicName, 1, null, null))
+ producer.flush()
+ } finally {
+ Utils.closeQuietly(producer, "producer")
+ }
+
+ val testGroupId = "test_group_id"
+ val testClassicClientId = "test_classic_client_id"
+ val testConsumerClientId = "test_consumer_client_id"
+
+ val newConsumerConfig = new Properties(consumerConfig)
+ newConsumerConfig.put(ConsumerConfig.GROUP_ID_CONFIG, testGroupId)
+ newConsumerConfig.put(ConsumerConfig.CLIENT_ID_CONFIG, testClassicClientId)
+ consumerConfig.put(ConsumerConfig.GROUP_PROTOCOL_CONFIG, GroupProtocol.CLASSIC.name)
+
+ classicConsumer = createConsumer(configOverrides = newConsumerConfig)
+ classicConsumer.subscribe(List(testTopicName).asJava)
+ classicConsumer.poll(JDuration.ofMillis(1000))
+
+ newConsumerConfig.put(ConsumerConfig.CLIENT_ID_CONFIG, testConsumerClientId)
+ consumerConfig.put(ConsumerConfig.GROUP_PROTOCOL_CONFIG, GroupProtocol.CONSUMER.name)
+ consumerConsumer = createConsumer(configOverrides = newConsumerConfig)
+ consumerConsumer.subscribe(List(testTopicName).asJava)
+ consumerConsumer.poll(JDuration.ofMillis(1000))
+
+ TestUtils.waitUntilTrue(() => {
+ val describeConsumerGroupResult = client.describeConsumerGroups(Seq(testGroupId).asJava).all.get
+ describeConsumerGroupResult.containsKey(testGroupId) &&
+ describeConsumerGroupResult.get(testGroupId).members.size == 2
+ }, s"Expected the offset for partition 0 to eventually become 1.")
+
+ val describeConsumerGroupResult = client.describeConsumerGroups(Seq(testGroupId).asJava).all.get
+ val group = describeConsumerGroupResult.get(testGroupId)
+ assertNotNull(group)
+ assertTrue(group.groupEpoch.isPresent && group.groupEpoch.get > 0)
+ assertTrue(group.targetAssignmentEpoch.isPresent && group.targetAssignmentEpoch.get > 0)
+
+ val classicMember = group.members.asScala.find(_.clientId == testClassicClientId)
+ assertTrue(classicMember.isDefined)
+ assertTrue(classicMember.get.memberEpoch.isPresent && classicMember.get.memberEpoch.get > 0)
+ assertTrue(classicMember.get.upgraded.isPresent && !classicMember.get.upgraded.get)
+
+ val consumerMember = group.members.asScala.find(_.clientId == testConsumerClientId)
+ assertTrue(consumerMember.isDefined)
+ assertTrue(classicMember.get.memberEpoch.isPresent && classicMember.get.memberEpoch.get > 0)
+ assertTrue(consumerMember.get.upgraded.isPresent && consumerMember.get.upgraded.get)
Review Comment:
ditto.
##########
core/src/test/scala/integration/kafka/api/PlaintextAdminIntegrationTest.scala:
##########
@@ -2062,6 +2067,86 @@ class PlaintextAdminIntegrationTest extends BaseAdminIntegrationTest {
}
}
+ /**
+ * Test the consumer group APIs.
+ */
+ @ParameterizedTest
+ @ValueSource(strings = Array("kraft"))
+ def testConsumerGroupWithMemberMigration(quorum: String): Unit = {
+ val config = createConfig
+ client = Admin.create(config)
+ var classicConsumer: Consumer[Array[Byte], Array[Byte]] = null
+ var consumerConsumer: Consumer[Array[Byte], Array[Byte]] = null
+ 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 testNumPartitions = 2
+
+ client.createTopics(util.Arrays.asList(
+ new NewTopic(testTopicName, testNumPartitions, 1.toShort),
+ )).all.get
+ waitForTopics(client, List(testTopicName), List())
+
+ val producer = createProducer()
+ try {
+ producer.send(new ProducerRecord(testTopicName, 0, null, null))
+ producer.send(new ProducerRecord(testTopicName, 1, null, null))
+ producer.flush()
+ } finally {
+ Utils.closeQuietly(producer, "producer")
+ }
+
+ val testGroupId = "test_group_id"
+ val testClassicClientId = "test_classic_client_id"
+ val testConsumerClientId = "test_consumer_client_id"
+
+ val newConsumerConfig = new Properties(consumerConfig)
+ newConsumerConfig.put(ConsumerConfig.GROUP_ID_CONFIG, testGroupId)
+ newConsumerConfig.put(ConsumerConfig.CLIENT_ID_CONFIG, testClassicClientId)
+ consumerConfig.put(ConsumerConfig.GROUP_PROTOCOL_CONFIG, GroupProtocol.CLASSIC.name)
+
+ classicConsumer = createConsumer(configOverrides = newConsumerConfig)
+ classicConsumer.subscribe(List(testTopicName).asJava)
+ classicConsumer.poll(JDuration.ofMillis(1000))
+
+ newConsumerConfig.put(ConsumerConfig.CLIENT_ID_CONFIG, testConsumerClientId)
+ consumerConfig.put(ConsumerConfig.GROUP_PROTOCOL_CONFIG, GroupProtocol.CONSUMER.name)
+ consumerConsumer = createConsumer(configOverrides = newConsumerConfig)
+ consumerConsumer.subscribe(List(testTopicName).asJava)
+ consumerConsumer.poll(JDuration.ofMillis(1000))
+
+ TestUtils.waitUntilTrue(() => {
+ val describeConsumerGroupResult = client.describeConsumerGroups(Seq(testGroupId).asJava).all.get
+ describeConsumerGroupResult.containsKey(testGroupId) &&
+ describeConsumerGroupResult.get(testGroupId).members.size == 2
+ }, s"Expected the offset for partition 0 to eventually become 1.")
+
+ val describeConsumerGroupResult = client.describeConsumerGroups(Seq(testGroupId).asJava).all.get
+ val group = describeConsumerGroupResult.get(testGroupId)
+ assertNotNull(group)
+ assertTrue(group.groupEpoch.isPresent && group.groupEpoch.get > 0)
+ assertTrue(group.targetAssignmentEpoch.isPresent && group.targetAssignmentEpoch.get > 0)
+
+ val classicMember = group.members.asScala.find(_.clientId == testClassicClientId)
+ assertTrue(classicMember.isDefined)
+ assertTrue(classicMember.get.memberEpoch.isPresent && classicMember.get.memberEpoch.get > 0)
Review Comment:
`assertEquals(Optional.of(2), classicMember.get.memberEpoch)`.
##########
clients/src/main/java/org/apache/kafka/clients/admin/MemberDescription.java:
##########
@@ -131,13 +170,33 @@ public Optional<MemberAssignment> targetAssignment() {
return targetAssignment;
}
+ /**
+ * The epoch of the group member.
Review Comment:
nit: Let's also mention that this is only provided in the CONSUMER group case.
##########
core/src/test/scala/integration/kafka/api/PlaintextAdminIntegrationTest.scala:
##########
@@ -2062,6 +2067,86 @@ class PlaintextAdminIntegrationTest extends BaseAdminIntegrationTest {
}
}
+ /**
+ * Test the consumer group APIs.
+ */
+ @ParameterizedTest
+ @ValueSource(strings = Array("kraft"))
+ def testConsumerGroupWithMemberMigration(quorum: String): Unit = {
+ val config = createConfig
+ client = Admin.create(config)
+ var classicConsumer: Consumer[Array[Byte], Array[Byte]] = null
+ var consumerConsumer: Consumer[Array[Byte], Array[Byte]] = null
+ 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 testNumPartitions = 2
+
+ client.createTopics(util.Arrays.asList(
+ new NewTopic(testTopicName, testNumPartitions, 1.toShort),
+ )).all.get
+ waitForTopics(client, List(testTopicName), List())
+
+ val producer = createProducer()
+ try {
+ producer.send(new ProducerRecord(testTopicName, 0, null, null))
+ producer.send(new ProducerRecord(testTopicName, 1, null, null))
+ producer.flush()
+ } finally {
+ Utils.closeQuietly(producer, "producer")
+ }
+
+ val testGroupId = "test_group_id"
+ val testClassicClientId = "test_classic_client_id"
+ val testConsumerClientId = "test_consumer_client_id"
+
+ val newConsumerConfig = new Properties(consumerConfig)
+ newConsumerConfig.put(ConsumerConfig.GROUP_ID_CONFIG, testGroupId)
+ newConsumerConfig.put(ConsumerConfig.CLIENT_ID_CONFIG, testClassicClientId)
+ consumerConfig.put(ConsumerConfig.GROUP_PROTOCOL_CONFIG, GroupProtocol.CLASSIC.name)
+
+ classicConsumer = createConsumer(configOverrides = newConsumerConfig)
+ classicConsumer.subscribe(List(testTopicName).asJava)
+ classicConsumer.poll(JDuration.ofMillis(1000))
+
+ newConsumerConfig.put(ConsumerConfig.CLIENT_ID_CONFIG, testConsumerClientId)
+ consumerConfig.put(ConsumerConfig.GROUP_PROTOCOL_CONFIG, GroupProtocol.CONSUMER.name)
+ consumerConsumer = createConsumer(configOverrides = newConsumerConfig)
+ consumerConsumer.subscribe(List(testTopicName).asJava)
+ consumerConsumer.poll(JDuration.ofMillis(1000))
+
+ TestUtils.waitUntilTrue(() => {
+ val describeConsumerGroupResult = client.describeConsumerGroups(Seq(testGroupId).asJava).all.get
+ describeConsumerGroupResult.containsKey(testGroupId) &&
+ describeConsumerGroupResult.get(testGroupId).members.size == 2
+ }, s"Expected the offset for partition 0 to eventually become 1.")
+
+ val describeConsumerGroupResult = client.describeConsumerGroups(Seq(testGroupId).asJava).all.get
+ val group = describeConsumerGroupResult.get(testGroupId)
+ assertNotNull(group)
+ assertTrue(group.groupEpoch.isPresent && group.groupEpoch.get > 0)
Review Comment:
It should be 2 if you implement my previous suggestions.
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-17750: Extend kafka-consumer-groups command line tool to support new consumer group (part 2) [kafka]
Posted by "FrankYang0529 (via GitHub)" <gi...@apache.org>.
FrankYang0529 commented on PR #18034:
URL: https://github.com/apache/kafka/pull/18034#issuecomment-2530894904
Hi @dajac, could you please help me review this PR when you have time? I would like to keep working on last part of KIP-1099 (tool). Thanks.
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-17750: Extend kafka-consumer-groups command line tool to support new consumer group (part 2) [kafka]
Posted by "dajac (via GitHub)" <gi...@apache.org>.
dajac merged PR #18034:
URL: https://github.com/apache/kafka/pull/18034
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-17750: Extend kafka-consumer-groups command line tool to support new consumer group (part 2) [kafka]
Posted by "FrankYang0529 (via GitHub)" <gi...@apache.org>.
FrankYang0529 commented on code in PR #18034:
URL: https://github.com/apache/kafka/pull/18034#discussion_r1871644688
##########
clients/src/main/java/org/apache/kafka/clients/admin/MemberDescription.java:
##########
@@ -131,13 +140,30 @@ public Optional<MemberAssignment> targetAssignment() {
return targetAssignment;
}
+ /**
+ * The epoch of the group member.
+ */
+ public Optional<Integer> memberEpoch() {
+ return memberEpoch;
+ }
+
+ /**
+ * The flag indicating whether a member is classic.
+ */
+ public Optional<Boolean> isClassic() {
Review Comment:
Thanks for the suggestion. Updated PR and KIP.
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-17750: Extend kafka-consumer-groups command line tool to support new consumer group (part 2) [kafka]
Posted by "FrankYang0529 (via GitHub)" <gi...@apache.org>.
FrankYang0529 commented on code in PR #18034:
URL: https://github.com/apache/kafka/pull/18034#discussion_r1874333440
##########
clients/src/test/java/org/apache/kafka/clients/admin/internals/DescribeConsumerGroupsHandlerTest.java:
##########
@@ -164,17 +165,21 @@ public void testSuccessfulHandleConsumerGroupResponse() {
Optional.of(new MemberAssignment(Set.of(
new TopicPartition("foo", 1),
new TopicPartition("bar", 2)
- )))
+ ))),
+ Optional.of(10),
+ Optional.of(true)
Review Comment:
Add classic case to `testSuccessfulHandleConsumerGroupResponse`. For unknown case, I think it's covered by `testSuccessfulHandleClassicGroupResponse`. Thanks.
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-17750: Extend kafka-consumer-groups command line tool to support new consumer group (part 2) [kafka]
Posted by "FrankYang0529 (via GitHub)" <gi...@apache.org>.
FrankYang0529 commented on PR #18034:
URL: https://github.com/apache/kafka/pull/18034#issuecomment-2535963607
> Could you please file a minor to fix the deprecation warning introduced by this PR?
Thanks for finding this. Create a PR for it.
https://github.com/apache/kafka/pull/18139
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-17750: Extend kafka-consumer-groups command line tool to support new consumer group (part 2) [kafka]
Posted by "chia7712 (via GitHub)" <gi...@apache.org>.
chia7712 commented on code in PR #18034:
URL: https://github.com/apache/kafka/pull/18034#discussion_r1872062292
##########
clients/src/main/java/org/apache/kafka/clients/admin/MemberDescription.java:
##########
@@ -131,13 +158,33 @@ public Optional<MemberAssignment> targetAssignment() {
return targetAssignment;
}
+ /**
+ * The epoch of the group member.
+ */
+ public Optional<Integer> memberEpoch() {
+ return memberEpoch;
+ }
+
+ /**
+ * The flag indicating whether a member with a consumer group uses the consumer protocol.
+ * True for consumer member.
+ * False for classic member
+ * Empty for unknown or classic member in classic group.
+ */
+ public Optional<Boolean> upgraded() {
+ return upgraded;
+ }
+
@Override
public String toString() {
return "(memberId=" + memberId +
", groupInstanceId=" + groupInstanceId.orElse("null") +
", clientId=" + clientId +
", host=" + host +
", assignment=" + assignment +
- ", targetAssignment=" + targetAssignment + ")";
+ ", targetAssignment=" + targetAssignment +
+ ", memberEpoch=" + memberEpoch.orElse(null) +
+ ", isClassic=" + upgraded.orElse(null) +
Review Comment:
`isClassic` -> `upgraded`
##########
clients/src/main/java/org/apache/kafka/clients/admin/MemberDescription.java:
##########
@@ -30,21 +30,26 @@ public class MemberDescription {
private final String host;
private final MemberAssignment assignment;
private final Optional<MemberAssignment> targetAssignment;
+ private final Optional<Integer> memberEpoch;
+ private final Optional<Boolean> isClassic;
public MemberDescription(String memberId,
Review Comment:
Have we considered deprecating the other constructors? It seems odd to have so many constructors.
##########
clients/src/main/java/org/apache/kafka/clients/admin/ConsumerGroupDescription.java:
##########
@@ -108,7 +113,8 @@ public ConsumerGroupDescription(String groupId,
GroupState groupState,
Node coordinator,
Set<AclOperation> authorizedOperations) {
- this(groupId, isSimpleConsumerGroup, members, partitionAssignor, GroupType.CLASSIC, groupState, coordinator, authorizedOperations);
+ this(groupId, isSimpleConsumerGroup, members, partitionAssignor, GroupType.CLASSIC, groupState, coordinator, authorizedOperations,
+ Optional.empty(), Optional.empty());
}
public ConsumerGroupDescription(String groupId,
Review Comment:
We can remove the two constructors directly, as doing so does not result in any compatibility issues.
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-17750: Extend kafka-consumer-groups command line tool to support new consumer group (part 2) [kafka]
Posted by "FrankYang0529 (via GitHub)" <gi...@apache.org>.
FrankYang0529 commented on code in PR #18034:
URL: https://github.com/apache/kafka/pull/18034#discussion_r1873368429
##########
core/src/test/scala/integration/kafka/api/PlaintextAdminIntegrationTest.scala:
##########
@@ -2062,6 +2067,86 @@ class PlaintextAdminIntegrationTest extends BaseAdminIntegrationTest {
}
}
+ /**
+ * Test the consumer group APIs.
+ */
+ @ParameterizedTest
+ @ValueSource(strings = Array("kraft"))
+ def testConsumerGroupWithMemberMigration(quorum: String): Unit = {
+ val config = createConfig
+ client = Admin.create(config)
+ var classicConsumer: Consumer[Array[Byte], Array[Byte]] = null
+ var consumerConsumer: Consumer[Array[Byte], Array[Byte]] = null
+ 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 testNumPartitions = 2
+
+ client.createTopics(util.Arrays.asList(
+ new NewTopic(testTopicName, testNumPartitions, 1.toShort),
+ )).all.get
+ waitForTopics(client, List(testTopicName), List())
+
+ val producer = createProducer()
+ try {
+ producer.send(new ProducerRecord(testTopicName, 0, null, null))
+ producer.send(new ProducerRecord(testTopicName, 1, null, null))
+ producer.flush()
+ } finally {
+ Utils.closeQuietly(producer, "producer")
+ }
+
+ val testGroupId = "test_group_id"
+ val testClassicClientId = "test_classic_client_id"
+ val testConsumerClientId = "test_consumer_client_id"
+
+ val newConsumerConfig = new Properties(consumerConfig)
+ newConsumerConfig.put(ConsumerConfig.GROUP_ID_CONFIG, testGroupId)
+ newConsumerConfig.put(ConsumerConfig.CLIENT_ID_CONFIG, testClassicClientId)
+ consumerConfig.put(ConsumerConfig.GROUP_PROTOCOL_CONFIG, GroupProtocol.CLASSIC.name)
+
+ classicConsumer = createConsumer(configOverrides = newConsumerConfig)
+ classicConsumer.subscribe(List(testTopicName).asJava)
+ classicConsumer.poll(JDuration.ofMillis(1000))
+
+ newConsumerConfig.put(ConsumerConfig.CLIENT_ID_CONFIG, testConsumerClientId)
+ consumerConfig.put(ConsumerConfig.GROUP_PROTOCOL_CONFIG, GroupProtocol.CONSUMER.name)
+ consumerConsumer = createConsumer(configOverrides = newConsumerConfig)
+ consumerConsumer.subscribe(List(testTopicName).asJava)
+ consumerConsumer.poll(JDuration.ofMillis(1000))
+
+ TestUtils.waitUntilTrue(() => {
+ val describeConsumerGroupResult = client.describeConsumerGroups(Seq(testGroupId).asJava).all.get
+ describeConsumerGroupResult.containsKey(testGroupId) &&
+ describeConsumerGroupResult.get(testGroupId).members.size == 2
Review Comment:
Thanks for the suggestion. I change to wait until group becomes RECONCILING, because the group is during migration.
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-17750: Extend kafka-consumer-groups command line tool to support new consumer group (part 2) [kafka]
Posted by "FrankYang0529 (via GitHub)" <gi...@apache.org>.
FrankYang0529 commented on code in PR #18034:
URL: https://github.com/apache/kafka/pull/18034#discussion_r1873370363
##########
core/src/test/scala/integration/kafka/api/PlaintextAdminIntegrationTest.scala:
##########
@@ -1921,12 +1921,17 @@ class PlaintextAdminIntegrationTest extends BaseAdminIntegrationTest {
// Test that we can get information about the test consumer group.
assertTrue(describeWithFakeGroupResult.describedGroups().containsKey(testGroupId))
var testGroupDescription = describeWithFakeGroupResult.describedGroups().get(testGroupId).get()
+ assertEquals(groupType == GroupType.CLASSIC, testGroupDescription.groupEpoch.isEmpty)
+ assertEquals(groupType == GroupType.CLASSIC, testGroupDescription.targetAssignmentEpoch.isEmpty)
Review Comment:
I would like to verify values here, but the case is only for classic now. It's waiting for https://issues.apache.org/jira/browse/KAFKA-17960. Could we add it after we reopen consumer for this case? Thanks.
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-17750: Extend kafka-consumer-groups command line tool to support new consumer group (part 2) [kafka]
Posted by "dajac (via GitHub)" <gi...@apache.org>.
dajac commented on code in PR #18034:
URL: https://github.com/apache/kafka/pull/18034#discussion_r1875652264
##########
core/src/test/scala/integration/kafka/api/PlaintextAdminIntegrationTest.scala:
##########
@@ -1921,12 +1921,17 @@ class PlaintextAdminIntegrationTest extends BaseAdminIntegrationTest {
// Test that we can get information about the test consumer group.
assertTrue(describeWithFakeGroupResult.describedGroups().containsKey(testGroupId))
var testGroupDescription = describeWithFakeGroupResult.describedGroups().get(testGroupId).get()
+ assertEquals(groupType == GroupType.CLASSIC, testGroupDescription.groupEpoch.isEmpty)
+ assertEquals(groupType == GroupType.CLASSIC, testGroupDescription.targetAssignmentEpoch.isEmpty)
Review Comment:
Sure.
##########
core/src/test/scala/integration/kafka/api/PlaintextAdminIntegrationTest.scala:
##########
@@ -2062,6 +2067,86 @@ class PlaintextAdminIntegrationTest extends BaseAdminIntegrationTest {
}
}
+ /**
+ * Test the consumer group APIs.
+ */
+ @ParameterizedTest
+ @ValueSource(strings = Array("kraft"))
+ def testConsumerGroupWithMemberMigration(quorum: String): Unit = {
+ val config = createConfig
+ client = Admin.create(config)
+ var classicConsumer: Consumer[Array[Byte], Array[Byte]] = null
+ var consumerConsumer: Consumer[Array[Byte], Array[Byte]] = null
+ 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 testNumPartitions = 2
+
+ client.createTopics(util.Arrays.asList(
+ new NewTopic(testTopicName, testNumPartitions, 1.toShort),
+ )).all.get
+ waitForTopics(client, List(testTopicName), List())
+
+ val producer = createProducer()
+ try {
+ producer.send(new ProducerRecord(testTopicName, 0, null, null))
+ producer.send(new ProducerRecord(testTopicName, 1, null, null))
+ producer.flush()
+ } finally {
+ Utils.closeQuietly(producer, "producer")
+ }
+
+ val testGroupId = "test_group_id"
+ val testClassicClientId = "test_classic_client_id"
+ val testConsumerClientId = "test_consumer_client_id"
+
+ val newConsumerConfig = new Properties(consumerConfig)
+ newConsumerConfig.put(ConsumerConfig.GROUP_ID_CONFIG, testGroupId)
+ newConsumerConfig.put(ConsumerConfig.CLIENT_ID_CONFIG, testClassicClientId)
+ consumerConfig.put(ConsumerConfig.GROUP_PROTOCOL_CONFIG, GroupProtocol.CLASSIC.name)
+
+ classicConsumer = createConsumer(configOverrides = newConsumerConfig)
+ classicConsumer.subscribe(List(testTopicName).asJava)
+ classicConsumer.poll(JDuration.ofMillis(1000))
+
+ newConsumerConfig.put(ConsumerConfig.CLIENT_ID_CONFIG, testConsumerClientId)
+ consumerConfig.put(ConsumerConfig.GROUP_PROTOCOL_CONFIG, GroupProtocol.CONSUMER.name)
+ consumerConsumer = createConsumer(configOverrides = newConsumerConfig)
+ consumerConsumer.subscribe(List(testTopicName).asJava)
+ consumerConsumer.poll(JDuration.ofMillis(1000))
+
+ TestUtils.waitUntilTrue(() => {
+ val describeConsumerGroupResult = client.describeConsumerGroups(Seq(testGroupId).asJava).all.get
+ describeConsumerGroupResult.containsKey(testGroupId) &&
+ describeConsumerGroupResult.get(testGroupId).members.size == 2
Review Comment:
Why RECONCILING? The group should eventually switch to STABLE anyway. Could you please elaborate?
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-17750: Extend kafka-consumer-groups command line tool to support new consumer group (part 2) [kafka]
Posted by "chia7712 (via GitHub)" <gi...@apache.org>.
chia7712 commented on PR #18034:
URL: https://github.com/apache/kafka/pull/18034#issuecomment-2535952082
@FrankYang0529 Could you please file a minor to fix the deprecation warning introduced by this PR?
```
> Task :tools:compileTestJava
Note: /home/chia7712/project/kafka/tools/src/test/java/org/apache/kafka/tools/consumer/group/ConsumerGroupServiceTest.java uses or overrides a deprecated API.
Note: Recompile with -Xlint:deprecation for details.
```
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-17750: Extend kafka-consumer-groups command line tool to support new consumer group (part 2) [kafka]
Posted by "FrankYang0529 (via GitHub)" <gi...@apache.org>.
FrankYang0529 commented on code in PR #18034:
URL: https://github.com/apache/kafka/pull/18034#discussion_r1886058081
##########
core/src/test/scala/integration/kafka/api/PlaintextAdminIntegrationTest.scala:
##########
@@ -1921,12 +1921,17 @@ class PlaintextAdminIntegrationTest extends BaseAdminIntegrationTest {
// Test that we can get information about the test consumer group.
assertTrue(describeWithFakeGroupResult.describedGroups().containsKey(testGroupId))
var testGroupDescription = describeWithFakeGroupResult.describedGroups().get(testGroupId).get()
+ assertEquals(groupType == GroupType.CLASSIC, testGroupDescription.groupEpoch.isEmpty)
+ assertEquals(groupType == GroupType.CLASSIC, testGroupDescription.targetAssignmentEpoch.isEmpty)
Review Comment:
Create a minor PR for this: https://github.com/apache/kafka/pull/18203
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org