Repository: apache/kafka
File: clients/src/main/java/org/apache/kafka/clients/consumer/internals/ConsumerCoordinator.java (Java)

Code before:
        for (Map.Entry<TopicPartition, OffsetAndMetadata> entry : offsets.entrySet()) {
            TopicPartition topicPartition = entry.getKey();
            OffsetAndMetadata offsetAndMetadata = entry.getValue();
            if (offsetAndMetadata.offset() < 0) {
                return RequestFuture.failure(new IllegalArgumentException("Invalid offset: " + offsetAndMetadata.offset()));
            }

            OffsetCommitRequestData.OffsetCommitRequestTopic topic = requestTopicDataMap
                    .getOrDefault(topicPartition.topic(),
                            new OffsetCommitRequestData.OffsetCommitRequestTopic()
                                    .setName(topicPartition.topic())
                    );

            topic.partitions().add(new OffsetCommitRequestData.OffsetCommitRequestPartition()
                    .setPartitionIndex(topicPartition.partition())
                    .setCommittedOffset(offsetAndMetadata.offset())
                    .setCommittedLeaderEpoch(offsetAndMetadata.leaderEpoch().orElse(RecordBatch.NO_PARTITION_LEADER_EPOCH))
                    .setCommittedMetadata(offsetAndMetadata.metadata())
            );
            requestTopicDataMap.put(topicPartition.topic(), topic);
        }

        final Generation generation;
        final String groupInstanceId;
        if (subscriptions.hasAutoAssignedPartitions()) {
            generation = generationIfStable();
            groupInstanceId = rebalanceConfig.groupInstanceId.orElse(null);
            // if the generation is null, we are not part of an active group (and we expect to be).
            // the only thing we can do is fail the commit and let the user rejoin the group in poll().
            if (generation == null) {
                log.info("Failing OffsetCommit request since the consumer is not part of an active group");

                if (rebalanceInProgress()) {
                    // if the client knows it is already rebalancing, we can use RebalanceInProgressException instead of
                    // CommitFailedException to indicate this is not a fatal error
                    return RequestFuture.failure(new RebalanceInProgressException("Offset commit cannot be completed since the " +
                        "consumer is undergoing a rebalance for auto partition assignment. You can try completing the rebalance " +
                        "by calling poll() and then retry the operation."));
                } else {
                    return RequestFuture.failure(new CommitFailedException("Offset commit cannot be completed since the " +
                        "consumer is not part of an active group for auto partition assignment; it is likely that the consumer " +
                        "was kicked out of the group."));
                }
            }
        } else {
            generation = Generation.NO_GENERATION;
            groupInstanceId = null;
        }

        OffsetCommitRequest.Builder builder = OffsetCommitRequest.Builder.forTopicNames(
                new OffsetCommitRequestData()

Review comment (refactor):
```suggestion
            // (the reason why we add synchronized block?)
            synchronized (ConsumerCoordinator.this) {
```
nit. how about add comment about why we added `synchronized `here?
cause someone can just easily remove this `synchronized `block without comment in the future!

Code after:
        for (Map.Entry<TopicPartition, OffsetAndMetadata> entry : offsets.entrySet()) {
            TopicPartition topicPartition = entry.getKey();
            OffsetAndMetadata offsetAndMetadata = entry.getValue();
            if (offsetAndMetadata.offset() < 0) {
                return RequestFuture.failure(new IllegalArgumentException("Invalid offset: " + offsetAndMetadata.offset()));
            }

            OffsetCommitRequestData.OffsetCommitRequestTopic topic = requestTopicDataMap
                    .getOrDefault(topicPartition.topic(),
                            new OffsetCommitRequestData.OffsetCommitRequestTopic()
                                    .setName(topicPartition.topic())
                    );

            topic.partitions().add(new OffsetCommitRequestData.OffsetCommitRequestPartition()
                    .setPartitionIndex(topicPartition.partition())
                    .setCommittedOffset(offsetAndMetadata.offset())
                    .setCommittedLeaderEpoch(offsetAndMetadata.leaderEpoch().orElse(RecordBatch.NO_PARTITION_LEADER_EPOCH))
                    .setCommittedMetadata(offsetAndMetadata.metadata())
            );
            requestTopicDataMap.put(topicPartition.topic(), topic);
        }

        final Generation generation;
        final String groupInstanceId;
        if (subscriptions.hasAutoAssignedPartitions()) {
            synchronized (ConsumerCoordinator.this) {
                generation = generationIfStable();
                groupInstanceId = rebalanceConfig.groupInstanceId.orElse(null);
                // if the generation is null, we are not part of an active group (and we expect to be).
                // the only thing we can do is fail the commit and let the user rejoin the group in poll().
                if (generation == null) {
                    log.info("Failing OffsetCommit request since the consumer is not part of an active group");

                    if (rebalanceInProgress()) {
                        // if the client knows it is already rebalancing, we can use RebalanceInProgressException instead of
                        // CommitFailedException to indicate this is not a fatal error
                        return RequestFuture.failure(new RebalanceInProgressException("Offset commit cannot be completed since the " +
                            "consumer is undergoing a rebalance for auto partition assignment. You can try completing the rebalance " +
                            "by calling poll() and then retry the operation."));
                    } else {
                        return RequestFuture.failure(new CommitFailedException("Offset commit cannot be completed since the " +
                            "consumer is not part of an active group for auto partition assignment; it is likely that the consumer " +
                            "was kicked out of the group."));
                    }
                }
            }
        } else {
            generation = Generation.NO_GENERATION;
            groupInstanceId = null;
        }