Skip to content
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

Jump to bottom

KAFKA-17047: Refactored group coordinator classes to modern package (KIP-932) #16474

Merged
merged 5 commits into from
Jul 3, 2024
Merged

KAFKA-17047: Refactored group coordinator classes to modern package (KIP-932) #16474

Show file tree
Hide file tree
Changes from 1 commit
Commits
Show all changes
5 commits
Select commit Hold shift + click to select a range
01938ea
KAFKA-17047: Refactored group coordinator classes to modern package (…
apoorvmittal10 Jun 27, 2024
a82ee40
Merge remote-tracking branch 'upstream/trunk' into KAFKA-17047
apoorvmittal10 Jul 1, 2024
737bc47
Merge remote-tracking branch 'upstream/trunk' into KAFKA-17047
apoorvmittal10 Jul 2, 2024
81f73b3
Addressing review comments
apoorvmittal10 Jul 2, 2024
e005d8b
Merge remote-tracking branch 'upstream/trunk' into KAFKA-17047
apoorvmittal10 Jul 2, 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
2 changes: 1 addition & 1 deletion core/src/test/scala/unit/kafka/server/ConsumerProtocolMigrationTest.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -23,7 +23,7 @@ import org.apache.kafka.common.message.ListGroupsResponseData
import org.apache.kafka.common.protocol.{ApiKeys, Errors}
import org.apache.kafka.coordinator.group.Group
import org.apache.kafka.coordinator.group.classic.ClassicGroupState
import org.apache.kafka.coordinator.group.consumer.ConsumerGroup.ConsumerGroupState
import org.apache.kafka.coordinator.group.modern.consumer.ConsumerGroup.ConsumerGroupState
import org.apache.kafka.server.common.Features
import org.junit.jupiter.api.Assertions.assertEquals
import org.junit.jupiter.api.Tag
Expand Down
2 changes: 1 addition & 1 deletion core/src/test/scala/unit/kafka/server/ListGroupsRequestTest.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -21,8 +21,8 @@ import kafka.test.annotation.{ClusterConfigProperty, ClusterFeature, ClusterTest
import kafka.test.junit.ClusterTestExtensions
import org.apache.kafka.common.message.ListGroupsResponseData
import org.apache.kafka.common.protocol.{ApiKeys, Errors}
import org.apache.kafka.coordinator.group.consumer.ConsumerGroup.ConsumerGroupState
import org.apache.kafka.coordinator.group.classic.ClassicGroupState
import org.apache.kafka.coordinator.group.modern.consumer.ConsumerGroup.ConsumerGroupState
import org.apache.kafka.coordinator.group.Group
import org.apache.kafka.server.common.Features
import org.junit.jupiter.api.Assertions.{assertEquals, fail}
Expand Down
4 changes: 2 additions & 2 deletions ...oordinator/src/main/java/org/apache/kafka/coordinator/group/CoordinatorRecordHelpers.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -20,8 +20,6 @@
import org.apache.kafka.common.record.RecordBatch;
import org.apache.kafka.common.requests.OffsetCommitRequest;
import org.apache.kafka.coordinator.group.classic.ClassicGroup;
import org.apache.kafka.coordinator.group.consumer.ConsumerGroupMember;
import org.apache.kafka.coordinator.group.consumer.TopicMetadata;
import org.apache.kafka.coordinator.group.generated.ConsumerGroupCurrentMemberAssignmentKey;
import org.apache.kafka.coordinator.group.generated.ConsumerGroupCurrentMemberAssignmentValue;
import org.apache.kafka.coordinator.group.generated.ConsumerGroupMemberMetadataKey;
Expand All @@ -38,6 +36,8 @@
import org.apache.kafka.coordinator.group.generated.GroupMetadataValue;
import org.apache.kafka.coordinator.group.generated.OffsetCommitKey;
import org.apache.kafka.coordinator.group.generated.OffsetCommitValue;
import org.apache.kafka.coordinator.group.modern.TopicMetadata;
import org.apache.kafka.coordinator.group.modern.consumer.ConsumerGroupMember;
import org.apache.kafka.server.common.ApiMessageAndVersion;
import org.apache.kafka.server.common.MetadataVersion;

Expand Down
16 changes: 8 additions & 8 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 @@ -64,12 +64,6 @@
import org.apache.kafka.coordinator.group.classic.ClassicGroup;
import org.apache.kafka.coordinator.group.classic.ClassicGroupMember;
import org.apache.kafka.coordinator.group.classic.ClassicGroupState;
import org.apache.kafka.coordinator.group.consumer.Assignment;
import org.apache.kafka.coordinator.group.consumer.ConsumerGroup;
import org.apache.kafka.coordinator.group.consumer.ConsumerGroupMember;
import org.apache.kafka.coordinator.group.consumer.CurrentAssignmentBuilder;
import org.apache.kafka.coordinator.group.consumer.TargetAssignmentBuilder;
import org.apache.kafka.coordinator.group.consumer.TopicMetadata;
import org.apache.kafka.coordinator.group.generated.ConsumerGroupCurrentMemberAssignmentKey;
import org.apache.kafka.coordinator.group.generated.ConsumerGroupCurrentMemberAssignmentValue;
import org.apache.kafka.coordinator.group.generated.ConsumerGroupMemberMetadataKey;
Expand All @@ -85,7 +79,13 @@
import org.apache.kafka.coordinator.group.generated.GroupMetadataKey;
import org.apache.kafka.coordinator.group.generated.GroupMetadataValue;
import org.apache.kafka.coordinator.group.metrics.GroupCoordinatorMetricsShard;
import org.apache.kafka.coordinator.group.modern.Assignment;
import org.apache.kafka.coordinator.group.modern.MemberState;
import org.apache.kafka.coordinator.group.modern.TargetAssignmentBuilder;
import org.apache.kafka.coordinator.group.modern.TopicMetadata;
import org.apache.kafka.coordinator.group.modern.consumer.ConsumerGroup;
import org.apache.kafka.coordinator.group.modern.consumer.ConsumerGroupMember;
import org.apache.kafka.coordinator.group.modern.consumer.CurrentAssignmentBuilder;
import org.apache.kafka.coordinator.group.runtime.CoordinatorResult;
import org.apache.kafka.coordinator.group.runtime.CoordinatorTimer;
import org.apache.kafka.image.MetadataDelta;
Expand Down Expand Up @@ -140,9 +140,9 @@
import static org.apache.kafka.coordinator.group.classic.ClassicGroupState.EMPTY;
import static org.apache.kafka.coordinator.group.classic.ClassicGroupState.PREPARING_REBALANCE;
import static org.apache.kafka.coordinator.group.classic.ClassicGroupState.STABLE;
import static org.apache.kafka.coordinator.group.consumer.ConsumerGroupMember.hasAssignedPartitionsChanged;
import static org.apache.kafka.coordinator.group.metrics.GroupCoordinatorMetrics.CLASSIC_GROUP_COMPLETED_REBALANCES_SENSOR_NAME;
import static org.apache.kafka.coordinator.group.metrics.GroupCoordinatorMetrics.CONSUMER_GROUP_REBALANCES_SENSOR_NAME;
import static org.apache.kafka.coordinator.group.modern.consumer.ConsumerGroupMember.hasAssignedPartitionsChanged;

/**
* The GroupMetadataManager manages the metadata of all classic and consumer groups. It holds
Expand Down Expand Up @@ -2474,7 +2474,7 @@ public void replay(
ConsumerGroup consumerGroup = getOrMaybeCreatePersistedConsumerGroup(groupId, false);

if (value != null) {
consumerGroup.updateTargetAssignment(memberId, Assignment.fromRecord(value));
consumerGroup.updateTargetAssignment(memberId, Utils.fromRecord(value));
apoorvmittal10 marked this conversation as resolved.
Show resolved Hide resolved
} else {
consumerGroup.removeTargetAssignment(memberId);
}
Expand Down
18 changes: 18 additions & 0 deletions group-coordinator/src/main/java/org/apache/kafka/coordinator/group/Utils.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -21,6 +21,8 @@
import org.apache.kafka.common.message.ConsumerProtocolAssignment;
import org.apache.kafka.common.message.ConsumerProtocolSubscription;
import org.apache.kafka.coordinator.group.generated.ConsumerGroupCurrentMemberAssignmentValue;
import org.apache.kafka.coordinator.group.generated.ConsumerGroupTargetAssignmentMemberValue;
import org.apache.kafka.coordinator.group.modern.Assignment;
import org.apache.kafka.image.TopicImage;
import org.apache.kafka.image.TopicsImage;

Expand Down Expand Up @@ -191,4 +193,20 @@ public static Map<Uuid, Set<Integer>> assignmentFromTopicPartitions(
ConsumerGroupCurrentMemberAssignmentValue.TopicPartitions::topicId,
topicPartitions -> Collections.unmodifiableSet(new HashSet<>(topicPartitions.partitions()))));
}

/**
* Creates a {{@link Assignment}} from a {{@link ConsumerGroupTargetAssignmentMemberValue}}.
*
* @param record The record.
* @return A {{@link Assignment}}.
*/
public static Assignment fromRecord(
ConsumerGroupTargetAssignmentMemberValue record
) {
return new Assignment(
record.topicPartitions().stream().collect(Collectors.toMap(
ConsumerGroupTargetAssignmentMemberValue.TopicPartition::topicId,
topicPartitions -> new HashSet<>(topicPartitions.partitions())))
);
}
}
2 changes: 1 addition & 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 @@ -23,7 +23,7 @@
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.coordinator.group.modern.MemberAssignmentImpl;

import java.util.ArrayList;
import java.util.Collection;
Expand Down
2 changes: 1 addition & 1 deletion ...va/org/apache/kafka/coordinator/group/assignor/UniformHeterogeneousAssignmentBuilder.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -22,7 +22,7 @@
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.coordinator.group.modern.MemberAssignmentImpl;
import org.apache.kafka.server.common.TopicIdPartition;

import org.slf4j.Logger;
Expand Down
2 changes: 1 addition & 1 deletion ...java/org/apache/kafka/coordinator/group/assignor/UniformHomogeneousAssignmentBuilder.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -22,7 +22,7 @@
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.coordinator.group.modern.MemberAssignmentImpl;
import org.apache.kafka.server.common.TopicIdPartition;

import java.util.ArrayList;
Expand Down
2 changes: 1 addition & 1 deletion group-coordinator/src/main/java/org/apache/kafka/coordinator/group/classic/ClassicGroup.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -38,8 +38,8 @@
import org.apache.kafka.coordinator.group.Group;
import org.apache.kafka.coordinator.group.OffsetExpirationCondition;
import org.apache.kafka.coordinator.group.OffsetExpirationConditionImpl;
import org.apache.kafka.coordinator.group.consumer.ConsumerGroup;
import org.apache.kafka.coordinator.group.metrics.GroupCoordinatorMetricsShard;
import org.apache.kafka.coordinator.group.modern.consumer.ConsumerGroup;
import org.apache.kafka.image.MetadataImage;
import org.apache.kafka.server.common.MetadataVersion;

Expand Down
2 changes: 1 addition & 1 deletion ...tor/src/main/java/org/apache/kafka/coordinator/group/metrics/GroupCoordinatorMetrics.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -25,7 +25,7 @@
import org.apache.kafka.common.utils.Utils;
import org.apache.kafka.coordinator.group.Group;
import org.apache.kafka.coordinator.group.classic.ClassicGroupState;
import org.apache.kafka.coordinator.group.consumer.ConsumerGroup.ConsumerGroupState;
import org.apache.kafka.coordinator.group.modern.consumer.ConsumerGroup.ConsumerGroupState;
import org.apache.kafka.server.metrics.KafkaYammerMetrics;
import org.apache.kafka.timeline.SnapshotRegistry;

Expand Down
2 changes: 1 addition & 1 deletion ...rc/main/java/org/apache/kafka/coordinator/group/metrics/GroupCoordinatorMetricsShard.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -20,7 +20,7 @@
import org.apache.kafka.common.metrics.Sensor;
import org.apache.kafka.common.utils.Utils;
import org.apache.kafka.coordinator.group.classic.ClassicGroupState;
import org.apache.kafka.coordinator.group.consumer.ConsumerGroup.ConsumerGroupState;
import org.apache.kafka.coordinator.group.modern.consumer.ConsumerGroup.ConsumerGroupState;
import org.apache.kafka.timeline.SnapshotRegistry;
import org.apache.kafka.timeline.TimelineLong;

Expand Down
21 changes: 1 addition & 20 deletions ...oordinator/group/consumer/Assignment.java → .../coordinator/group/modern/Assignment.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -14,18 +14,15 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.apache.kafka.coordinator.group.consumer;
package org.apache.kafka.coordinator.group.modern;

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;
import java.util.HashSet;
import java.util.Map;
import java.util.Objects;
import java.util.Set;
import java.util.stream.Collectors;

/**
* An immutable assignment for a member.
Expand Down Expand Up @@ -69,20 +66,4 @@ public int hashCode() {
public String toString() {
return "Assignment(partitions=" + partitions + ')';
}

/**
* Creates a {{@link Assignment}} from a {{@link ConsumerGroupTargetAssignmentMemberValue}}.
*
* @param record The record.
* @return A {{@link Assignment}}.
*/
public static Assignment fromRecord(
apoorvmittal10 marked this conversation as resolved.
Show resolved Hide resolved
ConsumerGroupTargetAssignmentMemberValue record
) {
return new Assignment(
record.topicPartitions().stream().collect(Collectors.toMap(
ConsumerGroupTargetAssignmentMemberValue.TopicPartition::topicId,
topicPartitions -> new HashSet<>(topicPartitions.partitions())))
);
}
}
4 changes: 2 additions & 2 deletions ...dinator/group/consumer/GroupSpecImpl.java → ...ordinator/group/modern/GroupSpecImpl.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.consumer;
package org.apache.kafka.coordinator.group.modern;

