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

Code before:
    /**
     * 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.onNewMetadataImage(wrappedImage, wrappedDelta);
    }

    /**
     * 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);
    }

    /**

Review comment (refactor):
This is pretty ugly and it should not be here because the image and the delta can never be null. Unfortunately, we have a couple of tests passing nulls here. I will fix those separately and remove this code in a follow-up. I want the core change to be well scoped.

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);
    }

    /**