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

Code before:
         *   - 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)
         */
        @Override
        @SuppressWarnings("NPathComplexity")
        public void onResponse(final ClientResponse response) {
            metricsManager.recordRequestLatency(response.requestLatencyMs());
            long currentTimeMs = response.receivedTimeMs();
            OffsetCommitResponse commitResponse = (OffsetCommitResponse) response.responseBody();
            Set<String> unauthorizedTopics = new HashSet<>();
            boolean failedRequestRegistered = false;
            for (OffsetCommitResponseData.OffsetCommitResponseTopic topic : commitResponse.data().topics()) {
                for (OffsetCommitResponseData.OffsetCommitResponsePartition partition : topic.partitions()) {
                    // Version 10 drop topic name and support to topic id.
                    // We need to find offsetAndMetadata based on topic id and partition index only as
                    // topic name in the response will be emtpy.
                    // For older versions, topic id is zero, and we will find the offsetAndMetadata based on the topic name.
                    TopicPartition tp = (!Uuid.ZERO_UUID.equals(topic.topicId()) && metadata.topicNames().containsKey(topic.topicId())) ?
                        new TopicPartition(metadata.topicNames().get(topic.topicId()), partition.partitionIndex()) :
                        new TopicPartition(topic.name(), partition.partitionIndex());

                    Errors error = Errors.forCode(partition.errorCode());
                    if (error == Errors.NONE) {
                        OffsetAndMetadata offsetAndMetadata = offsets.get(tp);
                        if (offsetAndMetadata == null) {
                            throw new IllegalStateException("Can't find metadata for topic id " + topic.topicId() +
                                " topic name " + topic.name() + " partition " + partition.partitionIndex() + " using " + metadata.topicNames());
                        }
                        long offset = offsetAndMetadata.offset();
                        log.debug("OffsetCommit completed successfully for offset {} partition {}", offset, tp);
                        continue;
                    }

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

                    if (error == Errors.GROUP_AUTHORIZATION_FAILED) {
                        future.completeExceptionally(GroupAuthorizationException.forGroupId(groupId));
                        return;
                    } else if (error == Errors.COORDINATOR_NOT_AVAILABLE ||
                        error == Errors.NOT_COORDINATOR ||
                        error == Errors.REQUEST_TIMED_OUT) {
                        coordinatorRequestManager.markCoordinatorUnknown(error.message(), currentTimeMs);
                        future.completeExceptionally(error.exception());
                        return;
                    } else if (error == Errors.OFFSET_METADATA_TOO_LARGE ||
                        error == Errors.INVALID_COMMIT_OFFSET_SIZE) {
                        future.completeExceptionally(error.exception());
                        return;

Review comment (bug):
I'm not convinced that this is what we should do here. At this point, the commit request succeeded on the broker. It's just that the response contains a topic ID that the client cannot resolve, but it only needs the topic name to include it in a `log.debug` really.  

Code after:
                    : 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)
         */
        @Override
        @SuppressWarnings("NPathComplexity")
        public void onResponse(final ClientResponse response) {
            metricsManager.recordRequestLatency(response.requestLatencyMs());
            long currentTimeMs = response.receivedTimeMs();
            OffsetCommitResponse commitResponse = (OffsetCommitResponse) response.responseBody();
            Set<String> unauthorizedTopics = new HashSet<>();
            boolean failedRequestRegistered = false;
            for (OffsetCommitResponseData.OffsetCommitResponseTopic topic : commitResponse.data().topics()) {
                for (OffsetCommitResponseData.OffsetCommitResponsePartition partition : topic.partitions()) {
                    // Version 10 drops topic name, and supports topic id.
                    // We need to find offsetAndMetadata based on topic id and partition index only as
                    // topic name in the response will be emtpy.
                    // For older versions, topic id is zero, and we will find the offsetAndMetadata based on the topic name.
                    TopicPartition tp = (!Uuid.ZERO_UUID.equals(topic.topicId()) && metadata.topicNames().containsKey(topic.topicId())) ?
                        new TopicPartition(metadata.topicNames().get(topic.topicId()), partition.partitionIndex()) :
                        new TopicPartition(topic.name(), partition.partitionIndex());

                    Errors error = Errors.forCode(partition.errorCode());
                    if (error == Errors.NONE) {
                        OffsetAndMetadata offsetAndMetadata = offsets.get(tp);
                        if (offsetAndMetadata == null) {
                            log.debug("Can't find metadata for partition {}", tp);
                        } else {
                            log.debug("OffsetCommit completed successfully for offset {} partition {}", offsetAndMetadata.offset(), tp);
                        }
                        continue;
                    }

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

                    if (error == Errors.GROUP_AUTHORIZATION_FAILED) {
                        future.completeExceptionally(GroupAuthorizationException.forGroupId(groupId));
                        return;
                    } else if (error == Errors.COORDINATOR_NOT_AVAILABLE ||
                        error == Errors.NOT_COORDINATOR ||
                        error == Errors.REQUEST_TIMED_OUT) {