import org.apache.kafka.common.Uuid;
import org.apache.kafka.coordinator.group.api.assignor.GroupSpec;
Expand All @@ -28,7 +28,7 @@
import java.util.Objects;

/**
* The assignment specification for a consumer group.
* The assignment specification for a modern group.
Copy link
Contributor

Choose a reason for hiding this comment

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

not sure whether this is necessary change. Maybe we should align the term: "next gen", "modern", or "CONSUMER".

Copy link
Contributor Author

Choose a reason for hiding this comment

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

Thanks for your review @chia7712. I have aligned the changes to have the common functionality as modern. As this spec class will be used in target assignment which will be also utilized by upcoming share groups. Please let me know if there is correction needed elsewhere as well.

Copy link
Contributor

@chia7712 chia7712 Jul 1, 2024 •
edited
Loading

Choose a reason for hiding this comment

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

sorry for raising one more question

As this spec class will be used in target assignment which will be also utilized by upcoming share groups

it seems those share/consumer groups are on top of the new protocol, and the root package of those classes is called "modern". Hence, should we rename the value of group.coordinator.rebalance.protocols from CONSUMER to MODERN?

https://github.com/apache/kafka/blob/trunk/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/Group.java#L36

Copy link
Contributor Author

Choose a reason for hiding this comment

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

I don't think that we shall change the group.coordinator.rebalance.protocols itself. There shall remain 2 for those as classic and consumer. Package modern is more to share the common target assignment code between consumer and share groups. Share groups will be be specifically used for a consumption pattern where partitions get shared between clients to facilitate co-operative consumption.

Copy link
Contributor

Choose a reason for hiding this comment

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

I raise the "naming" issue, because it is hard to distinguish the term of "consumer protocol" and "consumer instance" in introducing the "AsyncConsumer" and "CONSUMER protocol" to users. I normally give up the term used by source code and then using term "next generation" be the replacement :_

Copy link
Contributor Author

Choose a reason for hiding this comment

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

I ll defer this question to @dajac as the naming for new consumer protocol (KIP-848) defined "consumer" as the new rebalance protocol itself. But I don't see much of relation between package names and protocol name, they can be independent.

*/
public class GroupSpecImpl implements GroupSpec {
/**
Expand Down
4 changes: 2 additions & 2 deletions .../group/consumer/MemberAssignmentImpl.java → ...or/group/modern/MemberAssignmentImpl.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.consumer;
package org.apache.kafka.coordinator.group.modern;

import org.apache.kafka.common.Uuid;
import org.apache.kafka.coordinator.group.api.assignor.MemberAssignment;
Expand All @@ -24,7 +24,7 @@
import java.util.Set;

/**
* The partition assignment for a consumer group member.
* The partition assignment for a modern group member.
*/
public class MemberAssignmentImpl implements MemberAssignment {
/**
Expand Down
2 changes: 1 addition & 1 deletion group-coordinator/src/main/java/org/apache/kafka/coordinator/group/modern/MemberState.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -16,7 +16,7 @@
*/
package org.apache.kafka.coordinator.group.modern;

import org.apache.kafka.coordinator.group.consumer.CurrentAssignmentBuilder;
import org.apache.kafka.coordinator.group.modern.consumer.CurrentAssignmentBuilder;

import java.util.HashMap;
import java.util.Map;
Expand Down
2 changes: 1 addition & 1 deletion .../MemberSubscriptionAndAssignmentImpl.java → .../MemberSubscriptionAndAssignmentImpl.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.consumer;
package org.apache.kafka.coordinator.group.modern;

import org.apache.kafka.common.Uuid;
import org.apache.kafka.coordinator.group.api.assignor.MemberAssignment;
Expand Down
2 changes: 0 additions & 2 deletions group-coordinator/src/main/java/org/apache/kafka/coordinator/group/modern/ModernGroup.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -21,8 +21,6 @@
import org.apache.kafka.coordinator.group.Group;
import org.apache.kafka.coordinator.group.Utils;
import org.apache.kafka.coordinator.group.api.assignor.SubscriptionType;
import org.apache.kafka.coordinator.group.consumer.Assignment;
import org.apache.kafka.coordinator.group.consumer.TopicMetadata;
import org.apache.kafka.image.ClusterImage;
import org.apache.kafka.image.TopicImage;
import org.apache.kafka.image.TopicsImage;
Expand Down
4 changes: 2 additions & 2 deletions ...onsumer/SubscribedTopicDescriberImpl.java → .../modern/SubscribedTopicDescriberImpl.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.consumer;
package org.apache.kafka.coordinator.group.modern;

import org.apache.kafka.common.Uuid;
import org.apache.kafka.coordinator.group.api.assignor.PartitionAssignor;
Expand All @@ -27,7 +27,7 @@

/**
* 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.
* topic and partition metadata for the topics that the modern group is subscribed to.
*/
public class SubscribedTopicDescriberImpl implements SubscribedTopicDescriber {
/**
Expand Down
3 changes: 1 addition & 2 deletions ...oup/consumer/TargetAssignmentBuilder.java → ...group/modern/TargetAssignmentBuilder.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.consumer;
package org.apache.kafka.coordinator.group.modern;

import org.apache.kafka.common.Uuid;
import org.apache.kafka.coordinator.group.CoordinatorRecord;
Expand All @@ -23,7 +23,6 @@
import org.apache.kafka.coordinator.group.api.assignor.PartitionAssignor;
import org.apache.kafka.coordinator.group.api.assignor.PartitionAssignorException;
import org.apache.kafka.coordinator.group.api.assignor.SubscriptionType;
import org.apache.kafka.coordinator.group.modern.ModernGroupMember;
import org.apache.kafka.image.TopicsImage;

import java.util.ArrayList;
Expand Down
2 changes: 1 addition & 1 deletion .../coordinator/group/consumer/TopicIds.java → ...ka/coordinator/group/modern/TopicIds.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.consumer;
package org.apache.kafka.coordinator.group.modern;

import org.apache.kafka.common.Uuid;
import org.apache.kafka.image.TopicImage;
Expand Down
2 changes: 1 addition & 1 deletion ...dinator/group/consumer/TopicMetadata.java → ...ordinator/group/modern/TopicMetadata.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.consumer;
package org.apache.kafka.coordinator.group.modern;

import org.apache.kafka.common.Uuid;
import org.apache.kafka.coordinator.group.generated.ConsumerGroupPartitionMetadataValue;
Expand Down
11 changes: 6 additions & 5 deletions ...dinator/group/consumer/ConsumerGroup.java → .../group/modern/consumer/ConsumerGroup.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.consumer;
package org.apache.kafka.coordinator.group.modern.consumer;

import org.apache.kafka.clients.consumer.internals.ConsumerProtocol;
import org.apache.kafka.common.Uuid;
Expand All @@ -34,6 +34,7 @@
import org.apache.kafka.coordinator.group.classic.ClassicGroup;
import org.apache.kafka.coordinator.group.generated.ConsumerGroupMemberMetadataValue;
import org.apache.kafka.coordinator.group.metrics.GroupCoordinatorMetricsShard;
import org.apache.kafka.coordinator.group.modern.Assignment;
import org.apache.kafka.coordinator.group.modern.MemberState;
import org.apache.kafka.coordinator.group.modern.ModernGroup;
import org.apache.kafka.coordinator.group.modern.ModernGroupMember;
Expand All @@ -55,10 +56,10 @@

import static org.apache.kafka.coordinator.group.Utils.toOptional;
import static org.apache.kafka.coordinator.group.Utils.toTopicPartitionMap;
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;
import static org.apache.kafka.coordinator.group.consumer.ConsumerGroup.ConsumerGroupState.STABLE;
import static org.apache.kafka.coordinator.group.modern.consumer.ConsumerGroup.ConsumerGroupState.ASSIGNING;
import static org.apache.kafka.coordinator.group.modern.consumer.ConsumerGroup.ConsumerGroupState.EMPTY;
import static org.apache.kafka.coordinator.group.modern.consumer.ConsumerGroup.ConsumerGroupState.RECONCILING;
import static org.apache.kafka.coordinator.group.modern.consumer.ConsumerGroup.ConsumerGroupState.STABLE;

/**
* A Consumer Group. All the metadata in this class are backed by
Expand Down
3 changes: 2 additions & 1 deletion ...r/group/consumer/ConsumerGroupMember.java → .../modern/consumer/ConsumerGroupMember.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -14,14 +14,15 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.apache.kafka.coordinator.group.consumer;
package org.apache.kafka.coordinator.group.modern.consumer;

import org.apache.kafka.common.Uuid;
import org.apache.kafka.common.message.ConsumerGroupDescribeResponseData;
import org.apache.kafka.common.message.JoinGroupRequestData;
import org.apache.kafka.coordinator.group.Utils;
import org.apache.kafka.coordinator.group.generated.ConsumerGroupCurrentMemberAssignmentValue;
import org.apache.kafka.coordinator.group.generated.ConsumerGroupMemberMetadataValue;
import org.apache.kafka.coordinator.group.modern.Assignment;
import org.apache.kafka.coordinator.group.modern.MemberState;
import org.apache.kafka.coordinator.group.modern.ModernGroupMember;
import org.apache.kafka.image.TopicImage;
Expand Down
3 changes: 2 additions & 1 deletion ...up/consumer/CurrentAssignmentBuilder.java → ...rn/consumer/CurrentAssignmentBuilder.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -14,11 +14,12 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.apache.kafka.coordinator.group.consumer;
package org.apache.kafka.coordinator.group.modern.consumer;

import org.apache.kafka.common.Uuid;
import org.apache.kafka.common.errors.FencedMemberEpochException;
import org.apache.kafka.common.message.ConsumerGroupHeartbeatRequestData;
import org.apache.kafka.coordinator.group.modern.Assignment;
import org.apache.kafka.coordinator.group.modern.MemberState;

import java.util.Collections;
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 @@ -18,7 +18,7 @@

import org.apache.kafka.common.Uuid;
import org.apache.kafka.coordinator.group.api.assignor.GroupAssignment;
import org.apache.kafka.coordinator.group.consumer.MemberSubscriptionAndAssignmentImpl;
import org.apache.kafka.coordinator.group.modern.MemberSubscriptionAndAssignmentImpl;

import java.util.AbstractMap;
import java.util.Arrays;
Expand Down
Loading