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

Code before:
                .map(epoch -> {
                    OffsetFetchRequestData data = new OffsetFetchRequestData()
                        .setRequireStable(true)
                        .setGroups(List.of(
                            new OffsetFetchRequestData.OffsetFetchRequestGroup()
                                .setGroupId(groupId)
                                .setMemberId(memberInfo.memberId)
                                .setMemberEpoch(epoch)
                                .setTopics(topics)));
                    return finalCanUseTopicIds
                        ? OffsetFetchRequest.Builder.forTopicIdsOrNames(data, throwOnFetchStableOffsetUnsupported, true)
                        : OffsetFetchRequest.Builder.forTopicNames(data, throwOnFetchStableOffsetUnsupported);
                })
                // Building request without passing member ID/epoch to leave the logic to choose
                // default values when not present on the request builder.
                .orElseGet(() -> {
                    OffsetFetchRequestData data = new OffsetFetchRequestData()
                        .setRequireStable(true)
                        .setGroups(List.of(
                            new OffsetFetchRequestData.OffsetFetchRequestGroup()
                                .setGroupId(groupId)
                                .setTopics(topics)));
                    return finalCanUseTopicIds
                        ? OffsetFetchRequest.Builder.forTopicIdsOrNames(data, throwOnFetchStableOffsetUnsupported, true)
                        : OffsetFetchRequest.Builder.forTopicNames(data, throwOnFetchStableOffsetUnsupported);
                });
            return buildRequestWithResponseHandling(builder);
        }

        /**
         * Handle OffsetFetch response, including successful and failed.
         */
        @Override
        void onResponse(final ClientResponse response) {
            long currentTimeMs = response.receivedTimeMs();
            var fetchResponse = (OffsetFetchResponse) response.responseBody();
            var groupResponse = fetchResponse.group(groupId);
            var error = Errors.forCode(groupResponse.errorCode());
            if (error != Errors.NONE) {
                onFailure(currentTimeMs, error);
                return;
            }
            onSuccess(currentTimeMs, groupResponse);
        }

        /**
         * Handle failed responses. This will retry if the error is retriable, or complete the
         * result future exceptionally in the case of non-recoverable or unexpected errors.
         */
        private void onFailure(final long currentTimeMs,
                               final Errors responseError) {

Review comment (refactor):
this whole section is getting too complicated, let's try to simplify.

We just want to create the builder, the only diff seems to be around the group data (with epoch or not). So what about extracting that bit and leaving the rest as a common code, no dups? Something along these lines:

```
            OffsetFetchRequestData.OffsetFetchRequestGroup groupData = new OffsetFetchRequestData.OffsetFetchRequestGroup()
                    .setGroupId(groupId)
                    .setTopics(topics);
            if (memberInfo.memberEpoch.isPresent()) {
                groupData = groupData.setMemberId(memberInfo.memberId)
                        .setMemberEpoch(memberInfo.memberEpoch.get());
            }

            OffsetFetchRequestData data = new OffsetFetchRequestData()
                    .setRequireStable(true)
                    .setGroups(List.of(groupData));
            OffsetFetchRequest.Builder builder = finalCanUseTopicIds
                    ? OffsetFetchRequest.Builder.forTopicIdsOrNames(data, throwOnFetchStableOffsetUnsupported, true)
                    : OffsetFetchRequest.Builder.forTopicNames(data, throwOnFetchStableOffsetUnsupported);
```
 
Would that work? Not sure if I'm missing any other diff

Code after:
            Map<String, Uuid> topicIds = metadata.topicIds();
            boolean canUseTopicIds = true;
            List<OffsetFetchRequestData.OffsetFetchRequestTopics> topics = new ArrayList<>();
            Map<String, List<TopicPartition>> tps = requestedPartitions.stream().collect(Collectors.groupingBy(TopicPartition::topic));
            for (Map.Entry<String, List<TopicPartition>> entry : tps.entrySet()) {
                String topic = entry.getKey();
                Uuid topicId = topicIds.getOrDefault(topic, Uuid.ZERO_UUID);
                if (Uuid.ZERO_UUID.equals(topicId)) {
                    canUseTopicIds = false;
                }
                topics.add(new OffsetFetchRequestData.OffsetFetchRequestTopics()
                    .setName(entry.getKey())
                    .setTopicId(topicId)
                    .setPartitionIndexes(entry.getValue().stream()
                        .map(TopicPartition::partition)
                        .collect(Collectors.toList())));
            }

            OffsetFetchRequestData.OffsetFetchRequestGroup groupData = new OffsetFetchRequestData.OffsetFetchRequestGroup()
                .setGroupId(groupId)
                .setTopics(topics);
            if (memberInfo.memberEpoch.isPresent()) {
                groupData = groupData.setMemberId(memberInfo.memberId)
                    .setMemberEpoch(memberInfo.memberEpoch.get());
            }
            OffsetFetchRequestData data = new OffsetFetchRequestData()
                .setRequireStable(true)
                .setGroups(List.of(groupData));
            OffsetFetchRequest.Builder builder = canUseTopicIds
                ? OffsetFetchRequest.Builder.forTopicIdsOrNames(data, throwOnFetchStableOffsetUnsupported, true)
                : OffsetFetchRequest.Builder.forTopicNames(data, throwOnFetchStableOffsetUnsupported);
            return buildRequestWithResponseHandling(builder);
        }

        /**
         * Handle OffsetFetch response, including successful and failed.
         */
        @Override
        void onResponse(final ClientResponse response) {
            long currentTimeMs = response.receivedTimeMs();
            var fetchResponse = (OffsetFetchResponse) response.responseBody();
            var groupResponse = fetchResponse.group(groupId);
            var error = Errors.forCode(groupResponse.errorCode());
            if (error != Errors.NONE) {
                onFailure(currentTimeMs, error);
                return;
            }
            onSuccess(currentTimeMs, groupResponse);
        }

        /**