You are viewing a plain text version of this content. The canonical link for it is here.
Posted to jira@kafka.apache.org by "mjsax (via GitHub)" <gi...@apache.org> on 2026/08/03 00:22:51 UTC
[PR] KAFKA-20885: Fix IQ metadata for source topics with fewer partitions [kafka]
mjsax opened a new pull request, #23050:
URL: https://github.com/apache/kafka/pull/23050
A subtopology has as many tasks as its source topic with the most partitions, so a smaller
source topic has no partition for the higher task IDs.
`EndpointToPartitionsManager` clamped the reported partitions by comparing the topic's
partition count against the *number* of tasks the member owns, then keeping the lowest task
IDs. That only coincides with the correct answer when a single member owns the whole
subtopology. For a member owning a non-prefix task set the check does not fire, and the task
IDs are reported verbatim as partitions of the smaller topic — advertising partitions that do
not exist.
Example: subtopology reads `A` (5 partitions) and `B` (2 partitions), member owns tasks `0_1`
and `0_3`. For `B` it should report only `B-1`, since task `0_3` has no `B` partition. It
reports `B-1` and `B-3`.
Filter the task IDs by value, and omit a source topic entirely when none of the member's tasks
map to one of its partitions. This mirrors `PartitionGrouper#partitionGroups`, which computes
the same mapping for the classic protocol.
### Testing
Two unit tests in `EndpointToPartitionsManagerTest`, both failing before the fix:
- a non-prefix task set where the smaller topic keeps some partitions
- a task set entirely beyond the smaller topic, which is then not reported at all
The existing coverage only exercised a single member owning the whole subtopology, which is
the one shape where the old check happened to be right.
Unblocks #23006, which otherwise cannot assert correct output for a realistic multi-member
task distribution.
🤖 Generated with [Claude Code](https://claude.com/claude-code)
--
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-20885: Fix IQ metadata for source topics with fewer partitions [kafka]
Posted by "mjsax (via GitHub)" <gi...@apache.org>.
mjsax merged PR #23050:
URL: https://github.com/apache/kafka/pull/23050
--
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-20885: Fix IQ metadata for source topics with fewer partitions [kafka]
Posted by "mjsax (via GitHub)" <gi...@apache.org>.
mjsax commented on PR #23050:
URL: https://github.com/apache/kafka/pull/23050#issuecomment-5173638289
Merged to `trunk` and cherry-picked to `4.3` and `4.2` branches.
--
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-20885: Fix IQ metadata for source topics with fewer partitions [kafka]
Posted by "mjsax (via GitHub)" <gi...@apache.org>.
mjsax commented on code in PR #23050:
URL: https://github.com/apache/kafka/pull/23050#discussion_r3707553256
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/streams/topics/EndpointToPartitionsManager.java:
##########
@@ -100,25 +99,32 @@ private static List<StreamsGroupHeartbeatResponseData.TopicPartition> topicParti
return topicPartitionsForTasks;
}
- private static List<StreamsGroupHeartbeatResponseData.TopicPartition> topicPartitionListForTask(final Set<Integer> taskSet,
- final Set<String> topicNames,
- final CoordinatorMetadataImage metadataImage) {
- return topicNames.stream().map(topic -> {
+ private static List<StreamsGroupHeartbeatResponseData.TopicPartition> topicPartitionListForTask(
+ final Set<Integer> taskSet,
+ final Set<String> topicNames,
+ final CoordinatorMetadataImage metadataImage
+ ) {
+ List<StreamsGroupHeartbeatResponseData.TopicPartition> topicPartitionsForTask = new ArrayList<>();
+ for (String topic : topicNames) {
Optional<CoordinatorMetadataImage.TopicMetadata> topicMetadata = metadataImage.topicMetadata(topic);
if (topicMetadata.isEmpty()) {
throw new IllegalStateException("Topic " + topic + " not found in metadata image");
}
int numPartitionsForTopic = topicMetadata.get().partitionCount();
- StreamsGroupHeartbeatResponseData.TopicPartition tp = new StreamsGroupHeartbeatResponseData.TopicPartition();
- tp.setTopic(topic);
- List<Integer> tpPartitions = new ArrayList<>(taskSet);
- if (numPartitionsForTopic < taskSet.size()) {
- Collections.sort(tpPartitions);
- tp.setPartitions(tpPartitions.subList(0, numPartitionsForTopic));
- } else {
- tp.setPartitions(tpPartitions);
+ // A subtopology has as many tasks as its source topic with the most partitions. A source topic with
+ // fewer partitions therefore has no partition for the higher task IDs, mirroring the client-side
+ // grouping in PartitionGrouper#partitionGroups.
+ List<Integer> partitions = taskSet.stream()
+ .filter(taskId -> taskId < numPartitionsForTopic)
Review Comment:
```suggestion
.filter(partitionId -> partitionId < numPartitionsForTopic)
```
--
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-20885: Fix IQ metadata for source topics with fewer partitions [kafka]
Posted by "bbejeck (via GitHub)" <gi...@apache.org>.
bbejeck commented on code in PR #23050:
URL: https://github.com/apache/kafka/pull/23050#discussion_r3707504266
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/streams/topics/EndpointToPartitionsManager.java:
##########
@@ -100,25 +99,32 @@ private static List<StreamsGroupHeartbeatResponseData.TopicPartition> topicParti
return topicPartitionsForTasks;
}
- private static List<StreamsGroupHeartbeatResponseData.TopicPartition> topicPartitionListForTask(final Set<Integer> taskSet,
- final Set<String> topicNames,
- final CoordinatorMetadataImage metadataImage) {
- return topicNames.stream().map(topic -> {
+ private static List<StreamsGroupHeartbeatResponseData.TopicPartition> topicPartitionListForTask(
+ final Set<Integer> taskSet,
+ final Set<String> topicNames,
+ final CoordinatorMetadataImage metadataImage
+ ) {
+ List<StreamsGroupHeartbeatResponseData.TopicPartition> topicPartitionsForTask = new ArrayList<>();
+ for (String topic : topicNames) {
Optional<CoordinatorMetadataImage.TopicMetadata> topicMetadata = metadataImage.topicMetadata(topic);
if (topicMetadata.isEmpty()) {
throw new IllegalStateException("Topic " + topic + " not found in metadata image");
}
int numPartitionsForTopic = topicMetadata.get().partitionCount();
- StreamsGroupHeartbeatResponseData.TopicPartition tp = new StreamsGroupHeartbeatResponseData.TopicPartition();
- tp.setTopic(topic);
- List<Integer> tpPartitions = new ArrayList<>(taskSet);
- if (numPartitionsForTopic < taskSet.size()) {
- Collections.sort(tpPartitions);
- tp.setPartitions(tpPartitions.subList(0, numPartitionsForTopic));
- } else {
- tp.setPartitions(tpPartitions);
+ // A subtopology has as many tasks as its source topic with the most partitions. A source topic with
+ // fewer partitions therefore has no partition for the higher task IDs, mirroring the client-side
+ // grouping in PartitionGrouper#partitionGroups.
+ List<Integer> partitions = taskSet.stream()
+ .filter(taskId -> taskId < numPartitionsForTopic)
Review Comment:
nit: maybe change `taskId` -> `partitionId`
--
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