Repository: apache/kafka
File: core/src/main/java/kafka/server/share/DelayedShareFetch.java (Java)

Code before:
     * This function checks if the remote fetch can be completed or not. It should always be called once you confirm pendingRemoteFetchesOpt.isPresent().
     * The operation can be completed if:
     * Case a: The partition is in an offline log directory on this broker
     * Case b: This broker does not know the partition it tries to fetch
     * Case c: This broker is no longer the leader of the partition it tries to fetch
     * Case d: This broker is no longer the leader or follower of the partition it tries to fetch
     * Case e: All remote storage read requests completed
     * @return boolean representing whether the remote fetch is completed or not.
     */
    private boolean maybeCompletePendingRemoteFetch() {
        boolean canComplete = false;

        for (TopicIdPartition topicIdPartition : pendingRemoteFetchesOpt.get().fetchOffsetMetadataMap().keySet()) {
            try {
                Partition partition = replicaManager.getPartitionOrException(topicIdPartition.topicPartition());
                if (!partition.isLeader()) {
                    throw new NotLeaderException("Broker is no longer the leader of topicPartition: " + topicIdPartition);
                }
            } catch (KafkaStorageException e) { // Case a
                log.debug("TopicPartition {} is in an offline log directory, satisfy {} immediately", topicIdPartition, shareFetch.fetchParams());
                canComplete = true;
            } catch (UnknownTopicOrPartitionException e) { // Case b
                log.debug("Broker no longer knows of topicPartition {}, satisfy {} immediately", topicIdPartition, shareFetch.fetchParams());
                canComplete = true;
            } catch (NotLeaderException e) { // Case c
                log.debug("Broker is no longer the leader or follower of topicPartition {}, satisfy {} immediately", topicIdPartition, shareFetch.fetchParams());
                canComplete = true;
            } catch (NotLeaderOrFollowerException e) { // Case d
                log.debug("Broker is no longer the leader of topicPartition {}, satisfy {} immediately", topicIdPartition, shareFetch.fetchParams());
                canComplete = true;
            }
            if (canComplete)
                break;
        }

        if (canComplete || pendingRemoteFetchesOpt.get().isDone()) { // Case e
            return forceComplete();
        } else
            return false;
    }

    /**
     * This function completes a share fetch request for which we have identified erroneous remote storage fetch in tryComplete()
     * It should only be called when we know that there is remote fetch in-flight/completed.
     */
    private void completeErroneousRemoteShareFetchRequest() {
        try {
            handleFetchException(shareFetch, partitionsAcquired.keySet(), remoteStorageFetchException.get());
        } finally {
            releasePartitionLocksAndAddToActionQueue(partitionsAcquired.keySet());
        }

Review comment (bug):
The text is incorrect. It should be for `NotLeaderOrFollowerException`.

Code after:
     * This function checks if the remote fetch can be completed or not. It should always be called once you confirm pendingRemoteFetchesOpt.isPresent().
     * The operation can be completed if:
     * Case a: The partition is in an offline log directory on this broker
     * Case b: This broker does not know the partition it tries to fetch
     * Case c: This broker is no longer the leader of the partition it tries to fetch
     * Case d: This broker is no longer the leader or follower of the partition it tries to fetch
     * Case e: All remote storage read requests completed
     * @return boolean representing whether the remote fetch is completed or not.
     */
    private boolean maybeCompletePendingRemoteFetch() {
        boolean canComplete = false;

        for (TopicIdPartition topicIdPartition : pendingRemoteFetchesOpt.get().fetchOffsetMetadataMap().keySet()) {
            try {
                Partition partition = replicaManager.getPartitionOrException(topicIdPartition.topicPartition());
                if (!partition.isLeader()) {
                    throw new NotLeaderException("Broker is no longer the leader of topicPartition: " + topicIdPartition);
                }
            } catch (KafkaStorageException e) { // Case a
                log.debug("TopicPartition {} is in an offline log directory, satisfy {} immediately", topicIdPartition, shareFetch.fetchParams());
                canComplete = true;
            } catch (UnknownTopicOrPartitionException e) { // Case b
                log.debug("Broker no longer knows of topicPartition {}, satisfy {} immediately", topicIdPartition, shareFetch.fetchParams());
                canComplete = true;
            } catch (NotLeaderException e) { // Case c
                log.debug("Broker is no longer the leader of topicPartition {}, satisfy {} immediately", topicIdPartition, shareFetch.fetchParams());
                canComplete = true;
            } catch (NotLeaderOrFollowerException e) { // Case d
                log.debug("Broker is no longer the leader or follower of topicPartition {}, satisfy {} immediately", topicIdPartition, shareFetch.fetchParams());
                canComplete = true;
            }
            if (canComplete)
                break;
        }

        if (canComplete || pendingRemoteFetchesOpt.get().isDone()) { // Case e
            return forceComplete();
        } else
            return false;
    }

    /**
     * This function completes a share fetch request for which we have identified erroneous remote storage fetch in tryComplete()
     * It should only be called when we know that there is remote fetch in-flight/completed.
     */
    private void completeErroneousRemoteShareFetchRequest() {
        try {
            handleFetchException(shareFetch, partitionsAcquired.keySet(), remoteStorageFetchException.get());
        } finally {
            releasePartitionLocksAndAddToActionQueue(partitionsAcquired.keySet());
        }