You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@kafka.apache.org by sq...@apache.org on 2026/10/24 11:40:10 UTC
(kafka) branch trunk updated: KAFKA-21318, KAFKA-20292: Report negative partitions as unassigned in GroupSpecImpl (#23936)
This is an automated email from the ASF dual-hosted git repository.
squah-confluent 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 647466c056c KAFKA-21318, KAFKA-20292: Report negative partitions as unassigned in GroupSpecImpl (#23936)
647466c056c is described below
commit 647466c056c0107853cc21b51d7bd2d869ea64d9
Author: Sean Quah <sq...@confluent.io>
AuthorDate: Sat Oct 24 12:40:01 2026 +0100
KAFKA-21318, KAFKA-20292: Report negative partitions as unassigned in GroupSpecImpl (#23936)
GroupSpec is part of the public assignor API, so custom assignors may
call isPartitionAssigned with a negative partition, or may have produced
a target assignment containing one. isPartitionAssigned threw an
IndexOutOfBoundsException in both cases. Report negative partitions as
unassigned, like other partitions which do not exist, and update the
tests.
Also update the javadoc of GroupSpecImpl's metadata image, and assign it
first in the constructor.
Co-Authored-By: Claude Opus 5.5 <no...@anthropic.com>
Reviewers: David Jacot <da...@apache.org>
---
.../kafka/coordinator/group/modern/GroupSpecImpl.java | 8 ++++----
.../coordinator/group/assignor/GroupSpecImplTest.java | 19 ++++++++++++++-----
2 files changed, 18 insertions(+), 9 deletions(-)
diff --git a/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/modern/GroupSpecImpl.java b/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/modern/GroupSpecImpl.java
index 0fae371cff8..0fcd9c0c19d 100644
--- a/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/modern/GroupSpecImpl.java
+++ b/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/modern/GroupSpecImpl.java
@@ -44,7 +44,7 @@ public class GroupSpecImpl implements GroupSpec {
private record AssignedPartitions(BitSet partitions, int partitionCount) { }
/**
- * The metadata image, used to size the assigned partitions of each topic.
+ * The metadata image, used to look up the partition count of each topic.
*/
private final CoordinatorMetadataImage metadataImage;
@@ -85,10 +85,10 @@ public class GroupSpecImpl implements GroupSpec {
SubscriptionType subscriptionType,
Optional<Map<Uuid, Set<Integer>>> topicPartitionAllowedMap
) {
+ this.metadataImage = Objects.requireNonNull(metadataImage);
this.members = Objects.requireNonNull(members);
this.subscriptionType = Objects.requireNonNull(subscriptionType);
this.topicPartitionAllowedMap = Objects.requireNonNull(topicPartitionAllowedMap);
- this.metadataImage = Objects.requireNonNull(metadataImage);
}
/**
@@ -118,7 +118,7 @@ public class GroupSpecImpl implements GroupSpec {
assignedTopicPartitions = computeAssignedTopicPartitions();
}
AssignedPartitions assignedPartitions = assignedTopicPartitions.get(topicId);
- return assignedPartitions != null && assignedPartitions.partitions().get(partitionId);
+ return assignedPartitions != null && partitionId >= 0 && assignedPartitions.partitions().get(partitionId);
}
private Map<Uuid, AssignedPartitions> computeAssignedTopicPartitions() {
@@ -135,7 +135,7 @@ public class GroupSpecImpl implements GroupSpec {
}
);
for (int partition : entry.getValue()) {
- if (partition >= assignedPartitions.partitionCount()) {
+ if (partition < 0 || partition >= assignedPartitions.partitionCount()) {
// Drop partitions that do not exist.
continue;
}
diff --git a/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/assignor/GroupSpecImplTest.java b/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/assignor/GroupSpecImplTest.java
index fa8ec725a32..415ab1ae58a 100644
--- a/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/assignor/GroupSpecImplTest.java
+++ b/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/assignor/GroupSpecImplTest.java
@@ -81,16 +81,25 @@ public class GroupSpecImplTest {
@Test
void testIsPartitionAssigned() {
+ Uuid deletedTopicId = Uuid.randomUuid();
members.put(TEST_MEMBER, new MemberSubscriptionAndAssignmentImpl(
Optional.empty(),
Optional.empty(),
- Set.of(topicId),
- new Assignment(Map.of(topicId, Set.of(1)))
+ Set.of(topicId, deletedTopicId),
+ new Assignment(Map.of(
+ // Partitions -1 and 3 do not exist.
+ topicId, Set.of(-1, 0, 2, 3),
+ // The deleted topic is not in the metadata image, so none of its partitions exist.
+ deletedTopicId, Set.of(0)
+ ))
));
- assertTrue(groupSpec.isPartitionAssigned(topicId, 1));
- assertFalse(groupSpec.isPartitionAssigned(topicId, 2));
- assertFalse(groupSpec.isPartitionAssigned(Uuid.randomUuid(), 2));
+ assertFalse(groupSpec.isPartitionAssigned(topicId, -1));
+ assertTrue(groupSpec.isPartitionAssigned(topicId, 0));
+ assertFalse(groupSpec.isPartitionAssigned(topicId, 1));
+ assertTrue(groupSpec.isPartitionAssigned(topicId, 2));
+ assertFalse(groupSpec.isPartitionAssigned(topicId, 3));
+ assertFalse(groupSpec.isPartitionAssigned(deletedTopicId, 0));
}
@Test