Skip to content

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

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
dajac merged 11 commits into from
Jun 6, 2024
Merged

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

Show file tree
Hide file tree
Changes from 4 commits
Commits
Show all changes
11 commits
Select commit Hold shift + click to select a range
7dd66cc
add new module
dajac Jun 4, 2024
751a1b6
introduce api package
dajac Jun 4, 2024
40be636
rename module
dajac Jun 4, 2024
0bc37f1
minor
dajac Jun 4, 2024
e4e776e
add unstable to all interfaces/classes
dajac Jun 5, 2024
ba759aa
fix deps
dajac Jun 5, 2024
9999430
change interface to address comment
dajac Jun 5, 2024
445eecb
fix checkstyle
dajac Jun 5, 2024
f1e3d32
fix
dajac Jun 5, 2024
cf884b9
address minor comments
dajac Jun 6, 2024
4479672
address minor comment
dajac Jun 6, 2024
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
62 changes: 62 additions & 0 deletions build.gradle
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1345,6 +1345,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"
Expand All @@ -1358,6 +1418,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
Expand Down Expand Up @@ -2902,6 +2963,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')
Expand Down
2 changes: 1 addition & 1 deletion core/src/main/scala/kafka/server/KafkaConfig.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down
2 changes: 1 addition & 1 deletion ...ignor/ConsumerGroupPartitionAssignor.java → ...ignor/ConsumerGroupPartitionAssignor.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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.annotation.InterfaceStability;

Expand Down
2 changes: 1 addition & 1 deletion ...nator/group/assignor/GroupAssignment.java → ...r/group/api/assignor/GroupAssignment.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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 java.util.Map;
import java.util.Objects;
Expand Down
2 changes: 1 addition & 1 deletion ...coordinator/group/assignor/GroupSpec.java → ...dinator/group/api/assignor/GroupSpec.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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;

@chia7712 chia7712 Jun 4, 2024

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I noticed that not all interfaces have @InterfaceStability.Unstable. Could you share the context to me?

@chia7712 chia7712 Jun 4, 2024

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

For another, GroupSpec#MemberSubscriptionSpec return the interface but GroupSpec#memberAssignment return a map struct. If it is public API now, maybe return MemberAssignment is more flexible if we want to enrich it in the future?

@dajac dajac Jun 5, 2024

Copy link
Copy Markdown
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I noticed that not all interfaces have @InterfaceStability.Unstable. Could you share the context to me?

Fixed. Thanks.

For another, GroupSpec#MemberSubscriptionSpec return the interface but GroupSpec#memberAssignment return a map struct. If it is public API now, maybe return MemberAssignment is more flexible if we want to enrich it in the future?

Good question. Reusing MemberAssignment is not ideal because we want to have the ability to pass internal objects. In this case, it could be backed by the Assignment. So, we could think of using an interface though.

On thing to consider is whether the MemberAssignment returned by the assignor would always be the same as the one provided to the assignor. The actual assignment (the map) is for sure. What do you think?

@chia7712 chia7712 Jun 5, 2024

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

So, we could think of using an interface though.

+1

On thing to consider is whether the MemberAssignment returned by the assignor would always be the same as the one provided to the assignor. The actual assignment (the map) is for sure. What do you think?

not sure whether I have got the point. The member assignment in GroupAssignment returned from PartitionAssignor#assign is the new assignment. Hence, it should be different to assignment in GroupSpec.

@dajac dajac Jun 5, 2024

Copy link
Copy Markdown
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

You suggested to use MemberAssignment in your previous comment. MemberAssignment is already used in GroupAssignment returned by PartitionAssignor#assign. I am not sure if you meant to reuse the same MemberAssignment in both places.

@chia7712 chia7712 Jun 5, 2024

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

You suggested to use MemberAssignment in your previous comment. MemberAssignment is already used in GroupAssignment returned by PartitionAssignor#assign. I am not sure if you meant to reuse the same MemberAssignment in both places.

ummm, I guess I did not notice MemberAssignment is used by GroupAssignment before :_

Anyway, having a interface to replace Map struct is good way to me. Or we can keep current version if the tag "unstable" give the room to modify those public stuff :)

@dajac dajac Jun 5, 2024

Copy link
Copy Markdown
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I am playing with this. I will suggest something a bit later today.

@dajac dajac Jun 5, 2024

Copy link
Copy Markdown
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

@chia7712 I gave it a go. You can check the last commit to see how it looks like. I also moved some classes between internal packages while here.


Expand Down
2 changes: 1 addition & 1 deletion ...ator/group/assignor/MemberAssignment.java → .../group/api/assignor/MemberAssignment.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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;

Expand Down
2 changes: 1 addition & 1 deletion ...roup/assignor/MemberSubscriptionSpec.java → .../api/assignor/MemberSubscriptionSpec.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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;

