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

Code before:
            final IsolationLevel isolationLevel = event.isolationLevel();
            final Long lag = subscriptions.partitionLag(topicPartition, isolationLevel);

            final OptionalLong lagOpt;
            if (lag == null) {
                if (subscriptions.partitionEndOffset(topicPartition, isolationLevel) == null) {
                    // The LIST_OFFSETS lag lookup is serialized, so if there's an inflight request it must
                    // finish before another request can be issued. This serialization mechanism is controlled
                    // by the 'end offset requested' flag in SubscriptionState.
                    if (subscriptions.partitionEndOffsetRequested(topicPartition)) {
                        log.info("Not requesting the log end offset for {} to compute lag as an outstanding request already exists", topicPartition);
                    } else {
                        // If the log end offset is unknown and there isn't already an in-flight list offset
                        // request, issue one with the goal that the lag will be available the next time the
                        // user calls currentLag().
                        log.info("Requesting the log end offset for {} in order to compute lag", topicPartition);
                        subscriptions.requestPartitionEndOffset(topicPartition);

                        // Emulates the Consumer.endOffsets() logic...
                        Map<TopicPartition, Long> timestampToSearch = Collections.singletonMap(
                            topicPartition,
                            ListOffsetsRequest.LATEST_TIMESTAMP
                        );

                        // The currentLag() API is a "best effort" attempt at calling the LIST_OFFSETS RPC. If it
                        // fails, don't retry the attempt internally, but let the user attempt it again.
                        requestManagers.offsetsRequestManager.fetchOffsets(
                            timestampToSearch,
                            false,
                            false
                        );
                    }
                }

                lagOpt = OptionalLong.empty();
            } else {
                lagOpt = OptionalLong.of(lag);
            }

            event.future().complete(lagOpt);
        } catch (Exception e) {
            event.future().completeExceptionally(e);
        }
    }

    private void process(final StreamsOnTasksRevokedCallbackCompletedEvent event) {
        if (requestManagers.streamsMembershipManager.isEmpty()) {
            log.warn("An internal error occurred; the Streams membership manager was not present, so the notification " +
                "of the onTasksRevoked callback execution could not be sent");
            return;
        }

Review comment (bug):
here is were I think we may be getting something wrong. Agree we don't want to retry the ListOffset triggered from currentLag if it fails. But we do want to retry discovering a leader in order to be able to send the request (which is what the internal retries I pointed in the comment above do).

So if this retry=false is only used to skip that leader re-discovery retry, I think we should remove it (no param needed probably?)

Code after:
            final OptionalLong lagOpt;
            if (lag == null) {
                if (subscriptions.partitionEndOffset(topicPartition, isolationLevel) == null &&
                    !subscriptions.partitionEndOffsetRequested(topicPartition)) {
                    // If the log end offset is unknown and there isn't already an in-flight list offset
                    // request, issue one with the goal that the lag will be available the next time the
                    // user calls currentLag().
                    log.info("Requesting the log end offset for {} in order to compute lag", topicPartition);
                    subscriptions.requestPartitionEndOffset(topicPartition);

                    // Emulates the Consumer.endOffsets() logic...
                    Map<TopicPartition, Long> timestampToSearch = Collections.singletonMap(
                        topicPartition,
                        ListOffsetsRequest.LATEST_TIMESTAMP
                    );

                    requestManagers.offsetsRequestManager.fetchOffsets(timestampToSearch, false);
                }

                lagOpt = OptionalLong.empty();
            } else {
                lagOpt = OptionalLong.of(lag);
            }

            event.future().complete(lagOpt);
        } catch (Exception e) {
            event.future().completeExceptionally(e);
        }
    }

    private void process(final StreamsOnTasksRevokedCallbackCompletedEvent event) {
        if (requestManagers.streamsMembershipManager.isEmpty()) {
            log.warn("An internal error occurred; the Streams membership manager was not present, so the notification " +
                "of the onTasksRevoked callback execution could not be sent");
            return;
        }
        requestManagers.streamsMembershipManager.get().onTasksRevokedCallbackCompleted(event);
    }

    private void process(final StreamsOnTasksAssignedCallbackCompletedEvent event) {
        if (requestManagers.streamsMembershipManager.isEmpty()) {
            log.warn("An internal error occurred; the Streams membership manager was not present, so the notification " +
                "of the onTasksAssigned callback execution could not be sent");
            return;
        }
        requestManagers.streamsMembershipManager.get().onTasksAssignedCallbackCompleted(event);
    }

    private void process(final StreamsOnAllTasksLostCallbackCompletedEvent event) {
        if (requestManagers.streamsMembershipManager.isEmpty()) {
            log.warn("An internal error occurred; the Streams membership manager was not present, so the notification " +