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

Code before:
            Map<String, Uuid> topicIds = metadata.topicIds();
            boolean canUseTopicIds = true;
            Map<String, OffsetCommitRequestData.OffsetCommitRequestTopic> requestTopicDataMap = new HashMap<>();
            for (Map.Entry<TopicPartition, OffsetAndMetadata> entry : offsets.entrySet()) {
                TopicPartition topicPartition = entry.getKey();
                OffsetAndMetadata offsetAndMetadata = entry.getValue();
                Uuid topicId = topicIds.getOrDefault(topicPartition.topic(), Uuid.ZERO_UUID);
                if (topicId.equals(Uuid.ZERO_UUID)) {
                    canUseTopicIds = false;
                }

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

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

            OffsetCommitRequestData data = new OffsetCommitRequestData()
                    .setGroupId(this.groupId)
                    .setGroupInstanceId(groupInstanceId.orElse(null))
                    .setTopics(new ArrayList<>(requestTopicDataMap.values()));
            data = data.setMemberId(memberInfo.memberId);
            if (memberInfo.memberEpoch.isPresent()) {
                data = data.setGenerationIdOrMemberEpoch(memberInfo.memberEpoch.get());
                lastEpochSentOnCommit = memberInfo.memberEpoch;
            } else {
                lastEpochSentOnCommit = Optional.empty();
            }

            OffsetCommitRequest.Builder builder = canUseTopicIds
                    ? OffsetCommitRequest.Builder.forTopicIdsOrNames(data, true)
                    : OffsetCommitRequest.Builder.forTopicNames(data);

            return buildRequestWithResponseHandling(builder);
        }

        /**
         * Handle OffsetCommitResponse. This will complete the request future successfully if no
         * errors are found in the response. If the response contains errors, this will:
         *   - handle expected errors and fail the future with specific exceptions depending on the error
         *   - fail the future with a non-recoverable KafkaException for all unexpected errors (even if retriable)

Review comment (bug):
This pair is an interesting bit. It surfaces the gap that here we could be wrongly committing positions for a re-created topic (same topic name, but different topic ID). That is covered with [KAFKA-13447](https://issues.apache.org/jira/browse/KAFKA-13447) still open, but raising it to make sure we're considering it. 

I expect we need to ensure that the topic ID we have here (from metadata) is the same one we got the position from. If it's not the same (topic re-created), then we need to :
- ensure we don't commit the position
- surface error to the user (if commit sync or in callback if async. Swallow/log for auto-commit on the interval or before revocation?)
Makes sense? I guess we want to handle this on the other jira, but closely related. 


Code after:
                                 final MemberInfo memberInfo) {
            super(logContext, CommitRequestManager.class.getSimpleName(), retryBackoffMs, 2,
                retryBackoffMaxMs, jitter, memberInfo, deadlineTimer(time, deadlineMs));
            this.offsets = offsets;
            this.groupId = groupId;
            this.groupInstanceId = groupInstanceId;
            this.future = new CompletableFuture<>();
        }

        public NetworkClientDelegate.UnsentRequest toUnsentRequest() {
            Map<String, Uuid> topicIds = metadata.topicIds();
            boolean canUseTopicIds = true;
            Map<String, OffsetCommitRequestData.OffsetCommitRequestTopic> requestTopicDataMap = new HashMap<>();
            for (Map.Entry<TopicPartition, OffsetAndMetadata> entry : offsets.entrySet()) {
                TopicPartition topicPartition = entry.getKey();
                OffsetAndMetadata offsetAndMetadata = entry.getValue();
                Uuid topicId = topicIds.getOrDefault(topicPartition.topic(), Uuid.ZERO_UUID);
                if (topicId.equals(Uuid.ZERO_UUID)) {
                    canUseTopicIds = false;
                }

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

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

            OffsetCommitRequestData data = new OffsetCommitRequestData()
                    .setGroupId(this.groupId)
                    .setGroupInstanceId(groupInstanceId.orElse(null))
                    .setTopics(new ArrayList<>(requestTopicDataMap.values()));
            data = data.setMemberId(memberInfo.memberId);
            if (memberInfo.memberEpoch.isPresent()) {
                data = data.setGenerationIdOrMemberEpoch(memberInfo.memberEpoch.get());
                lastEpochSentOnCommit = memberInfo.memberEpoch;
            } else {
                lastEpochSentOnCommit = Optional.empty();
            }

            OffsetCommitRequest.Builder builder = canUseTopicIds
                    ? OffsetCommitRequest.Builder.forTopicIdsOrNames(data, true)