@Override
        public GroupCoordinatorShard build() {
            if (logContext == null) logContext = new LogContext();
            if (config == null)
                throw new IllegalArgumentException("Config must be set.");
            if (snapshotRegistry == null)
                throw new IllegalArgumentException("SnapshotRegistry must be set.");
            if (time == null)
                throw new IllegalArgumentException("Time must be set.");
            if (timer == null)
                throw new IllegalArgumentException("Timer must be set.");
            if (coordinatorMetrics == null || !(coordinatorMetrics instanceof GroupCoordinatorMetrics))
                throw new IllegalArgumentException("CoordinatorMetrics must be set and be of type GroupCoordinatorMetrics.");
            if (topicPartition == null)
                throw new IllegalArgumentException("TopicPartition must be set.");

            GroupCoordinatorMetricsShard metricsShard = ((GroupCoordinatorMetrics) coordinatorMetrics)
                .newMetricsShard(snapshotRegistry, topicPartition);

            GroupMetadataManager groupMetadataManager = new GroupMetadataManager.Builder()
                .withLogContext(logContext)
                .withSnapshotRegistry(snapshotRegistry)
                .withTime(time)
                .withTimer(timer)
                .withConsumerGroupAssignors(config.consumerGroupAssignors())
                .withConsumerGroupMaxSize(config.consumerGroupMaxSize())
                .withConsumerGroupSessionTimeout(config.consumerGroupSessionTimeoutMs())
                .withConsumerGroupHeartbeatInterval(config.consumerGroupHeartbeatIntervalMs())
                .withClassicGroupMaxSize(config.classicGroupMaxSize())
                .withClassicGroupInitialRebalanceDelayMs(config.classicGroupInitialRebalanceDelayMs())
                .withClassicGroupNewMemberJoinTimeoutMs(config.classicGroupNewMemberJoinTimeoutMs())
                .withClassicGroupMinSessionTimeoutMs(config.classicGroupMinSessionTimeoutMs())
                .withClassicGroupMaxSessionTimeoutMs(config.classicGroupMaxSessionTimeoutMs())
                .withConsumerGroupMigrationPolicy(config.consumerGroupMigrationPolicy())
                .withShareGroupMaxSize(config.shareGroupMaxSize())
                .withShareGroupSessionTimeout(config.shareGroupSessionTimeoutMs())
                .withShareGroupHeartbeatInterval(config.shareGroupHeartbeatIntervalMs())
                .withGroupCoordinatorMetricsShard(metricsShard)
                .build();

            OffsetMetadataManager offsetMetadataManager = new OffsetMetadataManager.Builder()
                .withLogContext(logContext)
                .withSnapshotRegistry(snapshotRegistry)
                .withTime(time)
                .withGroupMetadataManager(groupMetadataManager)
                .withGroupCoordinatorConfig(config)
                .withGroupCoordinatorMetricsShard(metricsShard)
                .build();

            return new GroupCoordinatorShard(
                logContext,
                groupMetadataManager,
                offsetMetadataManager,
                time,
                timer,
                config,
                coordinatorMetrics,
                metricsShard
            );
        }
The group coordinator shard is a replicated state machine that manages the metadata of all
classic 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.
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupCoordinatorShard.java