Repository: apache/kafka
File: group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupCoordinatorService.java (Java)

Code before:
     */
    @Override
    public void onResignation(
        int groupMetadataPartitionIndex,
        OptionalInt groupMetadataPartitionLeaderEpoch
    ) {
        throwIfNotActive();
        runtime.scheduleUnloadOperation(
            new TopicPartition(Topic.GROUP_METADATA_TOPIC_NAME, groupMetadataPartitionIndex),
            groupMetadataPartitionLeaderEpoch
        );
    }

    /**
     * See {@link GroupCoordinator#onNewMetadataImage(CoordinatorMetadataImage, CoordinatorMetadataDelta)}.
     */
    @Override
    public void onNewMetadataImage(
        CoordinatorMetadataImage newImage,
        CoordinatorMetadataDelta delta
    ) {
        throwIfNotActive();
        metadataImage = newImage;
        runtime.onNewMetadataImage(newImage, delta);
    }

    /**
     * See {@link GroupCoordinator#groupMetadataTopicConfigs()}.
     */
    @Override
    public Properties groupMetadataTopicConfigs() {
        Properties properties = new Properties();
        properties.put(TopicConfig.CLEANUP_POLICY_CONFIG, TopicConfig.CLEANUP_POLICY_COMPACT);
        properties.put(TopicConfig.COMPRESSION_TYPE_CONFIG, BrokerCompressionType.PRODUCER.name);
        properties.put(TopicConfig.SEGMENT_BYTES_CONFIG, String.valueOf(config.offsetsTopicSegmentBytes()));
        return properties;
    }

    /**
     * See {@link GroupCoordinator#groupConfig(String)}.
     */
    @Override
    public Optional<GroupConfig> groupConfig(String groupId) {
        return groupConfigManager.groupConfig(groupId);
    }

    /**
     * See {@link GroupCoordinator#updateGroupConfig(String, Properties)}.
     */
    @Override
    public void updateGroupConfig(String groupId, Properties newGroupConfig) {

Review comment (bug):
Can the image be `null`? Since a `null` `wrappedImage` causes an NPE in `CoordinatorRuntime#onMetadataUpdate`, if `null` input is allowed, we should implement a fast-fail

https://github.com/apache/kafka/blob/ed3af7212405dd5294733ad33d192f12dc885fc2/coordinator-common/src/main/java/org/apache/kafka/coordinator/common/runtime/CoordinatorRuntime.java#L2546

Code after:

    /**
     * See {@link GroupCoordinator#onResignation(int, OptionalInt)}.
     */
    @Override
    public void onResignation(
        int groupMetadataPartitionIndex,
        OptionalInt groupMetadataPartitionLeaderEpoch
    ) {
        throwIfNotActive();
        runtime.scheduleUnloadOperation(
            new TopicPartition(Topic.GROUP_METADATA_TOPIC_NAME, groupMetadataPartitionIndex),
            groupMetadataPartitionLeaderEpoch
        );
    }

    /**
     * See {@link GroupCoordinator#onMetadataUpdate(MetadataDelta, MetadataImage)}.
     */
    @Override
    public void onMetadataUpdate(
        MetadataDelta delta,
        MetadataImage newImage
    ) {
        throwIfNotActive();
        var wrappedImage = newImage == null ? null : new KRaftCoordinatorMetadataImage(newImage);
        var wrappedDelta = delta == null ? null : new KRaftCoordinatorMetadataDelta(delta);
        metadataImage = wrappedImage;
        runtime.onMetadataUpdate(wrappedDelta, wrappedImage);
    }

    /**
     * See {@link GroupCoordinator#groupMetadataTopicConfigs()}.
     */
    @Override
    public Properties groupMetadataTopicConfigs() {
        Properties properties = new Properties();
        properties.put(TopicConfig.CLEANUP_POLICY_CONFIG, TopicConfig.CLEANUP_POLICY_COMPACT);
        properties.put(TopicConfig.COMPRESSION_TYPE_CONFIG, BrokerCompressionType.PRODUCER.name);
        properties.put(TopicConfig.SEGMENT_BYTES_CONFIG, String.valueOf(config.offsetsTopicSegmentBytes()));
        return properties;
    }

    /**
     * See {@link GroupCoordinator#groupConfig(String)}.
     */
    @Override
    public Optional<GroupConfig> groupConfig(String groupId) {
        return groupConfigManager.groupConfig(groupId);
    }