This is an automated email from the ASF dual-hosted git repository.
mjsax pushed a commit to branch trunk
in repository https://gitbox.apache.org/repos/asf/kafka.git
The following commit(s) were added to refs/heads/trunk by this push:
new a6fef7fd25b KAFKA-20885: Fix IQ metadata for source topics with fewer
partitions (#23050)
a6fef7fd25b is described below
commit a6fef7fd25b06454ef30f89fb1bf1e327d5f3093
Author: Matthias J. Sax <[email protected]>
AuthorDate: Mon Aug 3 18:34:44 2026 -0700
KAFKA-20885: Fix IQ metadata for source topics with fewer partitions
(#23050)
EndpointToPartitionsManager clamped the partitions it reports for a
member by comparing the topic's partition count against the *number* of
tasks the member owns, and then keeping the lowest task IDs. That is
only correct when a single member owns the whole subtopology.
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. When a member owns a "non-prefix" task set, the size-based
check does not fire and the task IDs are reported verbatim as partitions
of that smaller topic, advertising partitions that do not exist.
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.
Reviewers: Bill Bejeck <[email protected]>
---
.../topics/EndpointToPartitionsManager.java | 36 +++++++++++---------
.../topics/EndpointToPartitionsManagerTest.java | 39 ++++++++++++++++++++++
2 files changed, 60 insertions(+), 15 deletions(-)
diff --git
a/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/streams/topics/EndpointToPartitionsManager.java
b/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/streams/topics/EndpointToPartitionsManager.java
index 455a584da40..0798eddf45c 100644
---
a/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/streams/topics/EndpointToPartitionsManager.java
+++
b/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/streams/topics/EndpointToPartitionsManager.java
@@ -24,7 +24,6 @@ import
org.apache.kafka.coordinator.group.streams.StreamsGroup;
import org.apache.kafka.coordinator.group.streams.StreamsGroupMember;
import java.util.ArrayList;
-import java.util.Collections;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
@@ -100,25 +99,32 @@ public class EndpointToPartitionsManager {
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(partitionId -> partitionId < numPartitionsForTopic)
+ .sorted()
+ .toList();
+ if (!partitions.isEmpty()) {
+ StreamsGroupHeartbeatResponseData.TopicPartition tp = new
StreamsGroupHeartbeatResponseData.TopicPartition();
+ tp.setTopic(topic);
+ tp.setPartitions(partitions);
+ topicPartitionsForTask.add(tp);
}
- return tp;
- }).toList();
+ }
+ return topicPartitionsForTask;
}
}
\ No newline at end of file
diff --git
a/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/streams/topics/EndpointToPartitionsManagerTest.java
b/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/streams/topics/EndpointToPartitionsManagerTest.java
index 97af1bd9ccf..ffe05c3e40d 100644
---
a/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/streams/topics/EndpointToPartitionsManagerTest.java
+++
b/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/streams/topics/EndpointToPartitionsManagerTest.java
@@ -113,6 +113,45 @@ class EndpointToPartitionsManagerTest {
assertTopicPartitionsAssigned(standbyPartitions, "Topic-B");
}
+ @Test
+ void
testEndpointToPartitionsExcludesPartitionsMissingFromSmallerSourceTopic() {
+ MetadataImage metadataImage = new MetadataImageBuilder()
+ .addTopic(Uuid.randomUuid(), "Topic-A", 5)
+ .addTopic(Uuid.randomUuid(), "Topic-B", 2)
+ .build();
+ configuredSubtopologyOne = new ConfiguredSubtopology(5,
Set.of("Topic-A", "Topic-B"), new HashMap<>(), new HashSet<>(), new
HashMap<>());
+
+ when(streamsGroupMember.assignedTasks()).thenReturn(
+ new TasksTupleWithEpochs(
+ mkTasksPerSubtopologyWithCommonEpoch(0, mkEntry("0", Set.of(1,
3))),
+ mkTasksPerSubtopology(mkEntry("0", Set.of(4))),
+ Map.of()
+ )
+ );
+
when(streamsGroup.configuredTopology()).thenReturn(Optional.of(configuredTopology));
+ SortedMap<String, ConfiguredSubtopology> configuredSubtopologyMap =
new TreeMap<>();
+ configuredSubtopologyMap.put("0", configuredSubtopologyOne);
+
when(configuredTopology.subtopologies()).thenReturn(Optional.of(configuredSubtopologyMap));
+
+ StreamsGroupHeartbeatResponseData.EndpointToPartitions result =
+
EndpointToPartitionsManager.endpointToPartitions(streamsGroupMember,
responseEndpoint, streamsGroup, new
KRaftCoordinatorMetadataImage(metadataImage));
+
+ List<StreamsGroupHeartbeatResponseData.TopicPartition>
activePartitions = result.activePartitions();
+
activePartitions.sort(Comparator.comparing(StreamsGroupHeartbeatResponseData.TopicPartition::topic));
+ assertEquals(2, activePartitions.size());
+ assertEquals("Topic-A", activePartitions.get(0).topic());
+ assertEquals(List.of(1, 3), activePartitions.get(0).partitions());
+ // Topic-B has only partitions 0 and 1, so active task 3 contributes
no Topic-B partition.
+ assertEquals("Topic-B", activePartitions.get(1).topic());
+ assertEquals(List.of(1), activePartitions.get(1).partitions());
+
+ // Standby task 4 is beyond the last partition of Topic-B, so Topic-B
is not reported as standby at all.
+ List<StreamsGroupHeartbeatResponseData.TopicPartition>
standbyPartitions = result.standbyPartitions();
+ assertEquals(1, standbyPartitions.size());
+ assertEquals("Topic-A", standbyPartitions.get(0).topic());
+ assertEquals(List.of(4), standbyPartitions.get(0).partitions());
+ }
+
private static void
assertTopicPartitionsAssigned(List<StreamsGroupHeartbeatResponseData.TopicPartition>
topicPartitions, String topicName) {
StreamsGroupHeartbeatResponseData.TopicPartition topicPartition =
topicPartitions.stream().filter(tp ->
tp.topic().equals(topicName)).findFirst().get();
assertEquals(topicName, topicPartition.topic());
