MINOR: Code cleanups in group-coordinator module

group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupCoordinatorService.java

@@ -133,17 +133,17 @@ public GroupCoordinatorService build() {
             String logPrefix = String.format("GroupCoordinator id=%d", nodeId);
             LogContext logContext = new LogContext(String.format("[%s] ", logPrefix));
 
-            CoordinatorBuilderSupplier<ReplicatedGroupCoordinator, Record> supplier = () ->
-                new ReplicatedGroupCoordinator.Builder(config);
+            CoordinatorBuilderSupplier<GroupCoordinatorShard, Record> supplier = () ->
+                new GroupCoordinatorShard.Builder(config);
 
             CoordinatorEventProcessor processor = new MultiThreadedEventProcessor(
                 logContext,
                 "group-coordinator-event-processor-",
                 config.numThreads
             );
 
-            CoordinatorRuntime<ReplicatedGroupCoordinator, Record> runtime =
-                new CoordinatorRuntime.Builder<ReplicatedGroupCoordinator, Record>()
+            CoordinatorRuntime<GroupCoordinatorShard, Record> runtime =
+                new CoordinatorRuntime.Builder<GroupCoordinatorShard, Record>()
                     .withTime(time)
                     .withTimer(timer)
                     .withLogPrefix(logPrefix)
@@ -176,7 +176,7 @@ public GroupCoordinatorService build() {
     /**
      * The coordinator runtime.
      */
-    private final CoordinatorRuntime<ReplicatedGroupCoordinator, Record> runtime;
+    private final CoordinatorRuntime<GroupCoordinatorShard, Record> runtime;
 
     /**
      * Boolean indicating whether the coordinator is active or not.
@@ -198,7 +198,7 @@ public GroupCoordinatorService build() {
     GroupCoordinatorService(
         LogContext logContext,
         GroupCoordinatorConfig config,
-        CoordinatorRuntime<ReplicatedGroupCoordinator, Record> runtime
+        CoordinatorRuntime<GroupCoordinatorShard, Record> runtime
     ) {
         this.log = logContext.logger(CoordinatorLoader.class);
         this.config = config;

group-coordinator/src/main/java/org/apache/kafka/coordinator/group/ReplicatedGroupCoordinator.java → group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupCoordinatorShard.java

@@ -16,7 +16,6 @@
  */
 package org.apache.kafka.coordinator.group;
 
-import org.apache.kafka.common.TopicPartition;
 import org.apache.kafka.common.message.ConsumerGroupHeartbeatRequestData;
 import org.apache.kafka.common.message.ConsumerGroupHeartbeatResponseData;
 import org.apache.kafka.common.message.JoinGroupRequestData;
@@ -46,8 +45,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.runtime.Coordinator;
-import org.apache.kafka.coordinator.group.runtime.CoordinatorBuilder;
+import org.apache.kafka.coordinator.group.runtime.CoordinatorShard;
+import org.apache.kafka.coordinator.group.runtime.CoordinatorShardBuilder;
 import org.apache.kafka.coordinator.group.runtime.CoordinatorResult;
 import org.apache.kafka.coordinator.group.runtime.CoordinatorTimer;
 import org.apache.kafka.image.MetadataDelta;
@@ -58,22 +57,21 @@
 import java.util.concurrent.CompletableFuture;
 
 /**
- * The group coordinator replicated state machine that manages the metadata of all generic and
- * consumer groups. It holds the hard and the soft state of the groups. This class has two kinds
- * of methods:
+ * The group coordinator shard is a replicated state machine that manages the metadata of all
+ * generic and consumer groups. It holds the hard and the soft state of the groups. This class
+ * has two kinds of methods:
  * 1) The request handlers which handle the requests and generate a response and records to
  *    mutate the hard state. Those records will be written by the runtime and applied to the
  *    hard state via the replay methods.
  * 2) The replay methods which apply records to the hard state. Those are used in the request
  *    handling as well as during the initial loading of the records from the partitions.
  */
-public class ReplicatedGroupCoordinator implements Coordinator<Record> {
+public class GroupCoordinatorShard implements CoordinatorShard<Record> {
 
-    public static class Builder implements CoordinatorBuilder<ReplicatedGroupCoordinator, Record> {
+    public static class Builder implements CoordinatorShardBuilder<GroupCoordinatorShard, Record> {
         private final GroupCoordinatorConfig config;
         private LogContext logContext;
         private SnapshotRegistry snapshotRegistry;
-        private TopicPartition topicPartition;
         private Time time;
         private CoordinatorTimer<Void, Record> timer;
 
@@ -84,47 +82,39 @@ public Builder(
         }
 
         @Override
-        public CoordinatorBuilder<ReplicatedGroupCoordinator, Record> withLogContext(
+        public CoordinatorShardBuilder<GroupCoordinatorShard, Record> withLogContext(
             LogContext logContext
         ) {
             this.logContext = logContext;
             return this;
         }
 
         @Override
-        public CoordinatorBuilder<ReplicatedGroupCoordinator, Record> withTime(
+        public CoordinatorShardBuilder<GroupCoordinatorShard, Record> withTime(
             Time time
         ) {
             this.time = time;
             return this;
         }
 
         @Override
-        public CoordinatorBuilder<ReplicatedGroupCoordinator, Record> withTimer(
+        public CoordinatorShardBuilder<GroupCoordinatorShard, Record> withTimer(
             CoordinatorTimer<Void, Record> timer
         ) {
             this.timer = timer;
             return this;
         }
 
         @Override
-        public CoordinatorBuilder<ReplicatedGroupCoordinator, Record> withSnapshotRegistry(
+        public CoordinatorShardBuilder<GroupCoordinatorShard, Record> withSnapshotRegistry(
             SnapshotRegistry snapshotRegistry
         ) {
             this.snapshotRegistry = snapshotRegistry;
             return this;
         }
 
         @Override
-        public CoordinatorBuilder<ReplicatedGroupCoordinator, Record> withTopicPartition(
-            TopicPartition topicPartition
-        ) {
-            this.topicPartition = topicPartition;
-            return this;
-        }
-
-        @Override
-        public ReplicatedGroupCoordinator build() {
+        public GroupCoordinatorShard build() {
             if (logContext == null) logContext = new LogContext();
             if (config == null)
                 throw new IllegalArgumentException("Config must be set.");
@@ -134,16 +124,13 @@ public ReplicatedGroupCoordinator build() {
                 throw new IllegalArgumentException("Time must be set.");
             if (timer == null)
                 throw new IllegalArgumentException("Timer must be set.");
-            if (topicPartition == null)
-                throw new IllegalArgumentException("TopicPartition must be set.");
 
             GroupMetadataManager groupMetadataManager = new GroupMetadataManager.Builder()
                 .withLogContext(logContext)
                 .withSnapshotRegistry(snapshotRegistry)
                 .withTime(time)
                 .withTimer(timer)
-                .withTopicPartition(topicPartition)
-                .withAssignors(config.consumerGroupAssignors)
+                .withConsumerGroupAssignors(config.consumerGroupAssignors)
                 .withConsumerGroupMaxSize(config.consumerGroupMaxSize)
                 .withConsumerGroupHeartbeatInterval(config.consumerGroupHeartbeatIntervalMs)
                 .withGenericGroupInitialRebalanceDelayMs(config.genericGroupInitialRebalanceDelayMs)
@@ -160,7 +147,7 @@ public ReplicatedGroupCoordinator build() {
                 .withOffsetMetadataMaxSize(config.offsetMetadataMaxSize)
                 .build();
 
-            return new ReplicatedGroupCoordinator(
+            return new GroupCoordinatorShard(
                 groupMetadataManager,
                 offsetMetadataManager
             );
@@ -183,7 +170,7 @@ public ReplicatedGroupCoordinator build() {
      * @param groupMetadataManager  The group metadata manager.
      * @param offsetMetadataManager The offset metadata manager.
      */
-    ReplicatedGroupCoordinator(
+    GroupCoordinatorShard(
         GroupMetadataManager groupMetadataManager,
         OffsetMetadataManager offsetMetadataManager
     ) {

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

@@ -17,7 +17,6 @@
 package org.apache.kafka.coordinator.group;
 
 import org.apache.kafka.common.KafkaException;
-import org.apache.kafka.common.TopicPartition;
 import org.apache.kafka.common.Uuid;
 import org.apache.kafka.common.errors.ApiException;
 import org.apache.kafka.common.errors.FencedMemberEpochException;
@@ -126,11 +125,10 @@ public static class Builder {
         private SnapshotRegistry snapshotRegistry = null;
         private Time time = null;
         private CoordinatorTimer<Void, Record> timer = null;
-        private List<PartitionAssignor> assignors = null;
+        private List<PartitionAssignor> consumerGroupAssignors = null;
         private int consumerGroupMaxSize = Integer.MAX_VALUE;
         private int consumerGroupHeartbeatIntervalMs = 5000;
         private int consumerGroupMetadataRefreshIntervalMs = Integer.MAX_VALUE;
-        private TopicPartition topicPartition = null;
         private MetadataImage metadataImage = null;
         private int consumerGroupSessionTimeoutMs = 45000;
         private int genericGroupMaxSize = Integer.MAX_VALUE;
@@ -159,8 +157,8 @@ Builder withTimer(CoordinatorTimer<Void, Record> timer) {
             return this;
         }
 
-        Builder withAssignors(List<PartitionAssignor> assignors) {
-            this.assignors = assignors;
+        Builder withConsumerGroupAssignors(List<PartitionAssignor> consumerGroupAssignors) {
+            this.consumerGroupAssignors = consumerGroupAssignors;
             return this;
         }
 
@@ -189,11 +187,6 @@ Builder withMetadataImage(MetadataImage metadataImage) {
             return this;
         }
 
-        Builder withTopicPartition(TopicPartition tp) {
-            this.topicPartition = tp;
-            return this;
-        }
-
         Builder withGenericGroupMaxSize(int genericGroupMaxSize) {
             this.genericGroupMaxSize = genericGroupMaxSize;
             return this;
@@ -227,20 +220,15 @@ GroupMetadataManager build() {
 
             if (timer == null)
                 throw new IllegalArgumentException("Timer must be set.");
-            if (assignors == null || assignors.isEmpty())
+            if (consumerGroupAssignors == null || consumerGroupAssignors.isEmpty())
                 throw new IllegalArgumentException("Assignors must be set before building.");
 
-            if (topicPartition == null) {
-                throw new IllegalStateException("TopicPartition must be set before building.");
-            }
-
             return new GroupMetadataManager(
-                topicPartition,
                 snapshotRegistry,
                 logContext,
                 time,
                 timer,
-                assignors,
+                consumerGroupAssignors,
                 metadataImage,
                 consumerGroupMaxSize,
                 consumerGroupSessionTimeoutMs,
@@ -255,11 +243,6 @@ GroupMetadataManager build() {
         }
     }
 
-    /**
-     * The topic partition associated with the metadata manager.
-     */
-    private final TopicPartition topicPartition;
-
     /**
      * The log context.
      */
@@ -365,7 +348,6 @@ GroupMetadataManager build() {
     private final int genericGroupMaxSessionTimeoutMs;
 
     private GroupMetadataManager(
-        TopicPartition topicPartition,
         SnapshotRegistry snapshotRegistry,
         LogContext logContext,
         Time time,
@@ -389,7 +371,6 @@ private GroupMetadataManager(
         this.timer = timer;
         this.metadataImage = metadataImage;
         this.assignors = assignors.stream().collect(Collectors.toMap(PartitionAssignor::name, Function.identity()));
-        this.topicPartition = topicPartition;
         this.defaultAssignor = assignors.get(0);
         this.groups = new TimelineHashMap<>(snapshotRegistry, 0);
         this.groupsByTopics = new TimelineHashMap<>(snapshotRegistry, 0);
@@ -1968,8 +1949,7 @@ private CoordinatorResult<Void, Record> completeGenericGroupJoin(
         } else {
             group.initNextGeneration();
             if (group.isInState(EMPTY)) {
-                log.info("Group {} with generation {} is now empty ({}-{})",
-                    groupId, group.generationId(), topicPartition.topic(), topicPartition.partition());
+                log.info("Group {} with generation {} is now empty.", groupId, group.generationId());
 
                 CompletableFuture<Void> appendFuture = new CompletableFuture<>();
                 appendFuture.whenComplete((__, t) -> {
@@ -1987,8 +1967,8 @@ private CoordinatorResult<Void, Record> completeGenericGroupJoin(
                 return new CoordinatorResult<>(records, appendFuture);
 
             } else {
-                log.info("Stabilized group {} generation {} ({}) with {} members",
-                    groupId, group.generationId(), topicPartition, group.size());
+                log.info("Stabilized group {} generation {} with {} members.",
+                    groupId, group.generationId(), group.size());
 
                 // Complete the awaiting join group response future for all the members after rebalancing
                 group.allMembers().forEach(member -> {
@@ -2267,9 +2247,8 @@ CoordinatorResult<Void, Record> prepareRebalance(
 
         group.transitionTo(PREPARING_REBALANCE);
 
-        log.info("Preparing to rebalance group {} in state {} with old generation {} ({}-{}) (reason: {})",
-            group.groupId(), group.currentState(), group.generationId(),
-            topicPartition.topic(), topicPartition.partition(), reason);
+        log.info("Preparing to rebalance group {} in state {} with old generation {} (reason: {}).",
+            group.groupId(), group.currentState(), group.generationId(), reason);
 
         return isInitialRebalance ? EMPTY_RESULT : maybeCompleteJoinElseSchedule(group);
     }

group-coordinator/src/main/java/org/apache/kafka/coordinator/group/runtime/CoordinatorBuilderSupplier.java

@@ -17,14 +17,14 @@
 package org.apache.kafka.coordinator.group.runtime;
 
 /**
- * Supplies a {@link CoordinatorBuilder} to the {@link CoordinatorRuntime}.
+ * Supplies a {@link CoordinatorShardBuilder} to the {@link CoordinatorRuntime}.
  *
  * @param <S> The type of the coordinator.
  * @param <U> The record type.
  */
-public interface CoordinatorBuilderSupplier<S extends Coordinator<U>, U> {
+public interface CoordinatorBuilderSupplier<S extends CoordinatorShard<U>, U> {
     /**
-     * @return A {@link CoordinatorBuilder}.
+     * @return A {@link CoordinatorShardBuilder}.
      */
-    CoordinatorBuilder<S, U> get();
+    CoordinatorShardBuilder<S, U> get();
 }

group-coordinator/src/main/java/org/apache/kafka/coordinator/group/runtime/CoordinatorRuntime.java

@@ -70,15 +70,15 @@
  * @param <S> The type of the state machine.
  * @param <U> The type of the record.
  */
-public class CoordinatorRuntime<S extends Coordinator<U>, U> implements AutoCloseable {
+public class CoordinatorRuntime<S extends CoordinatorShard<U>, U> implements AutoCloseable {
 
     /**
      * Builder to create a CoordinatorRuntime.
      *
      * @param <S> The type of the state machine.
      * @param <U> The type of the record.
      */
-    public static class Builder<S extends Coordinator<U>, U> {
+    public static class Builder<S extends CoordinatorShard<U>, U> {
         private String logPrefix;
         private LogContext logContext;
         private CoordinatorEventProcessor eventProcessor;
@@ -505,7 +505,6 @@ private void transitionTo(
                         .withSnapshotRegistry(snapshotRegistry)
                         .withTime(time)
                         .withTimer(timer)
-                        .withTopicPartition(tp)
                         .build();
                     break;
 

group-coordinator/src/main/java/org/apache/kafka/coordinator/group/runtime/Coordinator.java → group-coordinator/src/main/java/org/apache/kafka/coordinator/group/runtime/CoordinatorShard.java

@@ -20,10 +20,10 @@
 import org.apache.kafka.image.MetadataImage;
 
 /**
- * Coordinator is basically a replicated state machine managed by the
+ * CoordinatorShard is basically a replicated state machine managed by the
  * {@link CoordinatorRuntime}.
  */
-public interface Coordinator<U> extends CoordinatorPlayback<U> {
+public interface CoordinatorShard<U> extends CoordinatorPlayback<U> {
 
     /**
      * The coordinator has been loaded. This is used to apply any
@@ -34,7 +34,7 @@ public interface Coordinator<U> extends CoordinatorPlayback<U> {
     default void onLoaded(MetadataImage newImage) {}
 
     /**
-     * A new metadata image is available. This is only called after {@link Coordinator#onLoaded(MetadataImage)}
+     * A new metadata image is available. This is only called after {@link CoordinatorShard#onLoaded(MetadataImage)}
      * is called to signal that the coordinator has been fully loaded.
      *
      * @param newImage  The new metadata image.