root
/
kafka
mirror of https://github.com/apache/kafka.git
1
0
Fork 0
Code Issues Actions 2 Packages Projects Releases Wiki Activity

KAFKA-14701; Move `PartitionAssignor` to new `group-coordinator-api` module (#16198) 

This patch moves the `PartitionAssignor` interface and all the related classes to a newly created `group-coordinator/api` module, following the pattern used by the storage and tools modules.

Reviewers: Ritika Reddy <rreddy@confluent.io>, Jeff Kim <jeff.kim@confluent.io>, Chia-Ping Tsai <chia7712@gmail.com>
This commit is contained in:
David Jacot 2024-06-06 21:19:20 +02:00 committed by GitHub
parent 79ea7d6122
commit 7d832cf74f
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
43 changed files with 561 additions and 405 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

63
build.gradle
View File

@ -1017,6 +1017,7 @@ project(':core') {
api libs.scalaLibrary
implementation project(':server-common')
implementation project(':group-coordinator:group-coordinator-api')
implementation project(':group-coordinator')
implementation project(':transaction-coordinator')
implementation project(':metadata')
@ -1400,6 +1401,66 @@ project(':metadata') {
}
}
project(':group-coordinator:group-coordinator-api') {
base {
archivesName = "kafka-group-coordinator-api"
}
dependencies {
implementation project(':clients')
}
task createVersionFile() {
def receiptFile = file("$buildDir/kafka/$buildVersionFileName")
inputs.property "commitId", commitId
inputs.property "version", version
outputs.file receiptFile
doLast {
def data = [
commitId: commitId,
version: version,
]
receiptFile.parentFile.mkdirs()
def content = data.entrySet().collect { "$it.key=$it.value" }.sort().join("\n")
receiptFile.setText(content, "ISO-8859-1")
}
}
sourceSets {
main {
java {
srcDirs = ["src/main/java"]
}
}
test {
java {
srcDirs = ["src/test/java"]
}
}
}
jar {
dependsOn createVersionFile
from("$buildDir") {
include "kafka/$buildVersionFileName"
}
}
clean.doFirst {
delete "$buildDir/kafka/"
}
javadoc {
include "**/org/apache/kafka/coordinator/group/api/**"
}
checkstyle {
configProperties = checkstyleConfigProperties("import-control-group-coordinator.xml")
}
}
project(':group-coordinator') {
base {
archivesName = "kafka-group-coordinator"
@ -1413,6 +1474,7 @@ project(':group-coordinator') {
implementation project(':server-common')
implementation project(':clients')
implementation project(':metadata')
implementation project(':group-coordinator:group-coordinator-api')
implementation project(':storage')
implementation libs.jacksonDatabind
implementation libs.jacksonJDK8Datatypes
@ -2957,6 +3019,7 @@ project(':jmh-benchmarks') {
implementation project(':raft')
implementation project(':clients')
implementation project(':group-coordinator')
implementation project(':group-coordinator:group-coordinator-api')
implementation project(':metadata')
implementation project(':storage')
implementation project(':streams')

2
core/src/main/scala/kafka/server/KafkaConfig.scala
View File

@ -37,7 +37,7 @@ import org.apache.kafka.common.utils.Utils
import org.apache.kafka.coordinator.group.ConsumerGroupMigrationPolicy
import org.apache.kafka.coordinator.group.Group.GroupType
import org.apache.kafka.coordinator.group.GroupCoordinatorConfig
import org.apache.kafka.coordinator.group.assignor.ConsumerGroupPartitionAssignor
import org.apache.kafka.coordinator.group.api.assignor.ConsumerGroupPartitionAssignor
import org.apache.kafka.coordinator.transaction.{TransactionLogConfigs, TransactionStateManagerConfigs}
import org.apache.kafka.network.SocketServerConfigs
import org.apache.kafka.raft.QuorumConfig

6
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/assignor/ConsumerGroupPartitionAssignor.java → group-coordinator/group-coordinator-api/src/main/java/org/apache/kafka/coordinator/group/api/assignor/ConsumerGroupPartitionAssignor.java
View File

@ -14,15 +14,15 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.apache.kafka.coordinator.group.assignor;
package org.apache.kafka.coordinator.group.api.assignor;
import org.apache.kafka.common.annotation.InterfaceStability;
/**
* Server-side partition assignor for consumer groups used by the GroupCoordinator.
*
* The interface is kept in an internal module until KIP-848 is fully
* implemented and ready to be released.
* The new consumer group protocol is in preview so this interface is considered
* unstable until Apache Kafka 4.0.
*/
@InterfaceStability.Unstable
public interface ConsumerGroupPartitionAssignor extends PartitionAssignor {

8
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/assignor/GroupAssignment.java → group-coordinator/group-coordinator-api/src/main/java/org/apache/kafka/coordinator/group/api/assignor/GroupAssignment.java
View File

@ -14,7 +14,9 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.apache.kafka.coordinator.group.assignor;
package org.apache.kafka.coordinator.group.api.assignor;
import org.apache.kafka.common.annotation.InterfaceStability;
import java.util.Map;
import java.util.Objects;
@ -22,6 +24,7 @@ import java.util.Objects;
/**
* The partition assignment for a consumer group.
*/
@InterfaceStability.Unstable
public class GroupAssignment {
/**
* The member assignments keyed by member id.
@ -31,8 +34,7 @@ public class GroupAssignment {
public GroupAssignment(
Map<String, MemberAssignment> members
) {
Objects.requireNonNull(members);
this.members = members;
this.members = Objects.requireNonNull(members);
}
/**

18
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/assignor/GroupSpec.java → group-coordinator/group-coordinator-api/src/main/java/org/apache/kafka/coordinator/group/api/assignor/GroupSpec.java
View File

@ -14,17 +14,17 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.apache.kafka.coordinator.group.assignor;
package org.apache.kafka.coordinator.group.api.assignor;
import org.apache.kafka.common.Uuid;
import org.apache.kafka.common.annotation.InterfaceStability;
import java.util.Collection;
import java.util.Map;
import java.util.Set;
/**
* The group metadata specifications required to compute the target assignment.
*/
@InterfaceStability.Unstable
public interface GroupSpec {
/**
* @return All the member Ids of the consumer group.
@ -45,18 +45,18 @@ public interface GroupSpec {
/**
* Gets the member subscription specification for a member.
*
* @param memberId The member Id.
* @param memberId The member Id.
* @return The member's subscription metadata.
* @throws IllegalArgumentException If the member Id isn't found.
*/
MemberSubscriptionSpec memberSubscription(String memberId);
MemberSubscription memberSubscription(String memberId);
/**
* Gets the current assignment of the member.
*
* @param memberId The member Id.
* @return A map of topic Ids to sets of partition numbers.
* An empty map is returned if the member Id isn't found.
* @param memberId The member Id.
* @return The member's assignment or an empty assignment if the
* member does not have one.
*/
Map<Uuid, Set<Integer>> memberAssignment(String memberId);
MemberAssignment memberAssignment(String memberId);
}

34
group-coordinator/group-coordinator-api/src/main/java/org/apache/kafka/coordinator/group/api/assignor/MemberAssignment.java Normal file
View File

@ -0,0 +1,34 @@
/*
* Licensed to the Apache Software Foundation (ASF) under one or more
* contributor license agreements. See the NOTICE file distributed with
* this work for additional information regarding copyright ownership.
* The ASF licenses this file to You under the Apache License, Version 2.0
* (the "License"); you may not use this file except in compliance with
* the License. You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.apache.kafka.coordinator.group.api.assignor;
import org.apache.kafka.common.Uuid;
import org.apache.kafka.common.annotation.InterfaceStability;
import java.util.Map;
import java.util.Set;
/**
* The partition assignment for a consumer group member.
*/
@InterfaceStability.Unstable
public interface MemberAssignment {
/**
* @return The assigned partitions keyed by topic Ids.
*/
Map<Uuid, Set<Integer>> partitions();
}

6
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/assignor/MemberSubscriptionSpec.java → group-coordinator/group-coordinator-api/src/main/java/org/apache/kafka/coordinator/group/api/assignor/MemberSubscription.java
View File

@ -14,9 +14,10 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.apache.kafka.coordinator.group.assignor;
package org.apache.kafka.coordinator.group.api.assignor;
import org.apache.kafka.common.Uuid;
import org.apache.kafka.common.annotation.InterfaceStability;
import java.util.Optional;
import java.util.Set;
@ -24,7 +25,8 @@ import java.util.Set;
/**
* Interface representing the subscription metadata for a group member.
*/
public interface MemberSubscriptionSpec {
@InterfaceStability.Unstable
public interface MemberSubscription {
/**
* Gets the rack Id if present.
*

6
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/assignor/PartitionAssignor.java → group-coordinator/group-coordinator-api/src/main/java/org/apache/kafka/coordinator/group/api/assignor/PartitionAssignor.java
View File

@ -14,15 +14,15 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.apache.kafka.coordinator.group.assignor;
package org.apache.kafka.coordinator.group.api.assignor;
import org.apache.kafka.common.annotation.InterfaceStability;
/**
* Server-side partition assignor used by the GroupCoordinator.
*
* The interface is kept in an internal module until KIP-848 is fully
* implemented and ready to be released.
* The new consumer group protocol is in preview so this interface is considered
* unstable until Apache Kafka 4.0.
*/
@InterfaceStability.Unstable
public interface PartitionAssignor {

2
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/assignor/PartitionAssignorException.java → group-coordinator/group-coordinator-api/src/main/java/org/apache/kafka/coordinator/group/api/assignor/PartitionAssignorException.java
View File

@ -14,7 +14,7 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.apache.kafka.coordinator.group.assignor;
package org.apache.kafka.coordinator.group.api.assignor;
import org.apache.kafka.common.errors.ApiException;

2
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/assignor/SubscribedTopicDescriber.java → group-coordinator/group-coordinator-api/src/main/java/org/apache/kafka/coordinator/group/api/assignor/SubscribedTopicDescriber.java
View File

@ -14,7 +14,7 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.apache.kafka.coordinator.group.assignor;
package org.apache.kafka.coordinator.group.api.assignor;
import org.apache.kafka.common.Uuid;
import org.apache.kafka.common.annotation.InterfaceStability;

5
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/assignor/SubscriptionType.java → group-coordinator/group-coordinator-api/src/main/java/org/apache/kafka/coordinator/group/api/assignor/SubscriptionType.java
View File

@ -14,11 +14,14 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.apache.kafka.coordinator.group.assignor;
package org.apache.kafka.coordinator.group.api.assignor;
import org.apache.kafka.common.annotation.InterfaceStability;
/**
* The subscription type followed by a consumer group.
*/
@InterfaceStability.Unstable
public enum SubscriptionType {
/**
* A homogeneous subscription type means that all the members

2
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupCoordinatorConfig.java
View File

@ -17,7 +17,7 @@
package org.apache.kafka.coordinator.group;
import org.apache.kafka.common.record.CompressionType;
import org.apache.kafka.coordinator.group.assignor.ConsumerGroupPartitionAssignor;
import org.apache.kafka.coordinator.group.api.assignor.ConsumerGroupPartitionAssignor;
import org.apache.kafka.coordinator.group.assignor.RangeAssignor;
import org.apache.kafka.coordinator.group.assignor.UniformAssignor;

10
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupMetadataManager.java
View File

@ -58,10 +58,10 @@ import org.apache.kafka.common.requests.JoinGroupRequest;
import org.apache.kafka.common.requests.RequestContext;
import org.apache.kafka.common.utils.LogContext;
import org.apache.kafka.common.utils.Time;
import org.apache.kafka.coordinator.group.assignor.ConsumerGroupPartitionAssignor;
import org.apache.kafka.coordinator.group.assignor.MemberAssignment;
import org.apache.kafka.coordinator.group.assignor.PartitionAssignorException;
import org.apache.kafka.coordinator.group.assignor.SubscriptionType;
import org.apache.kafka.coordinator.group.api.assignor.ConsumerGroupPartitionAssignor;
import org.apache.kafka.coordinator.group.api.assignor.MemberAssignment;
import org.apache.kafka.coordinator.group.api.assignor.PartitionAssignorException;
import org.apache.kafka.coordinator.group.api.assignor.SubscriptionType;
import org.apache.kafka.coordinator.group.consumer.Assignment;
import org.apache.kafka.coordinator.group.consumer.ConsumerGroup;
import org.apache.kafka.coordinator.group.consumer.ConsumerGroupMember;
@ -1926,7 +1926,7 @@ public class GroupMetadataManager {
MemberAssignment newMemberAssignment = assignmentResult.targetAssignment().get(updatedMember.memberId());
if (newMemberAssignment != null) {
return new Assignment(newMemberAssignment.targetPartitions());
return new Assignment(newMemberAssignment.partitions());
} else {
return Assignment.EMPTY;
}

5
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/assignor/AbstractUniformAssignmentBuilder.java
View File

@ -17,6 +17,9 @@
package org.apache.kafka.coordinator.group.assignor;
import org.apache.kafka.common.Uuid;
import org.apache.kafka.coordinator.group.api.assignor.GroupAssignment;
import org.apache.kafka.coordinator.group.api.assignor.MemberAssignment;
import org.apache.kafka.coordinator.group.api.assignor.SubscribedTopicDescriber;
import org.apache.kafka.server.common.TopicIdPartition;
import java.util.Collection;
@ -42,7 +45,7 @@ public abstract class AbstractUniformAssignmentBuilder {
int partition
) {
memberAssignments.get(memberId)
.targetPartitions()
.partitions()
.computeIfAbsent(topicId, __ -> new HashSet<>())
.add(partition);
}

16
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/assignor/GeneralUniformAssignmentBuilder.java
View File

@ -17,6 +17,12 @@
package org.apache.kafka.coordinator.group.assignor;
import org.apache.kafka.common.Uuid;
import org.apache.kafka.coordinator.group.api.assignor.GroupAssignment;
import org.apache.kafka.coordinator.group.api.assignor.GroupSpec;
import org.apache.kafka.coordinator.group.api.assignor.MemberAssignment;
import org.apache.kafka.coordinator.group.api.assignor.PartitionAssignorException;
import org.apache.kafka.coordinator.group.api.assignor.SubscribedTopicDescriber;
import org.apache.kafka.coordinator.group.consumer.MemberAssignmentImpl;
import org.apache.kafka.server.common.TopicIdPartition;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@ -124,7 +130,7 @@ public class GeneralUniformAssignmentBuilder extends AbstractUniformAssignmentBu
}
subscribedTopicIds.add(topicId);
membersPerTopic.computeIfAbsent(topicId, k -> new ArrayList<>()).add(memberId);
targetAssignment.put(memberId, new MemberAssignment(new HashMap<>()));
targetAssignment.put(memberId, new MemberAssignmentImpl(new HashMap<>()));
})
);
this.unassignedPartitions = new HashSet<>(topicIdPartitions(subscribedTopicIds, subscribedTopicDescriber));
@ -191,7 +197,7 @@ public class GeneralUniformAssignmentBuilder extends AbstractUniformAssignmentBu
*/
private void assignStickyPartitions() {
groupSpec.memberIds().forEach(memberId ->
groupSpec.memberAssignment(memberId).forEach((topicId, currentAssignment) -> {
groupSpec.memberAssignment(memberId).partitions().forEach((topicId, currentAssignment) -> {
if (groupSpec.memberSubscription(memberId).subscribedTopicIds().contains(topicId)) {
currentAssignment.forEach(partition -> {
TopicIdPartition topicIdPartition = new TopicIdPartition(topicId, partition);
@ -244,7 +250,7 @@ public class GeneralUniformAssignmentBuilder extends AbstractUniformAssignmentBu
* @return true if the member can participate in reassignment, false otherwise.
*/
private boolean canMemberParticipateInReassignment(String memberId) {
Set<Uuid> assignedTopicIds = targetAssignment.get(memberId).targetPartitions().keySet();
Set<Uuid> assignedTopicIds = targetAssignment.get(memberId).partitions().keySet();
int currentAssignmentSize = assignmentManager.targetAssignmentSize(memberId);
int maxAssignmentSize = assignmentManager.maxAssignmentSize(memberId);
@ -292,7 +298,7 @@ public class GeneralUniformAssignmentBuilder extends AbstractUniformAssignmentBu
// Otherwise make sure it cannot get any more partitions.
for (Uuid topicId : groupSpec.memberSubscription(member).subscribedTopicIds()) {
Set<Integer> assignedPartitions = targetAssignment.get(member).targetPartitions().get(topicId);
Set<Integer> assignedPartitions = targetAssignment.get(member).partitions().get(topicId);
for (int i = 0; i < subscribedTopicDescriber.numPartitions(topicId); i++) {
TopicIdPartition topicIdPartition = new TopicIdPartition(topicId, i);
if (assignedPartitions == null || !assignedPartitions.contains(i)) {
@ -784,7 +790,7 @@ public class GeneralUniformAssignmentBuilder extends AbstractUniformAssignmentBu
* @param memberId Member that the partition needs to be revoked from.
*/
private void removePartitionFromTargetAssignment(TopicIdPartition topicIdPartition, String memberId) {
Map<Uuid, Set<Integer>> targetPartitionsMap = targetAssignment.get(memberId).targetPartitions();
Map<Uuid, Set<Integer>> targetPartitionsMap = targetAssignment.get(memberId).partitions();
Set<Integer> partitionsSet = targetPartitionsMap.get(topicIdPartition.topicId());
// Remove the partition from the assignment, if there are no more partitions from a particular topic,
// remove the topic from the assignment as well.

16
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/assignor/OptimizedUniformAssignmentBuilder.java
View File

@ -17,6 +17,12 @@
package org.apache.kafka.coordinator.group.assignor;
import org.apache.kafka.common.Uuid;
import org.apache.kafka.coordinator.group.api.assignor.GroupAssignment;
import org.apache.kafka.coordinator.group.api.assignor.GroupSpec;
import org.apache.kafka.coordinator.group.api.assignor.MemberAssignment;
import org.apache.kafka.coordinator.group.api.assignor.PartitionAssignorException;
import org.apache.kafka.coordinator.group.api.assignor.SubscribedTopicDescriber;
import org.apache.kafka.coordinator.group.consumer.MemberAssignmentImpl;
import org.apache.kafka.server.common.TopicIdPartition;
import java.util.ArrayList;
@ -160,7 +166,7 @@ public class OptimizedUniformAssignmentBuilder {
*/
private void maybeRevokePartitions() {
for (String memberId : groupSpec.memberIds()) {
Map<Uuid, Set<Integer>> oldAssignment = groupSpec.memberAssignment(memberId);
Map<Uuid, Set<Integer>> oldAssignment = groupSpec.memberAssignment(memberId).partitions();
Map<Uuid, Set<Integer>> newAssignment = null;
// The assignor expects to receive the assignment as an immutable map. It leverages
@ -219,9 +225,9 @@ public class OptimizedUniformAssignmentBuilder {
}
if (newAssignment == null) {
targetAssignment.put(memberId, new MemberAssignment(oldAssignment));
targetAssignment.put(memberId, new MemberAssignmentImpl(oldAssignment));
} else {
targetAssignment.put(memberId, new MemberAssignment(newAssignment));
targetAssignment.put(memberId, new MemberAssignmentImpl(newAssignment));
}
}
}
@ -236,12 +242,12 @@ public class OptimizedUniformAssignmentBuilder {
String memberId = unfilledMember.memberId;
int remainingQuota = unfilledMember.remainingQuota;
Map<Uuid, Set<Integer>> newAssignment = targetAssignment.get(memberId).targetPartitions();
Map<Uuid, Set<Integer>> newAssignment = targetAssignment.get(memberId).partitions();
if (isImmutableMap(newAssignment)) {
// If the new assignment is immutable, we must create a deep copy of it
// before altering it.
newAssignment = deepCopy(newAssignment);
targetAssignment.put(memberId, new MemberAssignment(newAssignment));
targetAssignment.put(memberId, new MemberAssignmentImpl(newAssignment));
}
for (int i = 0; i < remainingQuota && unassignedPartitionIndex < unassignedPartitions.size(); i++) {

25
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/assignor/RangeAssignor.java
View File

@ -17,6 +17,13 @@
package org.apache.kafka.coordinator.group.assignor;
import org.apache.kafka.common.Uuid;
import org.apache.kafka.coordinator.group.api.assignor.ConsumerGroupPartitionAssignor;
import org.apache.kafka.coordinator.group.api.assignor.GroupAssignment;
import org.apache.kafka.coordinator.group.api.assignor.GroupSpec;
import org.apache.kafka.coordinator.group.api.assignor.MemberAssignment;
import org.apache.kafka.coordinator.group.api.assignor.PartitionAssignorException;
import org.apache.kafka.coordinator.group.api.assignor.SubscribedTopicDescriber;
import org.apache.kafka.coordinator.group.consumer.MemberAssignmentImpl;
import java.util.ArrayList;
import java.util.Collection;
@ -28,7 +35,7 @@ import java.util.Map;
import java.util.Set;
import static java.lang.Math.min;
import static org.apache.kafka.coordinator.group.assignor.SubscriptionType.HOMOGENEOUS;
import static org.apache.kafka.coordinator.group.api.assignor.SubscriptionType.HOMOGENEOUS;
/**
* This Range Assignor inherits properties of both the range assignor and the sticky assignor.
@ -162,7 +169,9 @@ public class RangeAssignor implements ConsumerGroupPartitionAssignor {
List<MemberWithRemainingAssignments> potentiallyUnfilledMembers = new ArrayList<>();
for (String memberId : membersForTopic) {
Set<Integer> assignedPartitionsForTopic = groupSpec.memberAssignment(memberId)
Set<Integer> assignedPartitionsForTopic = groupSpec
.memberAssignment(memberId)
.partitions()
.getOrDefault(topicId, Collections.emptySet());
int currentAssignmentSize = assignedPartitionsForTopic.size();
@ -177,8 +186,8 @@ public class RangeAssignor implements ConsumerGroupPartitionAssignor {
for (int i = 0; i < retainedPartitionsCount; i++) {
assignedStickyPartitionsForTopic
.add(currentAssignmentListForTopic.get(i));
newTargetAssignment.computeIfAbsent(memberId, k -> new MemberAssignment(new HashMap<>()))
.targetPartitions()
newTargetAssignment.computeIfAbsent(memberId, k -> new MemberAssignmentImpl(new HashMap<>()))
.partitions()
.computeIfAbsent(topicId, k -> new HashSet<>())
.add(currentAssignmentListForTopic.get(i));
}
@ -198,8 +207,8 @@ public class RangeAssignor implements ConsumerGroupPartitionAssignor {
// add the extra partition that will be present at the index right after min quota was satisfied.
assignedStickyPartitionsForTopic
.add(currentAssignmentListForTopic.get(minRequiredQuota));
newTargetAssignment.computeIfAbsent(memberId, k -> new MemberAssignment(new HashMap<>()))
.targetPartitions()
newTargetAssignment.computeIfAbsent(memberId, k -> new MemberAssignmentImpl(new HashMap<>()))
.partitions()
.computeIfAbsent(topicId, k -> new HashSet<>())
.add(currentAssignmentListForTopic.get(minRequiredQuota));
} else {
@ -233,8 +242,8 @@ public class RangeAssignor implements ConsumerGroupPartitionAssignor {
List<Integer> partitionsToAssign = unassignedPartitionsForTopic
.subList(unassignedPartitionsListStartPointer, unassignedPartitionsListStartPointer + remaining);
unassignedPartitionsListStartPointer += remaining;
newTargetAssignment.computeIfAbsent(memberId, k -> new MemberAssignment(new HashMap<>()))
.targetPartitions()
newTargetAssignment.computeIfAbsent(memberId, k -> new MemberAssignmentImpl(new HashMap<>()))
.partitions()
.computeIfAbsent(topicId, k -> new HashSet<>())
.addAll(partitionsToAssign);
}

7
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/assignor/UniformAssignor.java
View File

@ -16,12 +16,17 @@
*/
package org.apache.kafka.coordinator.group.assignor;
import org.apache.kafka.coordinator.group.api.assignor.ConsumerGroupPartitionAssignor;
import org.apache.kafka.coordinator.group.api.assignor.GroupAssignment;
import org.apache.kafka.coordinator.group.api.assignor.GroupSpec;
import org.apache.kafka.coordinator.group.api.assignor.PartitionAssignorException;
import org.apache.kafka.coordinator.group.api.assignor.SubscribedTopicDescriber;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.util.Collections;
import static org.apache.kafka.coordinator.group.assignor.SubscriptionType.HOMOGENEOUS;
import static org.apache.kafka.coordinator.group.api.assignor.SubscriptionType.HOMOGENEOUS;
/**
* The Uniform Assignor distributes topic partitions among group members for a

4
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/consumer/Assignment.java
View File

@ -17,6 +17,7 @@
package org.apache.kafka.coordinator.group.consumer;
import org.apache.kafka.common.Uuid;
import org.apache.kafka.coordinator.group.api.assignor.MemberAssignment;
import org.apache.kafka.coordinator.group.generated.ConsumerGroupTargetAssignmentMemberValue;
import java.util.Collections;
@ -29,7 +30,7 @@ import java.util.stream.Collectors;
/**
* An immutable assignment for a member.
*/
public class Assignment {
public class Assignment implements MemberAssignment {
public static final Assignment EMPTY = new Assignment(Collections.emptyMap());
/**
@ -46,6 +47,7 @@ public class Assignment {
/**
* @return The assigned partitions.
*/
@Override
public Map<Uuid, Set<Integer>> partitions() {
return partitions;
}

6
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/consumer/ConsumerGroup.java
View File

@ -33,7 +33,7 @@ import org.apache.kafka.coordinator.group.OffsetExpirationCondition;
import org.apache.kafka.coordinator.group.OffsetExpirationConditionImpl;
import org.apache.kafka.coordinator.group.CoordinatorRecord;
import org.apache.kafka.coordinator.group.CoordinatorRecordHelpers;
import org.apache.kafka.coordinator.group.assignor.SubscriptionType;
import org.apache.kafka.coordinator.group.api.assignor.SubscriptionType;
import org.apache.kafka.coordinator.group.classic.ClassicGroup;
import org.apache.kafka.coordinator.group.generated.ConsumerGroupMemberMetadataValue;
import org.apache.kafka.coordinator.group.metrics.GroupCoordinatorMetricsShard;
@ -56,8 +56,8 @@ import java.util.Objects;
import java.util.Optional;
import java.util.Set;
import static org.apache.kafka.coordinator.group.assignor.SubscriptionType.HETEROGENEOUS;
import static org.apache.kafka.coordinator.group.assignor.SubscriptionType.HOMOGENEOUS;
import static org.apache.kafka.coordinator.group.api.assignor.SubscriptionType.HETEROGENEOUS;
import static org.apache.kafka.coordinator.group.api.assignor.SubscriptionType.HOMOGENEOUS;
import static org.apache.kafka.coordinator.group.consumer.ConsumerGroup.ConsumerGroupState.ASSIGNING;
import static org.apache.kafka.coordinator.group.consumer.ConsumerGroup.ConsumerGroupState.EMPTY;
import static org.apache.kafka.coordinator.group.consumer.ConsumerGroup.ConsumerGroupState.RECONCILING;

35
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/assignor/GroupSpecImpl.java → group-coordinator/src/main/java/org/apache/kafka/coordinator/group/consumer/GroupSpecImpl.java
View File

@ -14,15 +14,18 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.apache.kafka.coordinator.group.assignor;
package org.apache.kafka.coordinator.group.consumer;
import org.apache.kafka.common.Uuid;
import org.apache.kafka.coordinator.group.api.assignor.GroupSpec;
import org.apache.kafka.coordinator.group.api.assignor.MemberAssignment;
import org.apache.kafka.coordinator.group.api.assignor.MemberSubscription;
import org.apache.kafka.coordinator.group.api.assignor.SubscriptionType;
import java.util.Collection;
import java.util.Collections;
import java.util.Map;
import java.util.Objects;
import java.util.Set;
/**
* The assignment specification for a consumer group.
@ -31,7 +34,7 @@ public class GroupSpecImpl implements GroupSpec {
/**
* Member subscription metadata keyed by member Id.
*/
private final Map<String, MemberSubscriptionSpecImpl> memberSubscriptions;
private final Map<String, MemberSubscriptionAndAssignmentImpl> members;
/**
* The subscription type of the group.
@ -45,11 +48,11 @@ public class GroupSpecImpl implements GroupSpec {
private final Map<Uuid, Map<Integer, String>> invertedMemberAssignment;
public GroupSpecImpl(
Map<String, MemberSubscriptionSpecImpl> memberSubscriptions,
Map<String, MemberSubscriptionAndAssignmentImpl> members,
SubscriptionType subscriptionType,
Map<Uuid, Map<Integer, String>> invertedMemberAssignment
) {
this.memberSubscriptions = Objects.requireNonNull(memberSubscriptions);
this.members = Objects.requireNonNull(members);
this.subscriptionType = Objects.requireNonNull(subscriptionType);
this.invertedMemberAssignment = Objects.requireNonNull(invertedMemberAssignment);
}
@ -59,7 +62,7 @@ public class GroupSpecImpl implements GroupSpec {
*/
@Override
public Collection<String> memberIds() {
return memberSubscriptions.keySet();
return members.keySet();
}
/**
@ -86,8 +89,8 @@ public class GroupSpecImpl implements GroupSpec {
* {@inheritDoc}
*/
@Override
public MemberSubscriptionSpec memberSubscription(String memberId) {
MemberSubscriptionSpec memberSubscription = memberSubscriptions.get(memberId);
public MemberSubscription memberSubscription(String memberId) {
MemberSubscription memberSubscription = members.get(memberId);
if (memberSubscription == null) {
throw new IllegalArgumentException("Member Id " + memberId + " not found.");
}
@ -98,12 +101,12 @@ public class GroupSpecImpl implements GroupSpec {
* {@inheritDoc}
*/
@Override
public Map<Uuid, Set<Integer>> memberAssignment(String memberId) {
MemberSubscriptionSpecImpl memberSubscription = memberSubscriptions.get(memberId);
if (memberSubscription == null) {
return Collections.emptyMap();
public MemberAssignment memberAssignment(String memberId) {
MemberSubscriptionAndAssignmentImpl member = members.get(memberId);
if (member == null) {
return new MemberAssignmentImpl(Collections.emptyMap());
}
return memberSubscription.memberAssignment();
return member;
}
@Override
@ -112,13 +115,13 @@ public class GroupSpecImpl implements GroupSpec {
if (o == null || getClass() != o.getClass()) return false;
GroupSpecImpl that = (GroupSpecImpl) o;
return subscriptionType == that.subscriptionType &&
memberSubscriptions.equals(that.memberSubscriptions) &&
members.equals(that.members) &&
invertedMemberAssignment.equals(that.invertedMemberAssignment);
}
@Override
public int hashCode() {
int result = memberSubscriptions.hashCode();
int result = members.hashCode();
result = 31 * result + subscriptionType.hashCode();
result = 31 * result + invertedMemberAssignment.hashCode();
return result;
@ -126,7 +129,7 @@ public class GroupSpecImpl implements GroupSpec {
@Override
public String toString() {
return "GroupSpecImpl(memberSubscriptions=" + memberSubscriptions +
return "GroupSpecImpl(members=" + members +
", subscriptionType=" + subscriptionType +
", invertedMemberAssignment=" + invertedMemberAssignment +
')';

29
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/assignor/MemberAssignment.java → group-coordinator/src/main/java/org/apache/kafka/coordinator/group/consumer/MemberAssignmentImpl.java
View File

@ -14,9 +14,10 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.apache.kafka.coordinator.group.assignor;
package org.apache.kafka.coordinator.group.consumer;
import org.apache.kafka.common.Uuid;
import org.apache.kafka.coordinator.group.api.assignor.MemberAssignment;
import java.util.Map;
import java.util.Objects;
@ -25,39 +26,39 @@ import java.util.Set;
/**
* The partition assignment for a consumer group member.
*/
public class MemberAssignment {
public class MemberAssignmentImpl implements MemberAssignment {
/**
* The target partitions assigned to this member keyed by topicId.
* The partitions assigned to this member keyed by topicId.
*/
private final Map<Uuid, Set<Integer>> targetPartitions;
private final Map<Uuid, Set<Integer>> partitions;
public MemberAssignment(Map<Uuid, Set<Integer>> targetPartitions) {
Objects.requireNonNull(targetPartitions);
this.targetPartitions = targetPartitions;
public MemberAssignmentImpl(Map<Uuid, Set<Integer>> partitions) {
this.partitions = Objects.requireNonNull(partitions);
}
/**
* @return Target partitions keyed by topic Ids.
* @return The assigned partitions keyed by topic Ids.
*/
public Map<Uuid, Set<Integer>> targetPartitions() {
return this.targetPartitions;
@Override
public Map<Uuid, Set<Integer>> partitions() {
return this.partitions;
}
@Override
public boolean equals(Object o) {
if (this == o) return true;
if (o == null || getClass() != o.getClass()) return false;
MemberAssignment that = (MemberAssignment) o;
return targetPartitions.equals(that.targetPartitions);
MemberAssignmentImpl that = (MemberAssignmentImpl) o;
return partitions.equals(that.partitions);
}
@Override
public int hashCode() {
return targetPartitions.hashCode();
return partitions.hashCode();
}
@Override
public String toString() {
return "MemberAssignment(targetPartitions=" + targetPartitions + ')';
return "MemberAssignment(partitions=" + partitions + ')';
}
}

20
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/assignor/MemberSubscriptionSpecImpl.java → group-coordinator/src/main/java/org/apache/kafka/coordinator/group/consumer/MemberSubscriptionAndAssignmentImpl.java
View File

@ -14,10 +14,11 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.apache.kafka.coordinator.group.assignor;
package org.apache.kafka.coordinator.group.consumer;
import org.apache.kafka.common.Uuid;
import org.apache.kafka.coordinator.group.consumer.Assignment;
import org.apache.kafka.coordinator.group.api.assignor.MemberAssignment;
import org.apache.kafka.coordinator.group.api.assignor.MemberSubscription;
import java.util.Map;
import java.util.Objects;
@ -25,21 +26,21 @@ import java.util.Optional;
import java.util.Set;
/**
* Implementation of the {@link MemberSubscriptionSpec} interface.
* Implementation of the {@link MemberSubscription} and the {@link MemberAssignment} interfaces.
*/
public class MemberSubscriptionSpecImpl implements MemberSubscriptionSpec {
public class MemberSubscriptionAndAssignmentImpl implements MemberSubscription, MemberAssignment {
private final Optional<String> rackId;
private final Set<Uuid> subscribedTopicIds;
private final Assignment memberAssignment;
/**
* Constructs a new {@code MemberSubscriptionSpecImpl}.
* Constructs a new {@code MemberSubscriptionAndAssignmentImpl}.
*
* @param rackId The rack Id.
* @param subscribedTopicIds The set of subscribed topic Ids.
* @param memberAssignment The current member assignment.
*/
public MemberSubscriptionSpecImpl(
public MemberSubscriptionAndAssignmentImpl(
Optional<String> rackId,
Set<Uuid> subscribedTopicIds,
Assignment memberAssignment
@ -59,7 +60,8 @@ public class MemberSubscriptionSpecImpl implements MemberSubscriptionSpec {
return subscribedTopicIds;
}
public Map<Uuid, Set<Integer>> memberAssignment() {
@Override
public Map<Uuid, Set<Integer>> partitions() {
return memberAssignment.partitions();
}
@ -67,7 +69,7 @@ public class MemberSubscriptionSpecImpl implements MemberSubscriptionSpec {
public boolean equals(Object o) {
if (this == o) return true;
if (o == null || getClass() != o.getClass()) return false;
MemberSubscriptionSpecImpl that = (MemberSubscriptionSpecImpl) o;
MemberSubscriptionAndAssignmentImpl that = (MemberSubscriptionAndAssignmentImpl) o;
return rackId.equals(that.rackId) &&
subscribedTopicIds.equals(that.subscribedTopicIds) &&
memberAssignment.equals(that.memberAssignment);
@ -83,7 +85,7 @@ public class MemberSubscriptionSpecImpl implements MemberSubscriptionSpec {
@Override
public String toString() {
return "MemberSubscriptionSpecImpl(rackId=" + rackId.orElse("N/A") +
return "MemberSubscriptionAndAssignmentImpl(rackId=" + rackId.orElse("N/A") +
", subscribedTopicIds=" + subscribedTopicIds +
", memberAssignment=" + memberAssignment +
')';

10
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/consumer/SubscribedTopicMetadata.java → group-coordinator/src/main/java/org/apache/kafka/coordinator/group/consumer/SubscribedTopicDescriberImpl.java
View File

@ -17,8 +17,8 @@
package org.apache.kafka.coordinator.group.consumer;
import org.apache.kafka.common.Uuid;
import org.apache.kafka.coordinator.group.assignor.PartitionAssignor;
import org.apache.kafka.coordinator.group.assignor.SubscribedTopicDescriber;
import org.apache.kafka.coordinator.group.api.assignor.PartitionAssignor;
import org.apache.kafka.coordinator.group.api.assignor.SubscribedTopicDescriber;
import java.util.Collections;
import java.util.Map;
@ -29,14 +29,14 @@ import java.util.Set;
* The subscribed topic metadata class is used by the {@link PartitionAssignor} to obtain
* topic and partition metadata for the topics that the consumer group is subscribed to.
*/
public class SubscribedTopicMetadata implements SubscribedTopicDescriber {
public class SubscribedTopicDescriberImpl implements SubscribedTopicDescriber {
/**
* The topic Ids mapped to their corresponding {@link TopicMetadata}
* object, which contains topic and partition metadata.
*/
private final Map<Uuid, TopicMetadata> topicMetadata;
public SubscribedTopicMetadata(Map<Uuid, TopicMetadata> topicMetadata) {
public SubscribedTopicDescriberImpl(Map<Uuid, TopicMetadata> topicMetadata) {
this.topicMetadata = Objects.requireNonNull(topicMetadata);
}
@ -80,7 +80,7 @@ public class SubscribedTopicMetadata implements SubscribedTopicDescriber {
public boolean equals(Object o) {
if (this == o) return true;
if (o == null || getClass() != o.getClass()) return false;
SubscribedTopicMetadata that = (SubscribedTopicMetadata) o;
SubscribedTopicDescriberImpl that = (SubscribedTopicDescriberImpl) o;
return topicMetadata.equals(that.topicMetadata);
}

26
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/consumer/TargetAssignmentBuilder.java
View File

@ -18,13 +18,11 @@ package org.apache.kafka.coordinator.group.consumer;
import org.apache.kafka.common.Uuid;
import org.apache.kafka.coordinator.group.CoordinatorRecord;
import org.apache.kafka.coordinator.group.assignor.GroupSpecImpl;
import org.apache.kafka.coordinator.group.assignor.MemberSubscriptionSpecImpl;
import org.apache.kafka.coordinator.group.assignor.SubscriptionType;
import org.apache.kafka.coordinator.group.assignor.GroupAssignment;
import org.apache.kafka.coordinator.group.assignor.MemberAssignment;
import org.apache.kafka.coordinator.group.assignor.PartitionAssignor;
import org.apache.kafka.coordinator.group.assignor.PartitionAssignorException;
import org.apache.kafka.coordinator.group.api.assignor.SubscriptionType;
import org.apache.kafka.coordinator.group.api.assignor.GroupAssignment;
import org.apache.kafka.coordinator.group.api.assignor.MemberAssignment;
import org.apache.kafka.coordinator.group.api.assignor.PartitionAssignor;
import org.apache.kafka.coordinator.group.api.assignor.PartitionAssignorException;
import org.apache.kafka.image.TopicsImage;
import java.util.ArrayList;
@ -293,11 +291,11 @@ public class TargetAssignmentBuilder {
* @throws PartitionAssignorException if the target assignment cannot be computed.
*/
public TargetAssignmentResult build() throws PartitionAssignorException {
Map<String, MemberSubscriptionSpecImpl> memberSpecs = new HashMap<>();
Map<String, MemberSubscriptionAndAssignmentImpl> memberSpecs = new HashMap<>();
// Prepare the member spec for all members.
members.forEach((memberId, member) ->
memberSpecs.put(memberId, createMemberSubscriptionSpecImpl(
memberSpecs.put(memberId, createMemberSubscriptionAndAssignment(
member,
targetAssignment.getOrDefault(memberId, Assignment.EMPTY),
topicsImage
@ -319,7 +317,7 @@ public class TargetAssignmentBuilder {
}
}
memberSpecs.put(memberId, createMemberSubscriptionSpecImpl(
memberSpecs.put(memberId, createMemberSubscriptionAndAssignment(
updatedMemberOrNull,
assignment,
topicsImage
@ -343,7 +341,7 @@ public class TargetAssignmentBuilder {
subscriptionType,
invertedTargetAssignment
),
new SubscribedTopicMetadata(topicMetadataMap)
new SubscribedTopicDescriberImpl(topicMetadataMap)
);
// Compute delta from previous to new target assignment and create the
@ -377,19 +375,19 @@ public class TargetAssignmentBuilder {
) {
MemberAssignment newMemberAssignment = newGroupAssignment.members().get(memberId);
if (newMemberAssignment != null) {
return new Assignment(newMemberAssignment.targetPartitions());
return new Assignment(newMemberAssignment.partitions());
} else {
return Assignment.EMPTY;
}
}
// private for testing
static MemberSubscriptionSpecImpl createMemberSubscriptionSpecImpl(
static MemberSubscriptionAndAssignmentImpl createMemberSubscriptionAndAssignment(
ConsumerGroupMember member,
Assignment memberAssignment,
TopicsImage topicsImage
) {
return new MemberSubscriptionSpecImpl(
return new MemberSubscriptionAndAssignmentImpl(
Optional.ofNullable(member.rackId()),
new TopicIds(member.subscribedTopicNames(), topicsImage),
memberAssignment

12
group-coordinator/src/test/java/org/apache/kafka/coordinator/group/AssignmentTestUtil.java
View File

@ -17,8 +17,8 @@
package org.apache.kafka.coordinator.group;
import org.apache.kafka.common.Uuid;
import org.apache.kafka.coordinator.group.assignor.GroupAssignment;
import org.apache.kafka.coordinator.group.assignor.MemberSubscriptionSpecImpl;
import org.apache.kafka.coordinator.group.api.assignor.GroupAssignment;
import org.apache.kafka.coordinator.group.consumer.MemberSubscriptionAndAssignmentImpl;
import java.util.AbstractMap;
import java.util.Arrays;
@ -80,7 +80,7 @@ public class AssignmentTestUtil {
) {
assertEquals(expectedAssignment.size(), computedGroupAssignment.members().size());
computedGroupAssignment.members().forEach((memberId, memberAssignment) -> {
Map<Uuid, Set<Integer>> computedAssignmentForMember = memberAssignment.targetPartitions();
Map<Uuid, Set<Integer>> computedAssignmentForMember = memberAssignment.partitions();
assertEquals(expectedAssignment.get(memberId), computedAssignmentForMember);
});
}
@ -92,12 +92,12 @@ public class AssignmentTestUtil {
* @return Map of topic partition to member assignments.
*/
public static Map<Uuid, Map<Integer, String>> invertedTargetAssignment(
Map<String, MemberSubscriptionSpecImpl> members
Map<String, MemberSubscriptionAndAssignmentImpl> members
) {
Map<Uuid, Map<Integer, String>> invertedTargetAssignment = new HashMap<>();
for (Map.Entry<String, MemberSubscriptionSpecImpl> memberEntry : members.entrySet()) {
for (Map.Entry<String, MemberSubscriptionAndAssignmentImpl> memberEntry : members.entrySet()) {
String memberId = memberEntry.getKey();
Map<Uuid, Set<Integer>> memberAssignment = memberEntry.getValue().memberAssignment();
Map<Uuid, Set<Integer>> memberAssignment = memberEntry.getValue().partitions();
for (Map.Entry<Uuid, Set<Integer>> topicEntry : memberAssignment.entrySet()) {
Uuid topicId = topicEntry.getKey();

2
group-coordinator/src/test/java/org/apache/kafka/coordinator/group/GroupCoordinatorConfigTest.java
View File

@ -17,7 +17,7 @@
package org.apache.kafka.coordinator.group;
import org.apache.kafka.common.record.CompressionType;
import org.apache.kafka.coordinator.group.assignor.ConsumerGroupPartitionAssignor;
import org.apache.kafka.coordinator.group.api.assignor.ConsumerGroupPartitionAssignor;
import org.apache.kafka.coordinator.group.assignor.RangeAssignor;
import org.junit.jupiter.api.Test;

101
group-coordinator/src/test/java/org/apache/kafka/coordinator/group/GroupMetadataManagerTest.java
View File

@ -61,10 +61,11 @@ import org.apache.kafka.common.utils.LogContext;
import org.apache.kafka.common.utils.Utils;
import org.apache.kafka.coordinator.group.MockCoordinatorTimer.ExpiredTimeout;
import org.apache.kafka.coordinator.group.MockCoordinatorTimer.ScheduledTimeout;
import org.apache.kafka.coordinator.group.assignor.ConsumerGroupPartitionAssignor;
import org.apache.kafka.coordinator.group.assignor.GroupAssignment;
import org.apache.kafka.coordinator.group.assignor.MemberAssignment;
import org.apache.kafka.coordinator.group.assignor.PartitionAssignorException;
import org.apache.kafka.coordinator.group.api.assignor.ConsumerGroupPartitionAssignor;
import org.apache.kafka.coordinator.group.api.assignor.GroupAssignment;
import org.apache.kafka.coordinator.group.api.assignor.MemberAssignment;
import org.apache.kafka.coordinator.group.consumer.MemberAssignmentImpl;
import org.apache.kafka.coordinator.group.api.assignor.PartitionAssignorException;
import org.apache.kafka.coordinator.group.classic.ClassicGroupState;
import org.apache.kafka.coordinator.group.consumer.Assignment;
import org.apache.kafka.coordinator.group.consumer.ConsumerGroup;
@ -435,7 +436,7 @@ public class GroupMetadataManagerTest {
.build();
assignor.prepareGroupAssignment(new GroupAssignment(
Collections.singletonMap(memberId, new MemberAssignment(mkAssignment(
Collections.singletonMap(memberId, new MemberAssignmentImpl(mkAssignment(
mkTopicAssignment(fooTopicId, 0, 1, 2, 3, 4, 5),
mkTopicAssignment(barTopicId, 0, 1, 2)
)))
@ -540,7 +541,7 @@ public class GroupMetadataManagerTest {
.build();
assignor.prepareGroupAssignment(new GroupAssignment(
Collections.singletonMap(memberId, new MemberAssignment(mkAssignment(
Collections.singletonMap(memberId, new MemberAssignmentImpl(mkAssignment(
mkTopicAssignment(fooTopicId, 0, 1, 2, 3, 4, 5),
mkTopicAssignment(barTopicId, 0, 1, 2)
)))
@ -663,15 +664,15 @@ public class GroupMetadataManagerTest {
assignor.prepareGroupAssignment(new GroupAssignment(
new HashMap<String, MemberAssignment>() {
{
put(memberId1, new MemberAssignment(mkAssignment(
put(memberId1, new MemberAssignmentImpl(mkAssignment(
mkTopicAssignment(fooTopicId, 0, 1),
mkTopicAssignment(barTopicId, 0)
)));
put(memberId2, new MemberAssignment(mkAssignment(
put(memberId2, new MemberAssignmentImpl(mkAssignment(
mkTopicAssignment(fooTopicId, 2, 3),
mkTopicAssignment(barTopicId, 1)
)));
put(memberId3, new MemberAssignment(mkAssignment(
put(memberId3, new MemberAssignmentImpl(mkAssignment(
mkTopicAssignment(fooTopicId, 4, 5),
mkTopicAssignment(barTopicId, 2)
)));
@ -897,15 +898,15 @@ public class GroupMetadataManagerTest {
assignor.prepareGroupAssignment(new GroupAssignment(
new HashMap<String, MemberAssignment>() {
{
put(memberId1, new MemberAssignment(mkAssignment(
put(memberId1, new MemberAssignmentImpl(mkAssignment(
mkTopicAssignment(fooTopicId, 0, 1),
mkTopicAssignment(barTopicId, 0)
)));
put(memberId2, new MemberAssignment(mkAssignment(
put(memberId2, new MemberAssignmentImpl(mkAssignment(
mkTopicAssignment(fooTopicId, 2, 3),
mkTopicAssignment(barTopicId, 1)
)));
put(memberId3, new MemberAssignment(mkAssignment(
put(memberId3, new MemberAssignmentImpl(mkAssignment(
mkTopicAssignment(fooTopicId, 4, 5),
mkTopicAssignment(barTopicId, 2)
)));
@ -1047,12 +1048,12 @@ public class GroupMetadataManagerTest {
assignor.prepareGroupAssignment(new GroupAssignment(
new HashMap<String, MemberAssignment>() {
{
put(memberId1, new MemberAssignment(mkAssignment(
put(memberId1, new MemberAssignmentImpl(mkAssignment(
mkTopicAssignment(fooTopicId, 0, 1, 2),
mkTopicAssignment(barTopicId, 0, 1)
)));
// When the member rejoins, it gets the same assignments.
put(member2RejoinId, new MemberAssignment(mkAssignment(
put(member2RejoinId, new MemberAssignmentImpl(mkAssignment(
mkTopicAssignment(fooTopicId, 3, 4, 5),
mkTopicAssignment(barTopicId, 2)
)));
@ -1680,7 +1681,7 @@ public class GroupMetadataManagerTest {
assignor.prepareGroupAssignment(new GroupAssignment(
new HashMap<String, MemberAssignment>() {
{
put(memberId, new MemberAssignment(mkAssignment(
put(memberId, new MemberAssignmentImpl(mkAssignment(
mkTopicAssignment(fooTopicId, 0, 1)
)));
}
@ -1816,15 +1817,15 @@ public class GroupMetadataManagerTest {
assignor.prepareGroupAssignment(new GroupAssignment(
new HashMap<String, MemberAssignment>() {
{
put(memberId1, new MemberAssignment(mkAssignment(
put(memberId1, new MemberAssignmentImpl(mkAssignment(
mkTopicAssignment(fooTopicId, 0, 1),
mkTopicAssignment(barTopicId, 0)
)));
put(memberId2, new MemberAssignment(mkAssignment(
put(memberId2, new MemberAssignmentImpl(mkAssignment(
mkTopicAssignment(fooTopicId, 2, 3),
mkTopicAssignment(barTopicId, 2)
)));
put(memberId3, new MemberAssignment(mkAssignment(
put(memberId3, new MemberAssignmentImpl(mkAssignment(
mkTopicAssignment(fooTopicId, 4, 5),
mkTopicAssignment(barTopicId, 1)
)));
@ -2395,7 +2396,7 @@ public class GroupMetadataManagerTest {
// Prepare the assignment result.
assignor.prepareGroupAssignment(new GroupAssignment(
Collections.singletonMap(memberId, new MemberAssignment(mkAssignment(
Collections.singletonMap(memberId, new MemberAssignmentImpl(mkAssignment(
mkTopicAssignment(fooTopicId, 0, 1, 2, 3, 4, 5)
)))
));
@ -2506,7 +2507,7 @@ public class GroupMetadataManagerTest {
// Prepare the assignment result.
assignor.prepareGroupAssignment(new GroupAssignment(
Collections.singletonMap(memberId, new MemberAssignment(mkAssignment(
Collections.singletonMap(memberId, new MemberAssignmentImpl(mkAssignment(
mkTopicAssignment(fooTopicId, 0, 1, 2, 3, 4, 5)
)))
));
@ -2808,7 +2809,7 @@ public class GroupMetadataManagerTest {
.build();
assignor.prepareGroupAssignment(new GroupAssignment(
Collections.singletonMap(memberId, new MemberAssignment(mkAssignment(
Collections.singletonMap(memberId, new MemberAssignmentImpl(mkAssignment(
mkTopicAssignment(fooTopicId, 0, 1, 2, 3, 4, 5)
)))
));
@ -2883,7 +2884,7 @@ public class GroupMetadataManagerTest {
.build();
assignor.prepareGroupAssignment(new GroupAssignment(
Collections.singletonMap(memberId, new MemberAssignment(mkAssignment(
Collections.singletonMap(memberId, new MemberAssignmentImpl(mkAssignment(
mkTopicAssignment(fooTopicId, 0, 1, 2, 3, 4, 5)
)))
));
@ -2947,7 +2948,7 @@ public class GroupMetadataManagerTest {
.build();
assignor.prepareGroupAssignment(new GroupAssignment(
Collections.singletonMap(memberId, new MemberAssignment(mkAssignment(
Collections.singletonMap(memberId, new MemberAssignmentImpl(mkAssignment(
mkTopicAssignment(fooTopicId, 0, 1, 2, 3, 4, 5)
)))
));
@ -3031,7 +3032,7 @@ public class GroupMetadataManagerTest {
assignor.prepareGroupAssignment(new GroupAssignment(
new HashMap<String, MemberAssignment>() {
{
put(memberId1, new MemberAssignment(mkAssignment(
put(memberId1, new MemberAssignmentImpl(mkAssignment(
mkTopicAssignment(fooTopicId, 0, 1, 2)
)));
}
@ -3071,10 +3072,10 @@ public class GroupMetadataManagerTest {
assignor.prepareGroupAssignment(new GroupAssignment(
new HashMap<String, MemberAssignment>() {
{
put(memberId1, new MemberAssignment(mkAssignment(
put(memberId1, new MemberAssignmentImpl(mkAssignment(
mkTopicAssignment(fooTopicId, 0, 1)
)));
put(memberId2, new MemberAssignment(mkAssignment(
put(memberId2, new MemberAssignmentImpl(mkAssignment(
mkTopicAssignment(fooTopicId, 2)
)));
}
@ -3186,7 +3187,7 @@ public class GroupMetadataManagerTest {
assignor.prepareGroupAssignment(new GroupAssignment(
new HashMap<String, MemberAssignment>() {
{
put(memberId1, new MemberAssignment(mkAssignment(
put(memberId1, new MemberAssignmentImpl(mkAssignment(
mkTopicAssignment(fooTopicId, 0, 1, 2)
)));
}
@ -3226,10 +3227,10 @@ public class GroupMetadataManagerTest {
assignor.prepareGroupAssignment(new GroupAssignment(
new HashMap<String, MemberAssignment>() {
{
put(memberId1, new MemberAssignment(mkAssignment(
put(memberId1, new MemberAssignmentImpl(mkAssignment(
mkTopicAssignment(fooTopicId, 0, 1)
)));
put(memberId2, new MemberAssignment(mkAssignment(
put(memberId2, new MemberAssignmentImpl(mkAssignment(
mkTopicAssignment(fooTopicId, 2)
)));
}
@ -9233,7 +9234,7 @@ public class GroupMetadataManagerTest {
.build();
assignor.prepareGroupAssignment(new GroupAssignment(
Collections.singletonMap(memberId, new MemberAssignment(mkAssignment(
Collections.singletonMap(memberId, new MemberAssignmentImpl(mkAssignment(
mkTopicAssignment(fooTopicId, 0, 1, 2, 3, 4, 5),
mkTopicAssignment(barTopicId, 0, 1, 2)
)))
@ -9476,10 +9477,10 @@ public class GroupMetadataManagerTest {
assignor.prepareGroupAssignment(new GroupAssignment(
new HashMap<String, MemberAssignment>() {
{
put(memberId1, new MemberAssignment(mkAssignment(
put(memberId1, new MemberAssignmentImpl(mkAssignment(
mkTopicAssignment(fooTopicId, 0)
)));
put(memberId2, new MemberAssignment(mkAssignment(
put(memberId2, new MemberAssignmentImpl(mkAssignment(
mkTopicAssignment(barTopicId, 0)
)));
}
@ -9652,13 +9653,13 @@ public class GroupMetadataManagerTest {
assignor.prepareGroupAssignment(new GroupAssignment(
new HashMap<String, MemberAssignment>() {
{
put(memberId1, new MemberAssignment(mkAssignment(
put(memberId1, new MemberAssignmentImpl(mkAssignment(
mkTopicAssignment(fooTopicId, 0)
)));
put(memberId2, new MemberAssignment(mkAssignment(
put(memberId2, new MemberAssignmentImpl(mkAssignment(
mkTopicAssignment(barTopicId, 0)
)));
put(memberId3, new MemberAssignment(mkAssignment(
put(memberId3, new MemberAssignmentImpl(mkAssignment(
mkTopicAssignment(fooTopicId, 1)
)));
}
@ -9899,13 +9900,13 @@ public class GroupMetadataManagerTest {
assignor.prepareGroupAssignment(new GroupAssignment(
new HashMap<String, MemberAssignment>() {
{
put(memberId1, new MemberAssignment(mkAssignment(
put(memberId1, new MemberAssignmentImpl(mkAssignment(
mkTopicAssignment(fooTopicId, 0)
)));
put(memberId2, new MemberAssignment(mkAssignment(
put(memberId2, new MemberAssignmentImpl(mkAssignment(
mkTopicAssignment(barTopicId, 0)
)));
put(memberId3, new MemberAssignment(mkAssignment(
put(memberId3, new MemberAssignmentImpl(mkAssignment(
mkTopicAssignment(fooTopicId, 1)
)));
}
@ -10730,11 +10731,11 @@ public class GroupMetadataManagerTest {
assignor.prepareGroupAssignment(new GroupAssignment(
new HashMap<String, MemberAssignment>() {
{
put(memberId1, new MemberAssignment(mkAssignment(
put(memberId1, new MemberAssignmentImpl(mkAssignment(
mkTopicAssignment(fooTopicId, 0, 1, 2),
mkTopicAssignment(barTopicId, 0, 1)
)));
put(memberId2, new MemberAssignment(mkAssignment(
put(memberId2, new MemberAssignmentImpl(mkAssignment(
mkTopicAssignment(fooTopicId, 3, 4, 5)
)));
}
@ -10970,10 +10971,10 @@ public class GroupMetadataManagerTest {
assignor.prepareGroupAssignment(new GroupAssignment(
new HashMap<String, MemberAssignment>() {
{
put(memberId, new MemberAssignment(mkAssignment(
put(memberId, new MemberAssignmentImpl(mkAssignment(
mkTopicAssignment(fooTopicId, 0)
)));
put(newMemberId, new MemberAssignment(mkAssignment(
put(newMemberId, new MemberAssignmentImpl(mkAssignment(
mkTopicAssignment(barTopicId, 0)
)));
}
@ -11059,10 +11060,10 @@ public class GroupMetadataManagerTest {
assignor.prepareGroupAssignment(new GroupAssignment(
new HashMap<String, MemberAssignment>() {
{
put(memberId, new MemberAssignment(mkAssignment(
put(memberId, new MemberAssignmentImpl(mkAssignment(
mkTopicAssignment(fooTopicId, 0)
)));
put(newMemberId, new MemberAssignment(mkAssignment(
put(newMemberId, new MemberAssignmentImpl(mkAssignment(
mkTopicAssignment(fooTopicId, 1)
)));
}
@ -11389,11 +11390,11 @@ public class GroupMetadataManagerTest {
assignor.prepareGroupAssignment(new GroupAssignment(
new HashMap<String, MemberAssignment>() {
{
put(memberId1, new MemberAssignment(mkAssignment(
put(memberId1, new MemberAssignmentImpl(mkAssignment(
mkTopicAssignment(fooTopicId, 0),
mkTopicAssignment(zarTopicId, 0)
)));
put(memberId2, new MemberAssignment(mkAssignment(
put(memberId2, new MemberAssignmentImpl(mkAssignment(
mkTopicAssignment(barTopicId, 0),
mkTopicAssignment(fooTopicId, 1)
)));
@ -11619,11 +11620,11 @@ public class GroupMetadataManagerTest {
assignor.prepareGroupAssignment(new GroupAssignment(
new HashMap<String, MemberAssignment>() {
{
put(memberId1, new MemberAssignment(mkAssignment(
put(memberId1, new MemberAssignmentImpl(mkAssignment(
mkTopicAssignment(fooTopicId, 0, 1),
mkTopicAssignment(zarTopicId, 0)
)));
put(memberId2, new MemberAssignment(mkAssignment(
put(memberId2, new MemberAssignmentImpl(mkAssignment(
mkTopicAssignment(barTopicId, 0)
)));
}
@ -11861,11 +11862,11 @@ public class GroupMetadataManagerTest {
assignor.prepareGroupAssignment(new GroupAssignment(
new HashMap<String, MemberAssignment>() {
{
put(memberId1, new MemberAssignment(mkAssignment(
put(memberId1, new MemberAssignmentImpl(mkAssignment(
mkTopicAssignment(fooTopicId, 0, 1),
mkTopicAssignment(zarTopicId, 0)
)));
put(memberId2, new MemberAssignment(mkAssignment(
put(memberId2, new MemberAssignmentImpl(mkAssignment(
mkTopicAssignment(barTopicId, 0)
)));
}

2
group-coordinator/src/test/java/org/apache/kafka/coordinator/group/GroupMetadataManagerTestContext.java
View File

@ -46,7 +46,7 @@ import org.apache.kafka.common.security.auth.KafkaPrincipal;
import org.apache.kafka.common.security.auth.SecurityProtocol;
import org.apache.kafka.common.utils.LogContext;
import org.apache.kafka.common.utils.MockTime;
import org.apache.kafka.coordinator.group.assignor.ConsumerGroupPartitionAssignor;
import org.apache.kafka.coordinator.group.api.assignor.ConsumerGroupPartitionAssignor;
import org.apache.kafka.coordinator.group.classic.ClassicGroup;
import org.apache.kafka.coordinator.group.consumer.ConsumerGroup;
import org.apache.kafka.coordinator.group.consumer.ConsumerGroupBuilder;

12
group-coordinator/src/test/java/org/apache/kafka/coordinator/group/MockPartitionAssignor.java
View File

@ -17,11 +17,11 @@
package org.apache.kafka.coordinator.group;
import org.apache.kafka.common.Uuid;
import org.apache.kafka.coordinator.group.assignor.ConsumerGroupPartitionAssignor;
import org.apache.kafka.coordinator.group.assignor.GroupAssignment;
import org.apache.kafka.coordinator.group.assignor.GroupSpec;
import org.apache.kafka.coordinator.group.assignor.PartitionAssignorException;
import org.apache.kafka.coordinator.group.assignor.SubscribedTopicDescriber;
import org.apache.kafka.coordinator.group.api.assignor.ConsumerGroupPartitionAssignor;
import org.apache.kafka.coordinator.group.api.assignor.GroupAssignment;
import org.apache.kafka.coordinator.group.api.assignor.GroupSpec;
import org.apache.kafka.coordinator.group.api.assignor.PartitionAssignorException;
import org.apache.kafka.coordinator.group.api.assignor.SubscribedTopicDescriber;
import java.util.Map;
import java.util.Objects;
@ -51,6 +51,6 @@ public class MockPartitionAssignor implements ConsumerGroupPartitionAssignor {
public Map<Uuid, Set<Integer>> targetPartitions(String memberId) {
Objects.requireNonNull(prepareGroupAssignment);
return prepareGroupAssignment.members().get(memberId).targetPartitions();
return prepareGroupAssignment.members().get(memberId).partitions();
}
}

15
group-coordinator/src/test/java/org/apache/kafka/coordinator/group/NoOpPartitionAssignor.java
View File

@ -16,12 +16,12 @@
*/
package org.apache.kafka.coordinator.group;
import org.apache.kafka.coordinator.group.assignor.ConsumerGroupPartitionAssignor;
import org.apache.kafka.coordinator.group.assignor.GroupAssignment;
import org.apache.kafka.coordinator.group.assignor.GroupSpec;
import org.apache.kafka.coordinator.group.assignor.MemberAssignment;
import org.apache.kafka.coordinator.group.assignor.SubscribedTopicDescriber;
import org.apache.kafka.coordinator.group.api.assignor.ConsumerGroupPartitionAssignor;
import org.apache.kafka.coordinator.group.api.assignor.GroupAssignment;
import org.apache.kafka.coordinator.group.api.assignor.GroupSpec;
import org.apache.kafka.coordinator.group.api.assignor.SubscribedTopicDescriber;
import java.util.function.Function;
import java.util.stream.Collectors;
public class NoOpPartitionAssignor implements ConsumerGroupPartitionAssignor {
@ -36,9 +36,6 @@ public class NoOpPartitionAssignor implements ConsumerGroupPartitionAssignor {
public GroupAssignment assign(GroupSpec groupSpec, SubscribedTopicDescriber subscribedTopicDescriber) {
return new GroupAssignment(groupSpec.memberIds()
.stream()
.collect(Collectors.toMap(
memberId -> memberId,
memberId -> new MemberAssignment(groupSpec.memberAssignment(memberId))
)));
.collect(Collectors.toMap(Function.identity(), groupSpec::memberAssignment)));
}
}

90
group-coordinator/src/test/java/org/apache/kafka/coordinator/group/assignor/GeneralUniformAssignmentBuilderTest.java
View File

@ -17,8 +17,13 @@
package org.apache.kafka.coordinator.group.assignor;
import org.apache.kafka.common.Uuid;
import org.apache.kafka.coordinator.group.api.assignor.GroupAssignment;
import org.apache.kafka.coordinator.group.api.assignor.GroupSpec;
import org.apache.kafka.coordinator.group.api.assignor.PartitionAssignorException;
import org.apache.kafka.coordinator.group.consumer.Assignment;
import org.apache.kafka.coordinator.group.consumer.SubscribedTopicMetadata;
import org.apache.kafka.coordinator.group.consumer.GroupSpecImpl;
import org.apache.kafka.coordinator.group.consumer.MemberSubscriptionAndAssignmentImpl;
import org.apache.kafka.coordinator.group.consumer.SubscribedTopicDescriberImpl;
import org.apache.kafka.coordinator.group.consumer.TopicMetadata;
import org.junit.jupiter.api.Test;
@ -35,7 +40,7 @@ import static org.apache.kafka.coordinator.group.AssignmentTestUtil.mkAssignment
import static org.apache.kafka.coordinator.group.AssignmentTestUtil.mkTopicAssignment;
import static org.apache.kafka.coordinator.group.AssignmentTestUtil.invertedTargetAssignment;
import static org.apache.kafka.coordinator.group.CoordinatorRecordHelpersTest.mkMapOfPartitionRacks;
import static org.apache.kafka.coordinator.group.assignor.SubscriptionType.HETEROGENEOUS;
import static org.apache.kafka.coordinator.group.api.assignor.SubscriptionType.HETEROGENEOUS;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertThrows;
@ -55,7 +60,7 @@ public class GeneralUniformAssignmentBuilderTest {
@Test
public void testTwoMembersNoTopicSubscription() {
SubscribedTopicMetadata subscribedTopicMetadata = new SubscribedTopicMetadata(
SubscribedTopicDescriberImpl subscribedTopicMetadata = new SubscribedTopicDescriberImpl(
Collections.singletonMap(
topic1Uuid,
new TopicMetadata(
@ -67,13 +72,13 @@ public class GeneralUniformAssignmentBuilderTest {
)
);
Map<String, MemberSubscriptionSpecImpl> members = new TreeMap<>();
members.put(memberA, new MemberSubscriptionSpecImpl(
Map<String, MemberSubscriptionAndAssignmentImpl> members = new TreeMap<>();
members.put(memberA, new MemberSubscriptionAndAssignmentImpl(
Optional.empty(),
Collections.emptySet(),
Assignment.EMPTY
));
members.put(memberB, new MemberSubscriptionSpecImpl(
members.put(memberB, new MemberSubscriptionAndAssignmentImpl(
Optional.empty(),
Collections.emptySet(),
Assignment.EMPTY
@ -95,7 +100,7 @@ public class GeneralUniformAssignmentBuilderTest {
@Test
public void testTwoMembersSubscribedToNonexistentTopics() {
SubscribedTopicMetadata subscribedTopicMetadata = new SubscribedTopicMetadata(
SubscribedTopicDescriberImpl subscribedTopicMetadata = new SubscribedTopicDescriberImpl(
Collections.singletonMap(
topic1Uuid,
new TopicMetadata(
@ -107,13 +112,13 @@ public class GeneralUniformAssignmentBuilderTest {
)
);
Map<String, MemberSubscriptionSpecImpl> members = new TreeMap<>();
members.put(memberA, new MemberSubscriptionSpecImpl(
Map<String, MemberSubscriptionAndAssignmentImpl> members = new TreeMap<>();
members.put(memberA, new MemberSubscriptionAndAssignmentImpl(
Optional.empty(),
Collections.singleton(topic3Uuid),
Assignment.EMPTY
));
members.put(memberB, new MemberSubscriptionSpecImpl(
members.put(memberB, new MemberSubscriptionAndAssignmentImpl(
Optional.empty(),
Collections.singleton(topic2Uuid),
Assignment.EMPTY
@ -125,7 +130,8 @@ public class GeneralUniformAssignmentBuilderTest {
Collections.emptyMap()
);
assertThrows(PartitionAssignorException.class,
assertThrows(
PartitionAssignorException.class,
() -> assignor.assign(groupSpec, subscribedTopicMetadata));
}
@ -145,15 +151,15 @@ public class GeneralUniformAssignmentBuilderTest {
mkMapOfPartitionRacks(6)
));
Map<String, MemberSubscriptionSpecImpl> members = new TreeMap<>();
Map<String, MemberSubscriptionAndAssignmentImpl> members = new TreeMap<>();
members.put(memberA, new MemberSubscriptionSpecImpl(
members.put(memberA, new MemberSubscriptionAndAssignmentImpl(
Optional.empty(),
mkSet(topic1Uuid, topic3Uuid),
Assignment.EMPTY
));
members.put(memberB, new MemberSubscriptionSpecImpl(
members.put(memberB, new MemberSubscriptionAndAssignmentImpl(
Optional.empty(),
Collections.singleton(topic3Uuid),
Assignment.EMPTY
@ -164,7 +170,7 @@ public class GeneralUniformAssignmentBuilderTest {
HETEROGENEOUS,
Collections.emptyMap()
);
SubscribedTopicMetadata subscribedTopicMetadata = new SubscribedTopicMetadata(topicMetadata);
SubscribedTopicDescriberImpl subscribedTopicMetadata = new SubscribedTopicDescriberImpl(topicMetadata);
GroupAssignment computedAssignment = assignor.assign(
groupSpec,
@ -199,21 +205,21 @@ public class GeneralUniformAssignmentBuilderTest {
mkMapOfPartitionRacks(2)
));
Map<String, MemberSubscriptionSpecImpl> members = new TreeMap<>();
Map<String, MemberSubscriptionAndAssignmentImpl> members = new TreeMap<>();
members.put(memberA, new MemberSubscriptionSpecImpl(
members.put(memberA, new MemberSubscriptionAndAssignmentImpl(
Optional.empty(),
Collections.singleton(topic3Uuid),
Assignment.EMPTY
));
members.put(memberB, new MemberSubscriptionSpecImpl(
members.put(memberB, new MemberSubscriptionAndAssignmentImpl(
Optional.empty(),
Collections.singleton(topic3Uuid),
Assignment.EMPTY
));
members.put(memberC, new MemberSubscriptionSpecImpl(
members.put(memberC, new MemberSubscriptionAndAssignmentImpl(
Optional.empty(),
Collections.singleton(topic1Uuid),
Assignment.EMPTY
@ -224,7 +230,7 @@ public class GeneralUniformAssignmentBuilderTest {
HETEROGENEOUS,
Collections.emptyMap()
);
SubscribedTopicMetadata subscribedTopicMetadata = new SubscribedTopicMetadata(topicMetadata);
SubscribedTopicDescriberImpl subscribedTopicMetadata = new SubscribedTopicDescriberImpl(topicMetadata);
GroupAssignment computedAssignment = assignor.assign(
groupSpec,
@ -268,9 +274,9 @@ public class GeneralUniformAssignmentBuilderTest {
mkMapOfPartitionRacks(4)
));
Map<String, MemberSubscriptionSpecImpl> members = new TreeMap<>();
Map<String, MemberSubscriptionAndAssignmentImpl> members = new TreeMap<>();
members.put(memberA, new MemberSubscriptionSpecImpl(
members.put(memberA, new MemberSubscriptionAndAssignmentImpl(
Optional.empty(),
Collections.singleton(topic1Uuid),
new Assignment(mkAssignment(
@ -278,7 +284,7 @@ public class GeneralUniformAssignmentBuilderTest {
))
));
members.put(memberB, new MemberSubscriptionSpecImpl(
members.put(memberB, new MemberSubscriptionAndAssignmentImpl(
Optional.empty(),
mkSet(topic1Uuid, topic2Uuid),
new Assignment(mkAssignment(
@ -287,7 +293,7 @@ public class GeneralUniformAssignmentBuilderTest {
))
));
members.put(memberC, new MemberSubscriptionSpecImpl(
members.put(memberC, new MemberSubscriptionAndAssignmentImpl(
Optional.empty(),
mkSet(topic1Uuid, topic2Uuid, topic3Uuid),
new Assignment(mkAssignment(
@ -302,7 +308,7 @@ public class GeneralUniformAssignmentBuilderTest {
HETEROGENEOUS,
invertedTargetAssignment(members)
);
SubscribedTopicMetadata subscribedTopicMetadata = new SubscribedTopicMetadata(topicMetadata);
SubscribedTopicDescriberImpl subscribedTopicMetadata = new SubscribedTopicDescriberImpl(topicMetadata);
GroupAssignment computedAssignment = assignor.assign(
groupSpec,
@ -353,9 +359,9 @@ public class GeneralUniformAssignmentBuilderTest {
mkMapOfPartitionRacks(3)
));
Map<String, MemberSubscriptionSpecImpl> members = new TreeMap<>();
Map<String, MemberSubscriptionAndAssignmentImpl> members = new TreeMap<>();
members.put(memberA, new MemberSubscriptionSpecImpl(
members.put(memberA, new MemberSubscriptionAndAssignmentImpl(
Optional.empty(),
mkSet(topic1Uuid, topic3Uuid),
new Assignment(mkAssignment(
@ -364,7 +370,7 @@ public class GeneralUniformAssignmentBuilderTest {
))
));
members.put(memberB, new MemberSubscriptionSpecImpl(
members.put(memberB, new MemberSubscriptionAndAssignmentImpl(
Optional.empty(),
mkSet(topic1Uuid, topic2Uuid, topic3Uuid, topic4Uuid),
new Assignment(mkAssignment(
@ -378,7 +384,7 @@ public class GeneralUniformAssignmentBuilderTest {
HETEROGENEOUS,
invertedTargetAssignment(members)
);
SubscribedTopicMetadata subscribedTopicMetadata = new SubscribedTopicMetadata(topicMetadata);
SubscribedTopicDescriberImpl subscribedTopicMetadata = new SubscribedTopicDescriberImpl(topicMetadata);
GroupAssignment computedAssignment = assignor.assign(
groupSpec,
@ -415,9 +421,9 @@ public class GeneralUniformAssignmentBuilderTest {
mkMapOfPartitionRacks(7)
));
Map<String, MemberSubscriptionSpecImpl> members = new TreeMap<>();
Map<String, MemberSubscriptionAndAssignmentImpl> members = new TreeMap<>();
members.put(memberA, new MemberSubscriptionSpecImpl(
members.put(memberA, new MemberSubscriptionAndAssignmentImpl(
Optional.empty(),
Collections.singleton(topic1Uuid),
new Assignment(mkAssignment(
@ -426,7 +432,7 @@ public class GeneralUniformAssignmentBuilderTest {
))
));
members.put(memberB, new MemberSubscriptionSpecImpl(
members.put(memberB, new MemberSubscriptionAndAssignmentImpl(
Optional.empty(),
mkSet(topic1Uuid, topic2Uuid),
new Assignment(mkAssignment(
@ -436,7 +442,7 @@ public class GeneralUniformAssignmentBuilderTest {
));
// Add a new member to trigger a re-assignment.
members.put(memberC, new MemberSubscriptionSpecImpl(
members.put(memberC, new MemberSubscriptionAndAssignmentImpl(
Optional.empty(),
mkSet(topic1Uuid, topic2Uuid),
Assignment.EMPTY
@ -447,7 +453,7 @@ public class GeneralUniformAssignmentBuilderTest {
HETEROGENEOUS,
invertedTargetAssignment(members)
);
SubscribedTopicMetadata subscribedTopicMetadata = new SubscribedTopicMetadata(topicMetadata);
SubscribedTopicDescriberImpl subscribedTopicMetadata = new SubscribedTopicDescriberImpl(topicMetadata);
GroupAssignment computedAssignment = assignor.assign(
groupSpec,
@ -491,9 +497,9 @@ public class GeneralUniformAssignmentBuilderTest {
mkMapOfPartitionRacks(3)
));
Map<String, MemberSubscriptionSpecImpl> members = new TreeMap<>();
Map<String, MemberSubscriptionAndAssignmentImpl> members = new TreeMap<>();
members.put(memberA, new MemberSubscriptionSpecImpl(
members.put(memberA, new MemberSubscriptionAndAssignmentImpl(
Optional.empty(),
mkSet(topic1Uuid, topic3Uuid),
new Assignment(mkAssignment(
@ -502,7 +508,7 @@ public class GeneralUniformAssignmentBuilderTest {
))
));
members.put(memberB, new MemberSubscriptionSpecImpl(
members.put(memberB, new MemberSubscriptionAndAssignmentImpl(
Optional.empty(),
Collections.singleton(topic2Uuid),
new Assignment(mkAssignment(
@ -517,7 +523,7 @@ public class GeneralUniformAssignmentBuilderTest {
HETEROGENEOUS,
invertedTargetAssignment(members)
);
SubscribedTopicMetadata subscribedTopicMetadata = new SubscribedTopicMetadata(topicMetadata);
SubscribedTopicDescriberImpl subscribedTopicMetadata = new SubscribedTopicDescriberImpl(topicMetadata);
GroupAssignment computedAssignment = assignor.assign(
groupSpec,
@ -553,9 +559,9 @@ public class GeneralUniformAssignmentBuilderTest {
));
// Initial subscriptions were [T1, T2]
Map<String, MemberSubscriptionSpecImpl> members = new TreeMap<>();
Map<String, MemberSubscriptionAndAssignmentImpl> members = new TreeMap<>();
members.put(memberA, new MemberSubscriptionSpecImpl(
members.put(memberA, new MemberSubscriptionAndAssignmentImpl(
Optional.empty(),
Collections.singleton(topic1Uuid),
new Assignment(mkAssignment(
@ -564,7 +570,7 @@ public class GeneralUniformAssignmentBuilderTest {
))
));
members.put(memberB, new MemberSubscriptionSpecImpl(
members.put(memberB, new MemberSubscriptionAndAssignmentImpl(
Optional.empty(),
mkSet(topic1Uuid, topic2Uuid),
new Assignment(mkAssignment(
@ -578,7 +584,7 @@ public class GeneralUniformAssignmentBuilderTest {
HETEROGENEOUS,
invertedTargetAssignment(members)
);
SubscribedTopicMetadata subscribedTopicMetadata = new SubscribedTopicMetadata(topicMetadata);
SubscribedTopicDescriberImpl subscribedTopicMetadata = new SubscribedTopicDescriberImpl(topicMetadata);
GroupAssignment computedAssignment = assignor.assign(
groupSpec,

13
group-coordinator/src/test/java/org/apache/kafka/coordinator/group/assignor/GroupSpecImplTest.java
View File

@ -17,7 +17,10 @@
package org.apache.kafka.coordinator.group.assignor;
import org.apache.kafka.common.Uuid;
import org.apache.kafka.coordinator.group.api.assignor.SubscriptionType;
import org.apache.kafka.coordinator.group.consumer.Assignment;
import org.apache.kafka.coordinator.group.consumer.GroupSpecImpl;
import org.apache.kafka.coordinator.group.consumer.MemberSubscriptionAndAssignmentImpl;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
@ -36,7 +39,7 @@ import static org.junit.jupiter.api.Assertions.assertTrue;
public class GroupSpecImplTest {
private static final String TEST_MEMBER = "test-member";
private Map<String, MemberSubscriptionSpecImpl> members;
private Map<String, MemberSubscriptionAndAssignmentImpl> members;
private SubscriptionType subscriptionType;
private Map<Uuid, Map<Integer, String>> invertedTargetAssignment;
private GroupSpecImpl groupSpec;
@ -49,7 +52,7 @@ public class GroupSpecImplTest {
invertedTargetAssignment = new HashMap<>();
topicId = Uuid.randomUuid();
members.put(TEST_MEMBER, new MemberSubscriptionSpecImpl(
members.put(TEST_MEMBER, new MemberSubscriptionAndAssignmentImpl(
Optional.empty(),
mkSet(topicId),
Assignment.EMPTY
@ -96,13 +99,13 @@ public class GroupSpecImplTest {
topicId,
mkSet(0, 1)
);
members.put(TEST_MEMBER, new MemberSubscriptionSpecImpl(
members.put(TEST_MEMBER, new MemberSubscriptionAndAssignmentImpl(
Optional.empty(),
mkSet(topicId),
new Assignment(topicPartitions)
));
assertEquals(topicPartitions, groupSpec.memberAssignment(TEST_MEMBER));
assertEquals(Collections.emptyMap(), groupSpec.memberAssignment("unknown-member"));
assertEquals(topicPartitions, groupSpec.memberAssignment(TEST_MEMBER).partitions());
assertEquals(Collections.emptyMap(), groupSpec.memberAssignment("unknown-member").partitions());
}
}

95
group-coordinator/src/test/java/org/apache/kafka/coordinator/group/assignor/OptimizedUniformAssignmentBuilderTest.java
View File

@ -17,8 +17,13 @@
package org.apache.kafka.coordinator.group.assignor;
import org.apache.kafka.common.Uuid;
import org.apache.kafka.coordinator.group.api.assignor.GroupAssignment;
import org.apache.kafka.coordinator.group.api.assignor.GroupSpec;
import org.apache.kafka.coordinator.group.api.assignor.PartitionAssignorException;
import org.apache.kafka.coordinator.group.consumer.Assignment;
import org.apache.kafka.coordinator.group.consumer.SubscribedTopicMetadata;
import org.apache.kafka.coordinator.group.consumer.GroupSpecImpl;
import org.apache.kafka.coordinator.group.consumer.MemberSubscriptionAndAssignmentImpl;
import org.apache.kafka.coordinator.group.consumer.SubscribedTopicDescriberImpl;
import org.apache.kafka.coordinator.group.consumer.TopicMetadata;
import org.junit.jupiter.api.Test;
@ -39,7 +44,7 @@ import static org.apache.kafka.coordinator.group.AssignmentTestUtil.mkOrderedAss
import static org.apache.kafka.coordinator.group.AssignmentTestUtil.mkTopicAssignment;
import static org.apache.kafka.coordinator.group.AssignmentTestUtil.invertedTargetAssignment;
import static org.apache.kafka.coordinator.group.CoordinatorRecordHelpersTest.mkMapOfPartitionRacks;
import static org.apache.kafka.coordinator.group.assignor.SubscriptionType.HOMOGENEOUS;
import static org.apache.kafka.coordinator.group.api.assignor.SubscriptionType.HOMOGENEOUS;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertThrows;
import static org.junit.jupiter.api.Assertions.assertTrue;
@ -58,7 +63,7 @@ public class OptimizedUniformAssignmentBuilderTest {
@Test
public void testOneMemberNoTopicSubscription() {
SubscribedTopicMetadata subscribedTopicMetadata = new SubscribedTopicMetadata(
SubscribedTopicDescriberImpl subscribedTopicMetadata = new SubscribedTopicDescriberImpl(
Collections.singletonMap(
topic1Uuid,
new TopicMetadata(
@ -70,9 +75,9 @@ public class OptimizedUniformAssignmentBuilderTest {
)
);
Map<String, MemberSubscriptionSpecImpl> members = Collections.singletonMap(
Map<String, MemberSubscriptionAndAssignmentImpl> members = Collections.singletonMap(
memberA,
new MemberSubscriptionSpecImpl(
new MemberSubscriptionAndAssignmentImpl(
Optional.empty(),
Collections.emptySet(),
Assignment.EMPTY
@ -95,7 +100,7 @@ public class OptimizedUniformAssignmentBuilderTest {
@Test
public void testOneMemberSubscribedToNonexistentTopic() {
SubscribedTopicMetadata subscribedTopicMetadata = new SubscribedTopicMetadata(
SubscribedTopicDescriberImpl subscribedTopicMetadata = new SubscribedTopicDescriberImpl(
Collections.singletonMap(
topic1Uuid,
new TopicMetadata(
@ -107,9 +112,9 @@ public class OptimizedUniformAssignmentBuilderTest {
)
);
Map<String, MemberSubscriptionSpecImpl> members = Collections.singletonMap(
Map<String, MemberSubscriptionAndAssignmentImpl> members = Collections.singletonMap(
memberA,
new MemberSubscriptionSpecImpl(
new MemberSubscriptionAndAssignmentImpl(
Optional.empty(),
Collections.singleton(topic2Uuid),
Assignment.EMPTY
@ -142,15 +147,15 @@ public class OptimizedUniformAssignmentBuilderTest {
mkMapOfPartitionRacks(2)
));
Map<String, MemberSubscriptionSpecImpl> members = new TreeMap<>();
Map<String, MemberSubscriptionAndAssignmentImpl> members = new TreeMap<>();
members.put(memberA, new MemberSubscriptionSpecImpl(
members.put(memberA, new MemberSubscriptionAndAssignmentImpl(
Optional.empty(),
mkSet(topic1Uuid, topic3Uuid),
Assignment.EMPTY
));
members.put(memberB, new MemberSubscriptionSpecImpl(
members.put(memberB, new MemberSubscriptionAndAssignmentImpl(
Optional.empty(),
mkSet(topic1Uuid, topic3Uuid),
Assignment.EMPTY
@ -170,7 +175,7 @@ public class OptimizedUniformAssignmentBuilderTest {
HOMOGENEOUS,
Collections.emptyMap()
);
SubscribedTopicMetadata subscribedTopicMetadata = new SubscribedTopicMetadata(topicMetadata);
SubscribedTopicDescriberImpl subscribedTopicMetadata = new SubscribedTopicDescriberImpl(topicMetadata);
GroupAssignment computedAssignment = assignor.assign(
groupSpec,
@ -191,21 +196,21 @@ public class OptimizedUniformAssignmentBuilderTest {
mkMapOfPartitionRacks(2)
));
Map<String, MemberSubscriptionSpecImpl> members = new TreeMap<>();
Map<String, MemberSubscriptionAndAssignmentImpl> members = new TreeMap<>();
members.put(memberA, new MemberSubscriptionSpecImpl(
members.put(memberA, new MemberSubscriptionAndAssignmentImpl(
Optional.empty(),
Collections.singleton(topic3Uuid),
Assignment.EMPTY
));
members.put(memberB, new MemberSubscriptionSpecImpl(
members.put(memberB, new MemberSubscriptionAndAssignmentImpl(
Optional.empty(),
Collections.singleton(topic3Uuid),
Assignment.EMPTY
));
members.put(memberC, new MemberSubscriptionSpecImpl(
members.put(memberC, new MemberSubscriptionAndAssignmentImpl(
Optional.empty(),
Collections.singleton(topic3Uuid),
Assignment.EMPTY
@ -228,7 +233,7 @@ public class OptimizedUniformAssignmentBuilderTest {
HOMOGENEOUS,
Collections.emptyMap()
);
SubscribedTopicMetadata subscribedTopicMetadata = new SubscribedTopicMetadata(topicMetadata);
SubscribedTopicDescriberImpl subscribedTopicMetadata = new SubscribedTopicDescriberImpl(topicMetadata);
GroupAssignment computedAssignment = assignor.assign(
groupSpec,
@ -252,9 +257,9 @@ public class OptimizedUniformAssignmentBuilderTest {
));
}
Map<String, MemberSubscriptionSpecImpl> members = new TreeMap<>();
Map<String, MemberSubscriptionAndAssignmentImpl> members = new TreeMap<>();
for (int i = 1; i < 50; i++) {
members.put("member" + i, new MemberSubscriptionSpecImpl(
members.put("member" + i, new MemberSubscriptionAndAssignmentImpl(
Optional.empty(),
topicMetadata.keySet(),
Assignment.EMPTY
@ -266,7 +271,7 @@ public class OptimizedUniformAssignmentBuilderTest {
HOMOGENEOUS,
Collections.emptyMap()
);
SubscribedTopicMetadata subscribedTopicMetadata = new SubscribedTopicMetadata(topicMetadata);
SubscribedTopicDescriberImpl subscribedTopicMetadata = new SubscribedTopicDescriberImpl(topicMetadata);
GroupAssignment computedAssignment = assignor.assign(
groupSpec,
@ -292,9 +297,9 @@ public class OptimizedUniformAssignmentBuilderTest {
mkMapOfPartitionRacks(3)
));
Map<String, MemberSubscriptionSpecImpl> members = new TreeMap<>();
Map<String, MemberSubscriptionAndAssignmentImpl> members = new TreeMap<>();
members.put(memberA, new MemberSubscriptionSpecImpl(
members.put(memberA, new MemberSubscriptionAndAssignmentImpl(
Optional.empty(),
mkSet(topic1Uuid, topic2Uuid),
new Assignment(mkOrderedAssignment(
@ -303,7 +308,7 @@ public class OptimizedUniformAssignmentBuilderTest {
))
));
members.put(memberB, new MemberSubscriptionSpecImpl(
members.put(memberB, new MemberSubscriptionAndAssignmentImpl(
Optional.empty(),
mkSet(topic1Uuid, topic2Uuid),
new Assignment(mkOrderedAssignment(
@ -327,7 +332,7 @@ public class OptimizedUniformAssignmentBuilderTest {
HOMOGENEOUS,
invertedTargetAssignment(members)
);
SubscribedTopicMetadata subscribedTopicMetadata = new SubscribedTopicMetadata(topicMetadata);
SubscribedTopicDescriberImpl subscribedTopicMetadata = new SubscribedTopicDescriberImpl(topicMetadata);
GroupAssignment computedAssignment = assignor.assign(
groupSpec,
@ -355,9 +360,9 @@ public class OptimizedUniformAssignmentBuilderTest {
mkMapOfPartitionRacks(5)
));
Map<String, MemberSubscriptionSpecImpl> members = new TreeMap<>();
Map<String, MemberSubscriptionAndAssignmentImpl> members = new TreeMap<>();
members.put(memberA, new MemberSubscriptionSpecImpl(
members.put(memberA, new MemberSubscriptionAndAssignmentImpl(
Optional.empty(),
mkSet(topic1Uuid, topic2Uuid),
new Assignment(mkOrderedAssignment(
@ -366,7 +371,7 @@ public class OptimizedUniformAssignmentBuilderTest {
))
));
members.put(memberB, new MemberSubscriptionSpecImpl(
members.put(memberB, new MemberSubscriptionAndAssignmentImpl(
Optional.empty(),
mkSet(topic1Uuid, topic2Uuid),
new Assignment(mkOrderedAssignment(
@ -390,7 +395,7 @@ public class OptimizedUniformAssignmentBuilderTest {
HOMOGENEOUS,
invertedTargetAssignment(members)
);
SubscribedTopicMetadata subscribedTopicMetadata = new SubscribedTopicMetadata(topicMetadata);
SubscribedTopicDescriberImpl subscribedTopicMetadata = new SubscribedTopicDescriberImpl(topicMetadata);
GroupAssignment computedAssignment = assignor.assign(
groupSpec,
@ -417,9 +422,9 @@ public class OptimizedUniformAssignmentBuilderTest {
mkMapOfPartitionRacks(3)
));
Map<String, MemberSubscriptionSpecImpl> members = new TreeMap<>();
Map<String, MemberSubscriptionAndAssignmentImpl> members = new TreeMap<>();
members.put(memberA, new MemberSubscriptionSpecImpl(
members.put(memberA, new MemberSubscriptionAndAssignmentImpl(
Optional.empty(),
mkSet(topic1Uuid, topic2Uuid),
new Assignment(mkOrderedAssignment(
@ -428,7 +433,7 @@ public class OptimizedUniformAssignmentBuilderTest {
))
));
members.put(memberB, new MemberSubscriptionSpecImpl(
members.put(memberB, new MemberSubscriptionAndAssignmentImpl(
Optional.empty(),
mkSet(topic1Uuid, topic2Uuid),
new Assignment(mkOrderedAssignment(
@ -438,7 +443,7 @@ public class OptimizedUniformAssignmentBuilderTest {
));
// Add a new member to trigger a re-assignment.
members.put(memberC, new MemberSubscriptionSpecImpl(
members.put(memberC, new MemberSubscriptionAndAssignmentImpl(
Optional.empty(),
mkSet(topic1Uuid, topic2Uuid),
Assignment.EMPTY
@ -461,7 +466,7 @@ public class OptimizedUniformAssignmentBuilderTest {
HOMOGENEOUS,
invertedTargetAssignment(members)
);
SubscribedTopicMetadata subscribedTopicMetadata = new SubscribedTopicMetadata(topicMetadata);
SubscribedTopicDescriberImpl subscribedTopicMetadata = new SubscribedTopicDescriberImpl(topicMetadata);
GroupAssignment computedAssignment = assignor.assign(
groupSpec,
@ -488,9 +493,9 @@ public class OptimizedUniformAssignmentBuilderTest {
mkMapOfPartitionRacks(3)
));
Map<String, MemberSubscriptionSpecImpl> members = new TreeMap<>();
Map<String, MemberSubscriptionAndAssignmentImpl> members = new TreeMap<>();
members.put(memberA, new MemberSubscriptionSpecImpl(
members.put(memberA, new MemberSubscriptionAndAssignmentImpl(
Optional.empty(),
mkSet(topic1Uuid, topic2Uuid),
new Assignment(mkAssignment(
@ -499,7 +504,7 @@ public class OptimizedUniformAssignmentBuilderTest {
))
));
members.put(memberB, new MemberSubscriptionSpecImpl(
members.put(memberB, new MemberSubscriptionAndAssignmentImpl(
Optional.empty(),
mkSet(topic1Uuid, topic2Uuid),
new Assignment(mkAssignment(
@ -525,7 +530,7 @@ public class OptimizedUniformAssignmentBuilderTest {
HOMOGENEOUS,
invertedTargetAssignment(members)
);
SubscribedTopicMetadata subscribedTopicMetadata = new SubscribedTopicMetadata(topicMetadata);
SubscribedTopicDescriberImpl subscribedTopicMetadata = new SubscribedTopicDescriberImpl(topicMetadata);
GroupAssignment computedAssignment = assignor.assign(
groupSpec,
@ -553,9 +558,9 @@ public class OptimizedUniformAssignmentBuilderTest {
));
// Initial subscriptions were [T1, T2]
Map<String, MemberSubscriptionSpecImpl> members = new TreeMap<>();
Map<String, MemberSubscriptionAndAssignmentImpl> members = new TreeMap<>();
members.put(memberA, new MemberSubscriptionSpecImpl(
members.put(memberA, new MemberSubscriptionAndAssignmentImpl(
Optional.empty(),
Collections.singleton(topic2Uuid),
new Assignment(mkAssignment(
@ -564,7 +569,7 @@ public class OptimizedUniformAssignmentBuilderTest {
))
));
members.put(memberB, new MemberSubscriptionSpecImpl(
members.put(memberB, new MemberSubscriptionAndAssignmentImpl(
Optional.empty(),
Collections.singleton(topic2Uuid),
new Assignment(mkAssignment(
@ -586,7 +591,7 @@ public class OptimizedUniformAssignmentBuilderTest {
HOMOGENEOUS,
invertedTargetAssignment(members)
);
SubscribedTopicMetadata subscribedTopicMetadata = new SubscribedTopicMetadata(topicMetadata);
SubscribedTopicDescriberImpl subscribedTopicMetadata = new SubscribedTopicDescriberImpl(topicMetadata);
GroupAssignment computedAssignment = assignor.assign(
groupSpec,
@ -610,7 +615,7 @@ public class OptimizedUniformAssignmentBuilderTest {
* @param computedGroupAssignment Assignment computed by the uniform assignor.
*/
private void checkValidityAndBalance(
Map<String, MemberSubscriptionSpecImpl> memberSubscriptionSpec,
Map<String, MemberSubscriptionAndAssignmentImpl> memberSubscriptionSpec,
GroupAssignment computedGroupAssignment
) {
List<String> membersList = new ArrayList<>(computedGroupAssignment.members().keySet());
@ -618,7 +623,7 @@ public class OptimizedUniformAssignmentBuilderTest {
List<Integer> totalAssignmentSizesOfAllMembers = new ArrayList<>(membersList.size());
membersList.forEach(member -> {
Map<Uuid, Set<Integer>> computedAssignmentForMember = computedGroupAssignment
.members().get(member).targetPartitions();
.members().get(member).partitions();
int sum = computedAssignmentForMember.values().stream().mapToInt(Set::size).sum();
totalAssignmentSizesOfAllMembers.add(sum);
});
@ -626,7 +631,7 @@ public class OptimizedUniformAssignmentBuilderTest {
for (int i = 0; i < numMembers; i++) {
String memberId = membersList.get(i);
Map<Uuid, Set<Integer>> computedAssignmentForMember =
computedGroupAssignment.members().get(memberId).targetPartitions();
computedGroupAssignment.members().get(memberId).partitions();
// Each member is subscribed to topics of all the partitions assigned to it.
computedAssignmentForMember.keySet().forEach(topicId -> {
// Check if the topic exists in the subscription.
@ -638,7 +643,7 @@ public class OptimizedUniformAssignmentBuilderTest {
for (int j = i + 1; j < numMembers; j++) {
String otherMemberId = membersList.get(j);
Map<Uuid, Set<Integer>> computedAssignmentForOtherMember = computedGroupAssignment
.members().get(otherMemberId).targetPartitions();
.members().get(otherMemberId).partitions();
// Each partition should be assigned to at most one member
computedAssignmentForMember.keySet().forEach(topicId -> {
Set<Integer> intersection = new HashSet<>();

107
group-coordinator/src/test/java/org/apache/kafka/coordinator/group/assignor/RangeAssignorTest.java
View File

@ -17,8 +17,13 @@
package org.apache.kafka.coordinator.group.assignor;
import org.apache.kafka.common.Uuid;
import org.apache.kafka.coordinator.group.api.assignor.GroupAssignment;
import org.apache.kafka.coordinator.group.api.assignor.GroupSpec;
import org.apache.kafka.coordinator.group.api.assignor.PartitionAssignorException;
import org.apache.kafka.coordinator.group.consumer.Assignment;
import org.apache.kafka.coordinator.group.consumer.SubscribedTopicMetadata;
import org.apache.kafka.coordinator.group.consumer.GroupSpecImpl;
import org.apache.kafka.coordinator.group.consumer.MemberSubscriptionAndAssignmentImpl;
import org.apache.kafka.coordinator.group.consumer.SubscribedTopicDescriberImpl;
import org.apache.kafka.coordinator.group.consumer.TopicMetadata;
import org.junit.jupiter.api.Test;
@ -33,8 +38,8 @@ import static org.apache.kafka.common.utils.Utils.mkSet;
import static org.apache.kafka.coordinator.group.AssignmentTestUtil.mkAssignment;
import static org.apache.kafka.coordinator.group.AssignmentTestUtil.mkTopicAssignment;
import static org.apache.kafka.coordinator.group.AssignmentTestUtil.invertedTargetAssignment;
import static org.apache.kafka.coordinator.group.assignor.SubscriptionType.HETEROGENEOUS;
import static org.apache.kafka.coordinator.group.assignor.SubscriptionType.HOMOGENEOUS;
import static org.apache.kafka.coordinator.group.api.assignor.SubscriptionType.HETEROGENEOUS;
import static org.apache.kafka.coordinator.group.api.assignor.SubscriptionType.HOMOGENEOUS;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertNull;
import static org.junit.jupiter.api.Assertions.assertThrows;
@ -53,7 +58,7 @@ public class RangeAssignorTest {
@Test
public void testOneConsumerNoTopic() {
SubscribedTopicMetadata subscribedTopicMetadata = new SubscribedTopicMetadata(
SubscribedTopicDescriberImpl subscribedTopicMetadata = new SubscribedTopicDescriberImpl(
Collections.singletonMap(
topic1Uuid,
new TopicMetadata(
@ -65,9 +70,9 @@ public class RangeAssignorTest {
)
);
Map<String, MemberSubscriptionSpecImpl> members = Collections.singletonMap(
Map<String, MemberSubscriptionAndAssignmentImpl> members = Collections.singletonMap(
memberA,
new MemberSubscriptionSpecImpl(
new MemberSubscriptionAndAssignmentImpl(
Optional.empty(),
Collections.emptySet(),
Assignment.EMPTY
@ -90,7 +95,7 @@ public class RangeAssignorTest {
@Test
public void testOneConsumerSubscribedToNonExistentTopic() {
SubscribedTopicMetadata subscribedTopicMetadata = new SubscribedTopicMetadata(
SubscribedTopicDescriberImpl subscribedTopicMetadata = new SubscribedTopicDescriberImpl(
Collections.singletonMap(
topic1Uuid,
new TopicMetadata(
@ -102,9 +107,9 @@ public class RangeAssignorTest {
)
);
Map<String, MemberSubscriptionSpecImpl> members = Collections.singletonMap(
Map<String, MemberSubscriptionAndAssignmentImpl> members = Collections.singletonMap(
memberA,
new MemberSubscriptionSpecImpl(
new MemberSubscriptionAndAssignmentImpl(
Optional.empty(),
mkSet(topic2Uuid),
Assignment.EMPTY
@ -137,15 +142,15 @@ public class RangeAssignorTest {
Collections.emptyMap()
));
Map<String, MemberSubscriptionSpecImpl> members = new TreeMap<>();
Map<String, MemberSubscriptionAndAssignmentImpl> members = new TreeMap<>();
members.put(memberA, new MemberSubscriptionSpecImpl(
members.put(memberA, new MemberSubscriptionAndAssignmentImpl(
Optional.empty(),
mkSet(topic1Uuid, topic3Uuid),
Assignment.EMPTY
));
members.put(memberB, new MemberSubscriptionSpecImpl(
members.put(memberB, new MemberSubscriptionAndAssignmentImpl(
Optional.empty(),
mkSet(topic1Uuid, topic3Uuid),
Assignment.EMPTY
@ -156,7 +161,7 @@ public class RangeAssignorTest {
HOMOGENEOUS,
invertedTargetAssignment(members)
);
SubscribedTopicMetadata subscribedTopicMetadata = new SubscribedTopicMetadata(topicMetadata);
SubscribedTopicDescriberImpl subscribedTopicMetadata = new SubscribedTopicDescriberImpl(topicMetadata);
GroupAssignment computedAssignment = assignor.assign(
groupSpec,
@ -198,21 +203,21 @@ public class RangeAssignorTest {
Collections.emptyMap()
));
Map<String, MemberSubscriptionSpecImpl> members = new TreeMap<>();
Map<String, MemberSubscriptionAndAssignmentImpl> members = new TreeMap<>();
members.put(memberA, new MemberSubscriptionSpecImpl(
members.put(memberA, new MemberSubscriptionAndAssignmentImpl(
Optional.empty(),
mkSet(topic1Uuid, topic2Uuid),
Assignment.EMPTY
));
members.put(memberB, new MemberSubscriptionSpecImpl(
members.put(memberB, new MemberSubscriptionAndAssignmentImpl(
Optional.empty(),
mkSet(topic3Uuid),
Assignment.EMPTY
));
members.put(memberC, new MemberSubscriptionSpecImpl(
members.put(memberC, new MemberSubscriptionAndAssignmentImpl(
Optional.empty(),
mkSet(topic2Uuid, topic3Uuid),
Assignment.EMPTY
@ -223,7 +228,7 @@ public class RangeAssignorTest {
HETEROGENEOUS,
invertedTargetAssignment(members)
);
SubscribedTopicMetadata subscribedTopicMetadata = new SubscribedTopicMetadata(topicMetadata);
SubscribedTopicDescriberImpl subscribedTopicMetadata = new SubscribedTopicDescriberImpl(topicMetadata);
GroupAssignment computedAssignment = assignor.assign(
groupSpec,
@ -262,21 +267,21 @@ public class RangeAssignorTest {
Collections.emptyMap()
));
Map<String, MemberSubscriptionSpecImpl> members = new TreeMap<>();
Map<String, MemberSubscriptionAndAssignmentImpl> members = new TreeMap<>();
members.put(memberA, new MemberSubscriptionSpecImpl(
members.put(memberA, new MemberSubscriptionAndAssignmentImpl(
Optional.empty(),
mkSet(topic1Uuid, topic3Uuid),
Assignment.EMPTY
));
members.put(memberB, new MemberSubscriptionSpecImpl(
members.put(memberB, new MemberSubscriptionAndAssignmentImpl(
Optional.empty(),
mkSet(topic1Uuid, topic3Uuid),
Assignment.EMPTY
));
members.put(memberC, new MemberSubscriptionSpecImpl(
members.put(memberC, new MemberSubscriptionAndAssignmentImpl(
Optional.empty(),
mkSet(topic1Uuid, topic3Uuid),
Assignment.EMPTY
@ -287,7 +292,7 @@ public class RangeAssignorTest {
HOMOGENEOUS,
invertedTargetAssignment(members)
);
SubscribedTopicMetadata subscribedTopicMetadata = new SubscribedTopicMetadata(topicMetadata);
SubscribedTopicDescriberImpl subscribedTopicMetadata = new SubscribedTopicDescriberImpl(topicMetadata);
GroupAssignment computedAssignment = assignor.assign(
groupSpec,
@ -327,9 +332,9 @@ public class RangeAssignorTest {
Collections.emptyMap()
));
Map<String, MemberSubscriptionSpecImpl> members = new TreeMap<>();
Map<String, MemberSubscriptionAndAssignmentImpl> members = new TreeMap<>();
members.put(memberA, new MemberSubscriptionSpecImpl(
members.put(memberA, new MemberSubscriptionAndAssignmentImpl(
Optional.empty(),
mkSet(topic1Uuid, topic2Uuid),
new Assignment(mkAssignment(
@ -338,7 +343,7 @@ public class RangeAssignorTest {
))
));
members.put(memberB, new MemberSubscriptionSpecImpl(
members.put(memberB, new MemberSubscriptionAndAssignmentImpl(
Optional.empty(),
mkSet(topic1Uuid, topic2Uuid),
new Assignment(mkAssignment(
@ -348,7 +353,7 @@ public class RangeAssignorTest {
));
// Add a new consumer to trigger a re-assignment
members.put(memberC, new MemberSubscriptionSpecImpl(
members.put(memberC, new MemberSubscriptionAndAssignmentImpl(
Optional.empty(),
mkSet(topic1Uuid, topic2Uuid),
Assignment.EMPTY
@ -359,7 +364,7 @@ public class RangeAssignorTest {
HOMOGENEOUS,
invertedTargetAssignment(members)
);
SubscribedTopicMetadata subscribedTopicMetadata = new SubscribedTopicMetadata(topicMetadata);
SubscribedTopicDescriberImpl subscribedTopicMetadata = new SubscribedTopicDescriberImpl(topicMetadata);
GroupAssignment computedAssignment = assignor.assign(
groupSpec,
@ -398,9 +403,9 @@ public class RangeAssignorTest {
Collections.emptyMap()
));
Map<String, MemberSubscriptionSpecImpl> members = new TreeMap<>();
Map<String, MemberSubscriptionAndAssignmentImpl> members = new TreeMap<>();
members.put(memberA, new MemberSubscriptionSpecImpl(
members.put(memberA, new MemberSubscriptionAndAssignmentImpl(
Optional.empty(),
mkSet(topic1Uuid, topic2Uuid),
new Assignment(mkAssignment(
@ -409,7 +414,7 @@ public class RangeAssignorTest {
))
));
members.put(memberB, new MemberSubscriptionSpecImpl(
members.put(memberB, new MemberSubscriptionAndAssignmentImpl(
Optional.empty(),
mkSet(topic1Uuid, topic2Uuid),
new Assignment(mkAssignment(
@ -423,7 +428,7 @@ public class RangeAssignorTest {
HOMOGENEOUS,
invertedTargetAssignment(members)
);
SubscribedTopicMetadata subscribedTopicMetadata = new SubscribedTopicMetadata(topicMetadata);
SubscribedTopicDescriberImpl subscribedTopicMetadata = new SubscribedTopicDescriberImpl(topicMetadata);
GroupAssignment computedAssignment = assignor.assign(
groupSpec,
@ -459,9 +464,9 @@ public class RangeAssignorTest {
Collections.emptyMap()
));
Map<String, MemberSubscriptionSpecImpl> members = new TreeMap<>();
Map<String, MemberSubscriptionAndAssignmentImpl> members = new TreeMap<>();
members.put(memberA, new MemberSubscriptionSpecImpl(
members.put(memberA, new MemberSubscriptionAndAssignmentImpl(
Optional.empty(),
mkSet(topic1Uuid, topic2Uuid),
new Assignment(mkAssignment(
@ -470,7 +475,7 @@ public class RangeAssignorTest {
))
));
members.put(memberB, new MemberSubscriptionSpecImpl(
members.put(memberB, new MemberSubscriptionAndAssignmentImpl(
Optional.empty(),
mkSet(topic1Uuid, topic2Uuid),
new Assignment(mkAssignment(
@ -480,7 +485,7 @@ public class RangeAssignorTest {
));
// Add a new consumer to trigger a re-assignment
members.put(memberC, new MemberSubscriptionSpecImpl(
members.put(memberC, new MemberSubscriptionAndAssignmentImpl(
Optional.empty(),
mkSet(topic1Uuid, topic2Uuid),
Assignment.EMPTY
@ -491,7 +496,7 @@ public class RangeAssignorTest {
HOMOGENEOUS,
invertedTargetAssignment(members)
);
SubscribedTopicMetadata subscribedTopicMetadata = new SubscribedTopicMetadata(topicMetadata);
SubscribedTopicDescriberImpl subscribedTopicMetadata = new SubscribedTopicDescriberImpl(topicMetadata);
GroupAssignment computedAssignment = assignor.assign(
groupSpec,
@ -532,9 +537,9 @@ public class RangeAssignorTest {
Collections.emptyMap()
));
Map<String, MemberSubscriptionSpecImpl> members = new TreeMap<>();
Map<String, MemberSubscriptionAndAssignmentImpl> members = new TreeMap<>();
members.put(memberA, new MemberSubscriptionSpecImpl(
members.put(memberA, new MemberSubscriptionAndAssignmentImpl(
Optional.empty(),
mkSet(topic1Uuid, topic2Uuid),
new Assignment(mkAssignment(
@ -543,7 +548,7 @@ public class RangeAssignorTest {
))
));
members.put(memberB, new MemberSubscriptionSpecImpl(
members.put(memberB, new MemberSubscriptionAndAssignmentImpl(
Optional.empty(),
mkSet(topic1Uuid, topic2Uuid),
new Assignment(mkAssignment(
@ -553,7 +558,7 @@ public class RangeAssignorTest {
));
// Add a new consumer to trigger a re-assignment
members.put(memberC, new MemberSubscriptionSpecImpl(
members.put(memberC, new MemberSubscriptionAndAssignmentImpl(
Optional.empty(),
mkSet(topic1Uuid),
Assignment.EMPTY
@ -564,7 +569,7 @@ public class RangeAssignorTest {
HETEROGENEOUS,
invertedTargetAssignment(members)
);
SubscribedTopicMetadata subscribedTopicMetadata = new SubscribedTopicMetadata(topicMetadata);
SubscribedTopicDescriberImpl subscribedTopicMetadata = new SubscribedTopicDescriberImpl(topicMetadata);
GroupAssignment computedAssignment = assignor.assign(
groupSpec,
@ -603,11 +608,11 @@ public class RangeAssignorTest {
Collections.emptyMap()
));
Map<String, MemberSubscriptionSpecImpl> members = new TreeMap<>();
Map<String, MemberSubscriptionAndAssignmentImpl> members = new TreeMap<>();
// Consumer A was removed
members.put(memberB, new MemberSubscriptionSpecImpl(
members.put(memberB, new MemberSubscriptionAndAssignmentImpl(
Optional.empty(),
mkSet(topic1Uuid, topic2Uuid),
new Assignment(mkAssignment(
@ -621,7 +626,7 @@ public class RangeAssignorTest {
HOMOGENEOUS,
invertedTargetAssignment(members)
);
SubscribedTopicMetadata subscribedTopicMetadata = new SubscribedTopicMetadata(topicMetadata);
SubscribedTopicDescriberImpl subscribedTopicMetadata = new SubscribedTopicDescriberImpl(topicMetadata);
GroupAssignment computedAssignment = assignor.assign(
groupSpec,
@ -661,9 +666,9 @@ public class RangeAssignorTest {
// Let initial subscriptions be A -> T1, T2 // B -> T2 // C -> T2, T3
// Change the subscriptions to A -> T1 // B -> T1, T2, T3 // C -> T2
Map<String, MemberSubscriptionSpecImpl> members = new TreeMap<>();
Map<String, MemberSubscriptionAndAssignmentImpl> members = new TreeMap<>();
members.put(memberA, new MemberSubscriptionSpecImpl(
members.put(memberA, new MemberSubscriptionAndAssignmentImpl(
Optional.empty(),
mkSet(topic1Uuid),
new Assignment(mkAssignment(
@ -672,7 +677,7 @@ public class RangeAssignorTest {
))
));
members.put(memberB, new MemberSubscriptionSpecImpl(
members.put(memberB, new MemberSubscriptionAndAssignmentImpl(
Optional.empty(),
mkSet(topic1Uuid, topic2Uuid, topic3Uuid),
new Assignment(mkAssignment(
@ -680,7 +685,7 @@ public class RangeAssignorTest {
))
));
members.put(memberC, new MemberSubscriptionSpecImpl(
members.put(memberC, new MemberSubscriptionAndAssignmentImpl(
Optional.empty(),
mkSet(topic2Uuid),
new Assignment(mkAssignment(
@ -694,7 +699,7 @@ public class RangeAssignorTest {
HETEROGENEOUS,
invertedTargetAssignment(members)
);
SubscribedTopicMetadata subscribedTopicMetadata = new SubscribedTopicMetadata(topicMetadata);
SubscribedTopicDescriberImpl subscribedTopicMetadata = new SubscribedTopicDescriberImpl(topicMetadata);
GroupAssignment computedAssignment = assignor.assign(
groupSpec,
@ -723,7 +728,7 @@ public class RangeAssignorTest {
) {
assertEquals(expectedAssignment.size(), computedGroupAssignment.members().size());
for (String memberId : computedGroupAssignment.members().keySet()) {
Map<Uuid, Set<Integer>> computedAssignmentForMember = computedGroupAssignment.members().get(memberId).targetPartitions();
Map<Uuid, Set<Integer>> computedAssignmentForMember = computedGroupAssignment.members().get(memberId).partitions();
assertEquals(expectedAssignment.get(memberId), computedAssignmentForMember);
}
}

4
group-coordinator/src/test/java/org/apache/kafka/coordinator/group/consumer/ConsumerGroupTest.java
View File

@ -55,8 +55,8 @@ import static org.apache.kafka.common.utils.Utils.mkMap;
import static org.apache.kafka.coordinator.group.AssignmentTestUtil.mkAssignment;
import static org.apache.kafka.coordinator.group.AssignmentTestUtil.mkTopicAssignment;
import static org.apache.kafka.coordinator.group.CoordinatorRecordHelpersTest.mkMapOfPartitionRacks;
import static org.apache.kafka.coordinator.group.assignor.SubscriptionType.HETEROGENEOUS;
import static org.apache.kafka.coordinator.group.assignor.SubscriptionType.HOMOGENEOUS;
import static org.apache.kafka.coordinator.group.api.assignor.SubscriptionType.HETEROGENEOUS;
import static org.apache.kafka.coordinator.group.api.assignor.SubscriptionType.HOMOGENEOUS;
import static org.junit.jupiter.api.Assertions.assertDoesNotThrow;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertFalse;

10
group-coordinator/src/test/java/org/apache/kafka/coordinator/group/consumer/SubscribedTopicMetadataTest.java
View File

@ -33,7 +33,7 @@ import static org.junit.jupiter.api.Assertions.assertThrows;
public class SubscribedTopicMetadataTest {
private Map<Uuid, TopicMetadata> topicMetadataMap;
private SubscribedTopicMetadata subscribedTopicMetadata;
private SubscribedTopicDescriberImpl subscribedTopicMetadata;
@BeforeEach
public void setUp() {
@ -47,7 +47,7 @@ public class SubscribedTopicMetadataTest {
new TopicMetadata(topicId, topicName, 5, partitionRacks)
);
}
subscribedTopicMetadata = new SubscribedTopicMetadata(topicMetadataMap);
subscribedTopicMetadata = new SubscribedTopicDescriberImpl(topicMetadataMap);
}
@Test
@ -57,7 +57,7 @@ public class SubscribedTopicMetadataTest {
@Test
public void testTopicMetadataCannotBeNull() {
assertThrows(NullPointerException.class, () -> new SubscribedTopicMetadata(null));
assertThrows(NullPointerException.class, () -> new SubscribedTopicDescriberImpl(null));
}
@Test
@ -100,11 +100,11 @@ public class SubscribedTopicMetadataTest {
@Test
public void testEquals() {
assertEquals(new SubscribedTopicMetadata(topicMetadataMap), subscribedTopicMetadata);
assertEquals(new SubscribedTopicDescriberImpl(topicMetadataMap), subscribedTopicMetadata);
Map<Uuid, TopicMetadata> topicMetadataMap2 = new HashMap<>();
Uuid topicId = Uuid.randomUuid();
topicMetadataMap2.put(topicId, new TopicMetadata(topicId, "newTopic", 5, Collections.emptyMap()));
assertNotEquals(new SubscribedTopicMetadata(topicMetadataMap2), subscribedTopicMetadata);
assertNotEquals(new SubscribedTopicDescriberImpl(topicMetadataMap2), subscribedTopicMetadata);
}
}

66
group-coordinator/src/test/java/org/apache/kafka/coordinator/group/consumer/TargetAssignmentBuilderTest.java
View File

@ -19,13 +19,11 @@ package org.apache.kafka.coordinator.group.consumer;
import org.apache.kafka.common.Uuid;
import org.apache.kafka.coordinator.group.AssignmentTestUtil;
import org.apache.kafka.coordinator.group.MetadataImageBuilder;
import org.apache.kafka.coordinator.group.assignor.GroupSpecImpl;
import org.apache.kafka.coordinator.group.assignor.MemberSubscriptionSpec;
import org.apache.kafka.coordinator.group.assignor.MemberSubscriptionSpecImpl;
import org.apache.kafka.coordinator.group.assignor.SubscriptionType;
import org.apache.kafka.coordinator.group.assignor.GroupAssignment;
import org.apache.kafka.coordinator.group.assignor.MemberAssignment;
import org.apache.kafka.coordinator.group.assignor.PartitionAssignor;
import org.apache.kafka.coordinator.group.api.assignor.MemberSubscription;
import org.apache.kafka.coordinator.group.api.assignor.SubscriptionType;
import org.apache.kafka.coordinator.group.api.assignor.GroupAssignment;
import org.apache.kafka.coordinator.group.api.assignor.MemberAssignment;
import org.apache.kafka.coordinator.group.api.assignor.PartitionAssignor;
import org.apache.kafka.image.TopicsImage;
import org.junit.jupiter.api.Test;
@ -44,8 +42,8 @@ import static org.apache.kafka.coordinator.group.AssignmentTestUtil.mkTopicAssig
import static org.apache.kafka.coordinator.group.CoordinatorRecordHelpers.newTargetAssignmentEpochRecord;
import static org.apache.kafka.coordinator.group.CoordinatorRecordHelpers.newTargetAssignmentRecord;
import static org.apache.kafka.coordinator.group.CoordinatorRecordHelpersTest.mkMapOfPartitionRacks;
import static org.apache.kafka.coordinator.group.assignor.SubscriptionType.HOMOGENEOUS;
import static org.apache.kafka.coordinator.group.consumer.TargetAssignmentBuilder.createMemberSubscriptionSpecImpl;
import static org.apache.kafka.coordinator.group.api.assignor.SubscriptionType.HOMOGENEOUS;
import static org.apache.kafka.coordinator.group.consumer.TargetAssignmentBuilder.createMemberSubscriptionAndAssignment;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.mockito.ArgumentMatchers.any;
import static org.mockito.Mockito.mock;
@ -159,17 +157,17 @@ public class TargetAssignmentBuilderTest {
String memberId,
Map<Uuid, Set<Integer>> assignment
) {
memberAssignments.put(memberId, new MemberAssignment(assignment));
memberAssignments.put(memberId, new MemberAssignmentImpl(assignment));
}
public TargetAssignmentBuilder.TargetAssignmentResult build() {
TopicsImage topicsImage = topicsImageBuilder.build().topics();
// Prepare expected member specs.
Map<String, MemberSubscriptionSpecImpl> memberSubscriptions = new HashMap<>();
Map<String, MemberSubscriptionAndAssignmentImpl> memberSubscriptions = new HashMap<>();
// All the existing members are prepared.
members.forEach((memberId, member) ->
memberSubscriptions.put(memberId, createMemberSubscriptionSpecImpl(
memberSubscriptions.put(memberId, createMemberSubscriptionAndAssignment(
member,
targetAssignment.getOrDefault(memberId, Assignment.EMPTY),
topicsImage
@ -192,7 +190,7 @@ public class TargetAssignmentBuilderTest {
}
}
memberSubscriptions.put(memberId, createMemberSubscriptionSpecImpl(
memberSubscriptions.put(memberId, createMemberSubscriptionAndAssignment(
updatedMemberOrNull,
assignment,
topicsImage
@ -206,7 +204,7 @@ public class TargetAssignmentBuilderTest {
topicMetadataMap.put(topicMetadata.id(), topicMetadata));
// Prepare the expected subscription topic metadata.
SubscribedTopicMetadata subscribedTopicMetadata = new SubscribedTopicMetadata(topicMetadataMap);
SubscribedTopicDescriberImpl subscribedTopicMetadata = new SubscribedTopicDescriberImpl(topicMetadataMap);
SubscriptionType subscriptionType = HOMOGENEOUS;
// Prepare the member assignments per topic partition.
@ -277,13 +275,13 @@ public class TargetAssignmentBuilderTest {
mkTopicAssignment(barTopicId, 1, 2, 3)
));
MemberSubscriptionSpec subscriptionSpec = createMemberSubscriptionSpecImpl(
MemberSubscription subscriptionSpec = createMemberSubscriptionAndAssignment(
member,
assignment,
topicsImage
);
assertEquals(new MemberSubscriptionSpecImpl(
assertEquals(new MemberSubscriptionAndAssignmentImpl(
Optional.of("rackId"),
new TopicIds(mkSet("bar", "foo", "zar"), topicsImage),
assignment
@ -343,11 +341,11 @@ public class TargetAssignmentBuilderTest {
)), result.records());
Map<String, MemberAssignment> expectedAssignment = new HashMap<>();
expectedAssignment.put("member-1", new MemberAssignment(mkAssignment(
expectedAssignment.put("member-1", new MemberAssignmentImpl(mkAssignment(
mkTopicAssignment(fooTopicId, 1, 2, 3),
mkTopicAssignment(barTopicId, 1, 2, 3)
)));
expectedAssignment.put("member-2", new MemberAssignment(mkAssignment(
expectedAssignment.put("member-2", new MemberAssignmentImpl(mkAssignment(
mkTopicAssignment(fooTopicId, 4, 5, 6),
mkTopicAssignment(barTopicId, 4, 5, 6)
)));
@ -406,11 +404,11 @@ public class TargetAssignmentBuilderTest {
), result.records().get(2));
Map<String, MemberAssignment> expectedAssignment = new HashMap<>();
expectedAssignment.put("member-2", new MemberAssignment(mkAssignment(
expectedAssignment.put("member-2", new MemberAssignmentImpl(mkAssignment(
mkTopicAssignment(fooTopicId, 1, 2, 3),
mkTopicAssignment(barTopicId, 1, 2, 3)
)));
expectedAssignment.put("member-1", new MemberAssignment(mkAssignment(
expectedAssignment.put("member-1", new MemberAssignmentImpl(mkAssignment(
mkTopicAssignment(fooTopicId, 4, 5, 6),
mkTopicAssignment(barTopicId, 4, 5, 6)
)));
@ -480,15 +478,15 @@ public class TargetAssignmentBuilderTest {
), result.records().get(3));
Map<String, MemberAssignment> expectedAssignment = new HashMap<>();
expectedAssignment.put("member-1", new MemberAssignment(mkAssignment(
expectedAssignment.put("member-1", new MemberAssignmentImpl(mkAssignment(
mkTopicAssignment(fooTopicId, 1, 2),
mkTopicAssignment(barTopicId, 1, 2)
)));
expectedAssignment.put("member-2", new MemberAssignment(mkAssignment(
expectedAssignment.put("member-2", new MemberAssignmentImpl(mkAssignment(
mkTopicAssignment(fooTopicId, 3, 4),
mkTopicAssignment(barTopicId, 3, 4)
)));
expectedAssignment.put("member-3", new MemberAssignment(mkAssignment(
expectedAssignment.put("member-3", new MemberAssignmentImpl(mkAssignment(
mkTopicAssignment(fooTopicId, 5, 6),
mkTopicAssignment(barTopicId, 5, 6)
)));
@ -567,15 +565,15 @@ public class TargetAssignmentBuilderTest {
), result.records().get(3));
Map<String, MemberAssignment> expectedAssignment = new HashMap<>();
expectedAssignment.put("member-1", new MemberAssignment(mkAssignment(
expectedAssignment.put("member-1", new MemberAssignmentImpl(mkAssignment(
mkTopicAssignment(fooTopicId, 1, 2),
mkTopicAssignment(barTopicId, 1, 2)
)));
expectedAssignment.put("member-2", new MemberAssignment(mkAssignment(
expectedAssignment.put("member-2", new MemberAssignmentImpl(mkAssignment(
mkTopicAssignment(fooTopicId, 3, 4),
mkTopicAssignment(barTopicId, 3, 4)
)));
expectedAssignment.put("member-3", new MemberAssignment(mkAssignment(
expectedAssignment.put("member-3", new MemberAssignmentImpl(mkAssignment(
mkTopicAssignment(fooTopicId, 5, 6),
mkTopicAssignment(barTopicId, 5, 6)
)));
@ -645,15 +643,15 @@ public class TargetAssignmentBuilderTest {
), result.records().get(2));
Map<String, MemberAssignment> expectedAssignment = new HashMap<>();
expectedAssignment.put("member-1", new MemberAssignment(mkAssignment(
expectedAssignment.put("member-1", new MemberAssignmentImpl(mkAssignment(
mkTopicAssignment(fooTopicId, 1, 2),
mkTopicAssignment(barTopicId, 1, 2)
)));
expectedAssignment.put("member-2", new MemberAssignment(mkAssignment(
expectedAssignment.put("member-2", new MemberAssignmentImpl(mkAssignment(
mkTopicAssignment(fooTopicId, 3, 4, 5),
mkTopicAssignment(barTopicId, 3, 4, 5)
)));
expectedAssignment.put("member-3", new MemberAssignment(mkAssignment(
expectedAssignment.put("member-3", new MemberAssignmentImpl(mkAssignment(
mkTopicAssignment(fooTopicId, 6),
mkTopicAssignment(barTopicId, 6)
)));
@ -719,11 +717,11 @@ public class TargetAssignmentBuilderTest {
), result.records().get(2));
Map<String, MemberAssignment> expectedAssignment = new HashMap<>();
expectedAssignment.put("member-1", new MemberAssignment(mkAssignment(
expectedAssignment.put("member-1", new MemberAssignmentImpl(mkAssignment(
mkTopicAssignment(fooTopicId, 1, 2, 3),
mkTopicAssignment(barTopicId, 1, 2, 3)
)));
expectedAssignment.put("member-2", new MemberAssignment(mkAssignment(
expectedAssignment.put("member-2", new MemberAssignmentImpl(mkAssignment(
mkTopicAssignment(fooTopicId, 4, 5, 6),
mkTopicAssignment(barTopicId, 4, 5, 6)
)));
@ -794,16 +792,16 @@ public class TargetAssignmentBuilderTest {
), result.records().get(1));
Map<String, MemberAssignment> expectedAssignment = new HashMap<>();
expectedAssignment.put("member-1", new MemberAssignment(mkAssignment(
expectedAssignment.put("member-1", new MemberAssignmentImpl(mkAssignment(
mkTopicAssignment(fooTopicId, 1, 2),
mkTopicAssignment(barTopicId, 1, 2)
)));
expectedAssignment.put("member-2", new MemberAssignment(mkAssignment(
expectedAssignment.put("member-2", new MemberAssignmentImpl(mkAssignment(
mkTopicAssignment(fooTopicId, 3, 4),
mkTopicAssignment(barTopicId, 3, 4)
)));
expectedAssignment.put("member-3-a", new MemberAssignment(mkAssignment(
expectedAssignment.put("member-3-a", new MemberAssignmentImpl(mkAssignment(
mkTopicAssignment(fooTopicId, 5, 6),
mkTopicAssignment(barTopicId, 5, 6)
)));

6
jmh-benchmarks/src/main/java/org/apache/kafka/jmh/assignor/AssignorBenchmarkUtils.java
View File

@ -19,8 +19,8 @@ package org.apache.kafka.jmh.assignor;
import org.apache.kafka.common.Uuid;
import org.apache.kafka.common.metadata.PartitionRecord;
import org.apache.kafka.common.metadata.TopicRecord;
import org.apache.kafka.coordinator.group.assignor.GroupAssignment;
import org.apache.kafka.coordinator.group.assignor.MemberAssignment;
import org.apache.kafka.coordinator.group.api.assignor.GroupAssignment;
import org.apache.kafka.coordinator.group.api.assignor.MemberAssignment;
import org.apache.kafka.image.MetadataDelta;
import java.util.Arrays;
@ -41,7 +41,7 @@ public class AssignorBenchmarkUtils {
Map<Uuid, Map<Integer, String>> invertedTargetAssignment = new HashMap<>();
for (Map.Entry<String, MemberAssignment> memberEntry : groupAssignment.members().entrySet()) {
String memberId = memberEntry.getKey();
Map<Uuid, Set<Integer>> topicsAndPartitions = memberEntry.getValue().targetPartitions();
Map<Uuid, Set<Integer>> topicsAndPartitions = memberEntry.getValue().partitions();
for (Map.Entry<Uuid, Set<Integer>> topicEntry : topicsAndPartitions.entrySet()) {
Uuid topicId = topicEntry.getKey();

6
jmh-benchmarks/src/main/java/org/apache/kafka/jmh/assignor/ClientSideAssignorBenchmark.java
View File

@ -23,7 +23,7 @@ import org.apache.kafka.common.Cluster;
import org.apache.kafka.common.Node;
import org.apache.kafka.common.PartitionInfo;
import org.apache.kafka.common.TopicPartition;
import org.apache.kafka.coordinator.group.assignor.SubscriptionType;
import org.apache.kafka.coordinator.group.api.assignor.SubscriptionType;
import org.openjdk.jmh.annotations.Benchmark;
import org.openjdk.jmh.annotations.BenchmarkMode;
import org.openjdk.jmh.annotations.Fork;
@ -47,8 +47,8 @@ import java.util.Optional;
import java.util.concurrent.TimeUnit;
import static org.apache.kafka.clients.consumer.internals.AbstractStickyAssignor.DEFAULT_GENERATION;
import static org.apache.kafka.coordinator.group.assignor.SubscriptionType.HETEROGENEOUS;
import static org.apache.kafka.coordinator.group.assignor.SubscriptionType.HOMOGENEOUS;
import static org.apache.kafka.coordinator.group.api.assignor.SubscriptionType.HETEROGENEOUS;
import static org.apache.kafka.coordinator.group.api.assignor.SubscriptionType.HOMOGENEOUS;
@State(Scope.Benchmark)
@Fork(value = 1)

39
jmh-benchmarks/src/main/java/org/apache/kafka/jmh/assignor/ServerSideAssignorBenchmark.java
View File

@ -17,18 +17,19 @@
package org.apache.kafka.jmh.assignor;
import org.apache.kafka.common.Uuid;
import org.apache.kafka.coordinator.group.assignor.GroupSpecImpl;
import org.apache.kafka.coordinator.group.assignor.GroupAssignment;
import org.apache.kafka.coordinator.group.assignor.MemberAssignment;
import org.apache.kafka.coordinator.group.assignor.MemberSubscriptionSpecImpl;
import org.apache.kafka.coordinator.group.assignor.PartitionAssignor;
import org.apache.kafka.coordinator.group.consumer.MemberAssignmentImpl;
import org.apache.kafka.coordinator.group.consumer.GroupSpecImpl;
import org.apache.kafka.coordinator.group.api.assignor.GroupAssignment;
import org.apache.kafka.coordinator.group.api.assignor.MemberAssignment;
import org.apache.kafka.coordinator.group.consumer.MemberSubscriptionAndAssignmentImpl;
import org.apache.kafka.coordinator.group.api.assignor.PartitionAssignor;
import org.apache.kafka.coordinator.group.assignor.RangeAssignor;
import org.apache.kafka.coordinator.group.assignor.SubscribedTopicDescriber;
import org.apache.kafka.coordinator.group.assignor.SubscriptionType;
import org.apache.kafka.coordinator.group.api.assignor.SubscribedTopicDescriber;
import org.apache.kafka.coordinator.group.api.assignor.SubscriptionType;
import org.apache.kafka.coordinator.group.consumer.Assignment;
import org.apache.kafka.coordinator.group.consumer.TopicIds;
import org.apache.kafka.coordinator.group.assignor.UniformAssignor;
import org.apache.kafka.coordinator.group.consumer.SubscribedTopicMetadata;
import org.apache.kafka.coordinator.group.consumer.SubscribedTopicDescriberImpl;
import org.apache.kafka.coordinator.group.consumer.TopicMetadata;
import org.apache.kafka.image.MetadataDelta;
@ -61,8 +62,8 @@ import java.util.Optional;
import java.util.Set;
import java.util.concurrent.TimeUnit;
import static org.apache.kafka.coordinator.group.assignor.SubscriptionType.HETEROGENEOUS;
import static org.apache.kafka.coordinator.group.assignor.SubscriptionType.HOMOGENEOUS;
import static org.apache.kafka.coordinator.group.api.assignor.SubscriptionType.HETEROGENEOUS;
import static org.apache.kafka.coordinator.group.api.assignor.SubscriptionType.HOMOGENEOUS;
@State(Scope.Benchmark)
@Fork(value = 1)
@ -133,7 +134,7 @@ public class ServerSideAssignorBenchmark {
@Setup(Level.Trial)
public void setup() {
Map<Uuid, TopicMetadata> topicMetadata = createTopicMetadata();
subscribedTopicDescriber = new SubscribedTopicMetadata(topicMetadata);
subscribedTopicDescriber = new SubscribedTopicDescriberImpl(topicMetadata);
createGroupSpec();
@ -177,7 +178,7 @@ public class ServerSideAssignorBenchmark {
}
private void createGroupSpec() {
Map<String, MemberSubscriptionSpecImpl> members = new HashMap<>();
Map<String, MemberSubscriptionAndAssignmentImpl> members = new HashMap<>();
// In the rebalance case, we will add the last member as a trigger.
// This is done to keep the total members count consistent with the input.
@ -228,14 +229,14 @@ public class ServerSideAssignorBenchmark {
}
private void addMemberSpec(
Map<String, MemberSubscriptionSpecImpl> members,
Map<String, MemberSubscriptionAndAssignmentImpl> members,
int memberIndex,
Set<Uuid> subscribedTopicIds
) {
String memberId = "member" + memberIndex;
Optional<String> rackId = rackId(memberIndex);
members.put(memberId, new MemberSubscriptionSpecImpl(
members.put(memberId, new MemberSubscriptionAndAssignmentImpl(
rackId,
subscribedTopicIds,
Assignment.EMPTY
@ -260,18 +261,18 @@ public class ServerSideAssignorBenchmark {
Map<Uuid, Map<Integer, String>> invertedTargetAssignment = AssignorBenchmarkUtils.computeInvertedTargetAssignment(initialAssignment);
Map<String, MemberSubscriptionSpecImpl> updatedMemberSpec = new HashMap<>();
Map<String, MemberSubscriptionAndAssignmentImpl> updatedMemberSpec = new HashMap<>();
for (String memberId : groupSpec.memberIds()) {
MemberAssignment memberAssignment = members.getOrDefault(
memberId,
new MemberAssignment(Collections.emptyMap())
new MemberAssignmentImpl(Collections.emptyMap())
);
updatedMemberSpec.put(memberId, new MemberSubscriptionSpecImpl(
updatedMemberSpec.put(memberId, new MemberSubscriptionAndAssignmentImpl(
groupSpec.memberSubscription(memberId).rackId(),
groupSpec.memberSubscription(memberId).subscribedTopicIds(),
new Assignment(Collections.unmodifiableMap(memberAssignment.targetPartitions()))
new Assignment(Collections.unmodifiableMap(memberAssignment.partitions()))
));
}
@ -283,7 +284,7 @@ public class ServerSideAssignorBenchmark {
}
Optional<String> rackId = rackId(memberCount - 1);
updatedMemberSpec.put("newMember", new MemberSubscriptionSpecImpl(
updatedMemberSpec.put("newMember", new MemberSubscriptionAndAssignmentImpl(
rackId,
subscribedTopicIdsForNewMember,
Assignment.EMPTY

22
jmh-benchmarks/src/main/java/org/apache/kafka/jmh/assignor/TargetAssignmentBuilderBenchmark.java
View File

@ -17,16 +17,16 @@
package org.apache.kafka.jmh.assignor;
import org.apache.kafka.common.Uuid;
import org.apache.kafka.coordinator.group.assignor.GroupSpecImpl;
import org.apache.kafka.coordinator.group.assignor.GroupAssignment;
import org.apache.kafka.coordinator.group.assignor.MemberAssignment;
import org.apache.kafka.coordinator.group.assignor.MemberSubscriptionSpecImpl;
import org.apache.kafka.coordinator.group.assignor.PartitionAssignor;
import org.apache.kafka.coordinator.group.consumer.GroupSpecImpl;
import org.apache.kafka.coordinator.group.api.assignor.GroupAssignment;
import org.apache.kafka.coordinator.group.api.assignor.MemberAssignment;
import org.apache.kafka.coordinator.group.consumer.MemberSubscriptionAndAssignmentImpl;
import org.apache.kafka.coordinator.group.api.assignor.PartitionAssignor;
import org.apache.kafka.coordinator.group.consumer.TopicIds;
import org.apache.kafka.coordinator.group.assignor.UniformAssignor;
import org.apache.kafka.coordinator.group.consumer.Assignment;
import org.apache.kafka.coordinator.group.consumer.ConsumerGroupMember;
import org.apache.kafka.coordinator.group.consumer.SubscribedTopicMetadata;
import org.apache.kafka.coordinator.group.consumer.SubscribedTopicDescriberImpl;
import org.apache.kafka.coordinator.group.consumer.TargetAssignmentBuilder;
import org.apache.kafka.coordinator.group.consumer.TopicMetadata;
@ -59,7 +59,7 @@ import java.util.Optional;
import java.util.Set;
import java.util.concurrent.TimeUnit;
import static org.apache.kafka.coordinator.group.assignor.SubscriptionType.HOMOGENEOUS;
import static org.apache.kafka.coordinator.group.api.assignor.SubscriptionType.HOMOGENEOUS;
@State(Scope.Benchmark)
@Fork(value = 1)
@ -175,7 +175,7 @@ public class TargetAssignmentBuilderBenchmark {
GroupAssignment groupAssignment = partitionAssignor.assign(
groupSpec,
new SubscribedTopicMetadata(topicMetadataMap)
new SubscribedTopicDescriberImpl(topicMetadataMap)
);
invertedTargetAssignment = AssignorBenchmarkUtils.computeInvertedTargetAssignment(groupAssignment);
@ -183,7 +183,7 @@ public class TargetAssignmentBuilderBenchmark {
for (Map.Entry<String, MemberAssignment> entry : groupAssignment.members().entrySet()) {
String memberId = entry.getKey();
Map<Uuid, Set<Integer>> topicPartitions = entry.getValue().targetPartitions();
Map<Uuid, Set<Integer>> topicPartitions = entry.getValue().partitions();
initialTargetAssignment.put(memberId, new Assignment(topicPartitions));
}
@ -191,12 +191,12 @@ public class TargetAssignmentBuilderBenchmark {
}
private void createAssignmentSpec() {
Map<String, MemberSubscriptionSpecImpl> members = new HashMap<>();
Map<String, MemberSubscriptionAndAssignmentImpl> members = new HashMap<>();
for (int i = 0; i < memberCount - 1; i++) {
String memberId = "member" + i;
members.put(memberId, new MemberSubscriptionSpecImpl(
members.put(memberId, new MemberSubscriptionAndAssignmentImpl(
Optional.empty(),
new TopicIds(new HashSet<>(allTopicNames), topicsImage),
Assignment.EMPTY

1
settings.gradle
View File

@ -61,6 +61,7 @@ include 'clients',
'examples',
'generator',
'group-coordinator',
'group-coordinator:group-coordinator-api',
'jmh-benchmarks',
'log4j-appender',
'metadata',