Expand Down
2 changes: 1 addition & 1 deletion ...tor/group/assignor/PartitionAssignor.java → ...group/api/assignor/PartitionAssignor.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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.annotation.InterfaceStability;

Expand Down
2 changes: 1 addition & 1 deletion .../assignor/PartitionAssignorException.java → .../assignor/PartitionAssignorException.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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;

Expand Down
2 changes: 1 addition & 1 deletion ...up/assignor/SubscribedTopicDescriber.java → ...pi/assignor/SubscribedTopicDescriber.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand Down
2 changes: 1 addition & 1 deletion ...ator/group/assignor/SubscriptionType.java → .../group/api/assignor/SubscriptionType.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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;

/**
* The subscription type followed by a consumer group.
Expand Down
2 changes: 1 addition & 1 deletion ...-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupCoordinatorConfig.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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;

Expand Down
8 changes: 4 additions & 4 deletions group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupMetadataManager.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -58,10 +58,10 @@
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;
Expand Down
3 changes: 3 additions & 0 deletions ...in/java/org/apache/kafka/coordinator/group/assignor/AbstractUniformAssignmentBuilder.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand Down
5 changes: 5 additions & 0 deletions ...ain/java/org/apache/kafka/coordinator/group/assignor/GeneralUniformAssignmentBuilder.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -17,6 +17,11 @@
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.server.common.TopicIdPartition;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
Expand Down
3 changes: 3 additions & 0 deletions ...-coordinator/src/main/java/org/apache/kafka/coordinator/group/assignor/GroupSpecImpl.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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.GroupSpec;
import org.apache.kafka.coordinator.group.api.assignor.MemberSubscriptionSpec;
import org.apache.kafka.coordinator.group.api.assignor.SubscriptionType;

import java.util.Collection;
import java.util.Collections;
Expand Down
1 change: 1 addition & 0 deletions ...src/main/java/org/apache/kafka/coordinator/group/assignor/MemberSubscriptionSpecImpl.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -17,6 +17,7 @@
package org.apache.kafka.coordinator.group.assignor;

import org.apache.kafka.common.Uuid;
import org.apache.kafka.coordinator.group.api.assignor.MemberSubscriptionSpec;
import org.apache.kafka.coordinator.group.consumer.Assignment;

import java.util.Map;
Expand Down
5 changes: 5 additions & 0 deletions ...n/java/org/apache/kafka/coordinator/group/assignor/OptimizedUniformAssignmentBuilder.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -17,6 +17,11 @@
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.server.common.TopicIdPartition;

import java.util.ArrayList;
Expand Down
8 changes: 7 additions & 1 deletion ...-coordinator/src/main/java/org/apache/kafka/coordinator/group/assignor/RangeAssignor.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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.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 java.util.ArrayList;
import java.util.Collection;
Expand All @@ -28,7 +34,7 @@
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.
Expand Down
7 changes: 6 additions & 1 deletion ...oordinator/src/main/java/org/apache/kafka/coordinator/group/assignor/UniformAssignor.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down
6 changes: 3 additions & 3 deletions ...-coordinator/src/main/java/org/apache/kafka/coordinator/group/consumer/ConsumerGroup.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -31,7 +31,7 @@
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;
Expand All @@ -54,8 +54,8 @@
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;
Expand Down
4 changes: 2 additions & 2 deletions ...or/src/main/java/org/apache/kafka/coordinator/group/consumer/SubscribedTopicMetadata.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand Down
10 changes: 5 additions & 5 deletions ...or/src/main/java/org/apache/kafka/coordinator/group/consumer/TargetAssignmentBuilder.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -20,11 +20,11 @@
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;
Expand Down
2 changes: 1 addition & 1 deletion group-coordinator/src/test/java/org/apache/kafka/coordinator/group/AssignmentTestUtil.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -17,7 +17,7 @@
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.api.assignor.GroupAssignment;
import org.apache.kafka.coordinator.group.assignor.MemberSubscriptionSpecImpl;

import java.util.AbstractMap;
Expand Down
2 changes: 1 addition & 1 deletion ...rdinator/src/test/java/org/apache/kafka/coordinator/group/GroupCoordinatorConfigTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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;

Expand Down
8 changes: 4 additions & 4 deletions ...oordinator/src/test/java/org/apache/kafka/coordinator/group/GroupMetadataManagerTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -61,10 +61,10 @@
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.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;
Expand Down
2 changes: 1 addition & 1 deletion ...tor/src/test/java/org/apache/kafka/coordinator/group/GroupMetadataManagerTestContext.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -46,7 +46,7 @@
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;
Expand Down
10 changes: 5 additions & 5 deletions ...p-coordinator/src/test/java/org/apache/kafka/coordinator/group/MockPartitionAssignor.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand Down
Loading