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

Code before:
                    } else if (error == Errors.STALE_MEMBER_EPOCH) {
                        log.error("OffsetCommit failed for member {} with stale member epoch error. Last epoch sent: {}",
                            memberInfo.memberId,
                            lastEpochSentOnCommit.isPresent() ? lastEpochSentOnCommit.get() : "undefined");
                        future.completeExceptionally(error.exception());
                        return;
                    } else if (error == Errors.TOPIC_AUTHORIZATION_FAILED) {
                        // Collect all unauthorized topics before failing
                        unauthorizedTopics.add(tp.topic());
                    } else if (error == Errors.UNKNOWN_TOPIC_ID) {
                        // Collect all unknown topic ids
                        unknownTopicIds.add(topic.topicId());
                    } else {
                        // Fail with a non-retriable KafkaException for all unexpected errors
                        // (even if they are retriable)
                        future.completeExceptionally(new KafkaException("Unexpected error in commit: " + error.message()));
                        return;
                    }
                }
            }

            if (!unauthorizedTopics.isEmpty()) {
                log.error("OffsetCommit failed due to not authorized to commit to topics {}", unauthorizedTopics);
                future.completeExceptionally(new TopicAuthorizationException(unauthorizedTopics));
            } else if (!unknownTopicIds.isEmpty()) {
                log.error("OffsetCommit failed due to unknown topic id to commit to topic ids {}", unknownTopicIds);
                future.completeExceptionally(new UnknownTopicIdException(Errors.UNKNOWN_TOPIC_ID.message()));
            } else {
                future.complete(null);
            }
        }

        @Override
        String requestDescription() {
            return "OffsetCommit request for offsets " + offsets;
        }

        @Override
        CompletableFuture<?> future() {
            return future;
        }

        void resetFuture() {
            future = new CompletableFuture<>();
        }

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

Review comment (bug):
the message reads a bit off, maybe simply `OffsetCommit failed due to unknown topic IDs {}` would do?

Code after:
                        future.completeExceptionally(error.exception());
                        return;
                    } else if (error == Errors.OFFSET_METADATA_TOO_LARGE ||
                        error == Errors.INVALID_COMMIT_OFFSET_SIZE) {
                        future.completeExceptionally(error.exception());
                        return;
                    } else if (error == Errors.COORDINATOR_LOAD_IN_PROGRESS ||
                        error == Errors.UNKNOWN_TOPIC_OR_PARTITION ||
                        error == Errors.UNKNOWN_TOPIC_ID) {
                        // just retry
                        future.completeExceptionally(error.exception());
                        return;
                    } else if (error == Errors.UNKNOWN_MEMBER_ID) {
                        log.error("OffsetCommit failed with {}", error);
                        future.completeExceptionally(new CommitFailedException("OffsetCommit " +
                            "failed with unknown member ID. " + error.message()));
                        return;
                    } else if (error == Errors.STALE_MEMBER_EPOCH) {
                        log.error("OffsetCommit failed for member {} with stale member epoch error. Last epoch sent: {}",
                            memberInfo.memberId,
                            lastEpochSentOnCommit.isPresent() ? lastEpochSentOnCommit.get() : "undefined");
                        future.completeExceptionally(error.exception());
                        return;
                    } else if (error == Errors.TOPIC_AUTHORIZATION_FAILED) {
                        // Collect all unauthorized topics before failing
                        unauthorizedTopics.add(tp.topic());
                    } else {
                        // Fail with a non-retriable KafkaException for all unexpected errors
                        // (even if they are retriable)
                        future.completeExceptionally(new KafkaException("Unexpected error in commit: " + error.message()));
                        return;
                    }
                }
            }

            if (!unauthorizedTopics.isEmpty()) {
                log.error("OffsetCommit failed due to not authorized to commit to topics {}", unauthorizedTopics);
                future.completeExceptionally(new TopicAuthorizationException(unauthorizedTopics));
            } else {
                future.complete(null);
            }
        }

        @Override
        String requestDescription() {
            return "OffsetCommit request for offsets " + offsets;
        }

        @Override
        CompletableFuture<?> future() {
            return future;