KAFKA-18614, KAFKA-18613: Add streams group request plumbing #18979
Conversation
lucasbru
force-pushed
the
kip1071merge/streams_group_requests
branch
2 times, most recently
from
492f494 to
8b5a439
Compare
| @@ -1675,6 +1675,12 @@ class KafkaConfigTest { | |||
| assertEquals(Set(GroupType.CLASSIC, GroupType.CONSUMER, GroupType.SHARE), config.groupCoordinatorRebalanceProtocols) | |||
| assertTrue(config.isNewGroupCoordinatorEnabled) | |||
| assertTrue(config.shareGroupConfig.isShareGroupEnabled) | |||
|
|
|||
| // This is OK. | |||
There was a problem hiding this comment.
What's this for - can we remove it?
There was a problem hiding this comment.
You are right, can be removed. I was just following the style in rest of the test.
| * @param groupIds The IDs of the groups to describe. | ||
| * @param committedOffset A specified committed offset corresponding to this shard. | ||
| * | ||
| * @return A list containing the StreamsGroupDescribeResponseData.DescribedGroup. |
There was a problem hiding this comment.
nit: maybe update the javadoc comment that returned list will also contain any errors describing the group or something to that effect, but I'm not sure if this is valid request or not.
| @Test | ||
| public void testStreamsGroupDescribeWithErrors() { | ||
| String groupId = "groupId"; | ||
|
|
There was a problem hiding this comment.
nit: don't need a blank line here
|
|
||
| final List<CompletableFuture<List<StreamsGroupDescribeResponseData.DescribedGroup>>> futures = | ||
| new ArrayList<>(groupIds.size()); | ||
| final Map<TopicPartition, List<String>> groupsByTopicPartition = new HashMap<>(); |
There was a problem hiding this comment.
more of a general question - why track groups by TopicPartition?
There was a problem hiding this comment.
The group coordinator is sharded by topic partition of the consumer offset topic. So we group the group IDs by the topic partitions of the consumer offset, which acts as an "address" of the right group coordinator. We fetch the described groups by instance and merge the results.
lucasbru
force-pushed
the
kip1071merge/streams_group_requests
branch
from
8b5a439 to
fc9f1ad
Compare
| @@ -1675,6 +1675,12 @@ class KafkaConfigTest { | |||
| assertEquals(Set(GroupType.CLASSIC, GroupType.CONSUMER, GroupType.SHARE), config.groupCoordinatorRebalanceProtocols) | |||
| assertTrue(config.isNewGroupCoordinatorEnabled) | |||
| assertTrue(config.shareGroupConfig.isShareGroupEnabled) | |||
|
|
|||
| // This is OK. | |||
There was a problem hiding this comment.
You are right, can be removed. I was just following the style in rest of the test.
|
|
||
| final List<CompletableFuture<List<StreamsGroupDescribeResponseData.DescribedGroup>>> futures = | ||
| new ArrayList<>(groupIds.size()); | ||
| final Map<TopicPartition, List<String>> groupsByTopicPartition = new HashMap<>(); |
There was a problem hiding this comment.
The group coordinator is sharded by topic partition of the consumer offset topic. So we group the group IDs by the topic partitions of the consumer offset, which acts as an "address" of the right group coordinator. We fetch the described groups by instance and merge the results.
| * @param groupIds The IDs of the groups to describe. | ||
| * @param committedOffset A specified committed offset corresponding to this shard. | ||
| * | ||
| * @return A list containing the StreamsGroupDescribeResponseData.DescribedGroup. |
| @Test | ||
| public void testStreamsGroupDescribeWithErrors() { | ||
| String groupId = "groupId"; | ||
|
|
This change implements the basic RPC handling StreamsGroupHeartbeat and StreamsGroupDescribe. This includes:
The heartbeat implementation inside the
GroupMetadataManager, which actually implements the assignment / reconciliation logic, will come in a follow-up PR. Also, automatic creation of internal topics will be created in a follow-up PR.Committer Checklist (excluded from commit message)