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

Code before:
    // Visibility for testing
    Map<Uuid, InitMapValue> subscribedTopicsChangeMap(String groupId, Map<String, TopicMetadata> subscriptionMetadata) {
        if (subscriptionMetadata == null || subscriptionMetadata.isEmpty()) {
            return Map.of();
        }

        Map<Uuid, InitMapValue> topicPartitionChangeMap = new HashMap<>();
        ShareGroupStatePartitionMetadataInfo info = shareGroupPartitionMetadata.get(groupId);

        // Fresh initializing only
        long curTimestamp = time.milliseconds();
        long delta = config.offsetCommitTimeoutMs() * 2L;
        Map<Uuid, InitMapValue> alreadyInitialized = info == null ? new HashMap<>() :
            combineInitMaps(
                info.initializedTopics(),
                info.initializingTopics().entrySet().stream()
                    .filter(entry -> curTimestamp - entry.getValue().timestamp() < delta)
                    .collect(Collectors.toMap(Map.Entry::getKey, Map.Entry::getValue))
            );

        subscriptionMetadata.forEach((topicName, topicMetadata) -> {
            Set<Integer> alreadyInitializedPartSet = alreadyInitialized.containsKey(topicMetadata.id()) ? alreadyInitialized.get(topicMetadata.id()).partitions() : Set.of();
            if (alreadyInitializedPartSet.isEmpty() || alreadyInitializedPartSet.size() < topicMetadata.numPartitions()) {
                Set<Integer> partitionSet = IntStream.range(0, topicMetadata.numPartitions()).boxed().collect(Collectors.toSet());
                partitionSet.removeAll(alreadyInitializedPartSet);
                // alreadyInitialized contains all initialized topics and less than delta seconds old initializing topics
                // which means we are putting subscribed topics which are unseen or initializing for more than delta seconds. But, we
                // are also updating the timestamp here which means, old initializing will not be included repeatedly.
                topicPartitionChangeMap.computeIfAbsent(topicMetadata.id(), k -> new InitMapValue(topicMetadata.name(), partitionSet, curTimestamp));
            }
        });
        return topicPartitionChangeMap;
    }

    /**
     * Based on the diff between the subscribed topic partitions and the initialized topic partitions,
     * created initialize request for the non-initialized ones.
     *
     * @param groupId               The share group id for which partitions need to be initialized.
     * @param groupEpoch            The group epoch of the share group.
     * @param subscriptionMetadata  The subscription metadata for the share group.
     * @return An optional representing the persister initialize request.
     */
    private Optional<InitializeShareGroupStateParameters> maybeCreateInitializeShareGroupStateRequest(
        String groupId,
        int groupEpoch,
        Map<String, TopicMetadata> subscriptionMetadata,
        List<CoordinatorRecord> records
    ) {
        if (subscriptionMetadata == null || subscriptionMetadata.isEmpty() || metadataImage.isEmpty()) {
            return Optional.empty();

Review comment (suggestion):
This is a bit misleading. The delta is actually milliseconds I think. I suggest not even referring to the units at all. The point is to avoid duplicating requests within the delta, giving the opportunity for retries but minimising the risk of collisions which cause state epoch fencing.

Code after:
    // Visibility for testing
    Map<Uuid, InitMapValue> subscribedTopicsChangeMap(String groupId, Map<String, TopicMetadata> subscriptionMetadata) {
        if (subscriptionMetadata == null || subscriptionMetadata.isEmpty()) {
            return Map.of();
        }

        Map<Uuid, InitMapValue> topicPartitionChangeMap = new HashMap<>();
        ShareGroupStatePartitionMetadataInfo info = shareGroupStatePartitionMetadata.get(groupId);

        // We must only consider initializing topics whose timestamp is fresher than delta elapsed.
        // Any initializing topics which are older than delta and are part of the subscribed topics
        // must be returned so that they can be retried.
        long curTimestamp = time.milliseconds();
        long delta = config.offsetCommitTimeoutMs() * 2L;
        Map<Uuid, InitMapValue> alreadyInitialized = info == null ? new HashMap<>() :
            combineInitMaps(
                info.initializedTopics(),
                info.initializingTopics().entrySet().stream()
                    .filter(entry -> curTimestamp - entry.getValue().timestamp() < delta)
                    .collect(Collectors.toMap(Map.Entry::getKey, Map.Entry::getValue))
            );

        // Here will add any topics which are subscribed but not initialized and initializing
        // topics whose timestamp indicates that they are older than delta elapsed.
        subscriptionMetadata.forEach((topicName, topicMetadata) -> {
            Set<Integer> alreadyInitializedPartSet = alreadyInitialized.containsKey(topicMetadata.id()) ? alreadyInitialized.get(topicMetadata.id()).partitions() : Set.of();
            if (alreadyInitializedPartSet.isEmpty() || alreadyInitializedPartSet.size() < topicMetadata.numPartitions()) {
                Set<Integer> partitionSet = IntStream.range(0, topicMetadata.numPartitions()).boxed().collect(Collectors.toSet());
                partitionSet.removeAll(alreadyInitializedPartSet);
                // alreadyInitialized contains all initialized topics and initializing topics which are less than delta old
                // which means we are putting subscribed topics which are unseen or initializing for more than delta. But, we
                // are also updating the timestamp here which means, old initializing will not be included repeatedly.
                topicPartitionChangeMap.computeIfAbsent(topicMetadata.id(), k -> new InitMapValue(topicMetadata.name(), partitionSet, curTimestamp));
            }
        });
        return topicPartitionChangeMap;
    }

    /**
     * Based on the diff between the subscribed topic partitions and the initialized topic partitions,
     * created initialize request for the non-initialized ones.
     *
     * @param groupId               The share group id for which partitions need to be initialized.
     * @param groupEpoch            The group epoch of the share group.
     * @param subscriptionMetadata  The subscription metadata for the share group.
     * @return An optional representing the persister initialize request.
     */
    private Optional<InitializeShareGroupStateParameters> maybeCreateInitializeShareGroupStateRequest(
        String groupId,
        int groupEpoch,
        Map<String, TopicMetadata> subscriptionMetadata,