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

Code before:
            future = new CompletableFuture<>();
        }

        @Override
        void removeRequest() {
            if (!unsentOffsetFetchRequests().remove(this)) {
                log.warn("OffsetFetch request to remove not found in the outbound buffer: {}", this);
            }
        }

        /**
         * Handle OffsetFetch response that has no group level errors. This will look for
         * partition level errors and fail the future accordingly, also recording a failed request
         * attempt. If no partition level errors are found, this will complete the future with the
         * offsets contained in the response, and record a successful request attempt.
         */
        private void onSuccess(final long currentTimeMs,
                               final OffsetFetchResponseData.OffsetFetchResponseGroup response) {
            var offsets = new HashMap<TopicPartition, OffsetAndMetadata>();
            var unstableTxnOffsetTopicPartitions = new HashSet<TopicPartition>();
            var unauthorizedTopics = new HashSet<String>();
            var failedRequestRegistered = false;

            for (var topic : response.topics()) {
                // If the topic id is used, the topic name is empty in the response.
                String topicName = topic.name().isEmpty() ? metadata.topicNames().get(topic.topicId()) : topic.name();
                for (var partition : topic.partitions()) {
                    var tp = new TopicPartition(
                        topicName,
                        partition.partitionIndex()
                    );
                    var error = Errors.forCode(partition.errorCode());
                    if (error != Errors.NONE) {
                        log.debug("Failed to fetch offset for partition {}: {}", tp, error.message());

                        if (!failedRequestRegistered) {
                            onFailedAttempt(currentTimeMs);
                            failedRequestRegistered = true;
                        }

                        if (error == Errors.UNKNOWN_TOPIC_OR_PARTITION || error == Errors.UNKNOWN_TOPIC_ID) {
                            future.completeExceptionally(new KafkaException("Topic or Partition " + tp + " does not exist"));
                            return;
                        } else if (error == Errors.TOPIC_AUTHORIZATION_FAILED) {
                            unauthorizedTopics.add(tp.topic());
                        } else if (error == Errors.UNSTABLE_OFFSET_COMMIT) {
                            unstableTxnOffsetTopicPartitions.add(tp);
                        } else {
                            // Fail with a non-retriable KafkaException for all unexpected partition
                            // errors (even if they are retriable)
                            future.completeExceptionally(new KafkaException("Unexpected error in fetch offset " +

Review comment (bug):
after this, we could still have empty `topicName` if the `topicID` from the response is not in the metadata anymore right? And in this case of OffsetFetch that would be problematic because we do need a topic name to return along with the offsets. 

One option that comes to mind is to fail in the same way we do when we get `UNKNOWN_TOPIC_ID` , failing the future with "topic does not exist" (ln 1143). wdyt? 

Code after:
            future = new CompletableFuture<>();
        }

        @Override
        void removeRequest() {
            if (!unsentOffsetFetchRequests().remove(this)) {
                log.warn("OffsetFetch request to remove not found in the outbound buffer: {}", this);
            }
        }

        /**
         * Handle OffsetFetch response that has no group level errors. This will look for
         * partition level errors and fail the future accordingly, also recording a failed request
         * attempt. If no partition level errors are found, this will complete the future with the
         * offsets contained in the response, and record a successful request attempt.
         */
        private void onSuccess(final long currentTimeMs,
                               final OffsetFetchResponseData.OffsetFetchResponseGroup response) {
            var offsets = new HashMap<TopicPartition, OffsetAndMetadata>();
            var unstableTxnOffsetTopicPartitions = new HashSet<TopicPartition>();
            var unauthorizedTopics = new HashSet<String>();
            var failedRequestRegistered = false;

            for (var topic : response.topics()) {
                // If the topic id is used, the topic name is empty in the response.
                String topicName = topic.name().isEmpty() ? metadata.topicNames().get(topic.topicId()) : topic.name();
                for (var partition : topic.partitions()) {
                    var tp = new TopicPartition(
                        topicName,
                        partition.partitionIndex()
                    );
                    var error = Errors.forCode(partition.errorCode());
                    if (error != Errors.NONE || topicName == null) {
                        if (error != Errors.NONE) {
                            log.debug("Failed to fetch offset for partition {}: {}", tp, error.message());
                        } else { // unknown topic name
                            log.debug("Failed to fetch offset, topic does not exist");
                        }

                        if (!failedRequestRegistered) {
                            onFailedAttempt(currentTimeMs);
                            failedRequestRegistered = true;
                        }

                        if (error == Errors.UNKNOWN_TOPIC_OR_PARTITION || error == Errors.UNKNOWN_TOPIC_ID || topicName == null) {
                            future.completeExceptionally(new KafkaException("Topic does not exist"));
                            return;
                        } else if (error == Errors.TOPIC_AUTHORIZATION_FAILED) {
                            unauthorizedTopics.add(tp.topic());
                        } else if (error == Errors.UNSTABLE_OFFSET_COMMIT) {
                            unstableTxnOffsetTopicPartitions.add(tp);