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

Code before:
     *     <li>trigger an async request to validate positions (detect log truncation)</li>
     *     <li>fetch committed offsets if enabled, and use the response to update the positions</li>
     *     <li>fetch partition offsets for partitions that may still require a position, and use the response to
     *     update the positions</li>
     * </ul>
     *
     * @param deadlineMs Time in milliseconds when the triggering application event expires. Any error received after
     *                   this will be saved, and used to complete the result exceptionally on the next call to this
     *                   function.
     * @return Future that will complete with a boolean indicating if all assigned partitions have positions (based
     * on {@link SubscriptionState#hasAllFetchPositions()}). It will complete immediately, with true, if all positions
     * are already available. If some positions are missing, the future will complete once the offsets are retrieved and positions are updated.
     */
    public CompletableFuture<Boolean> updateFetchPositions(long deadlineMs) {
        CompletableFuture<Boolean> result = new CompletableFuture<>();

        try {
            if (maybeCompleteWithPreviousException(result)) {
                return result;
            }

            validatePositionsIfNeeded();

            if (subscriptionState.hasAllFetchPositions()) {
                // All positions are already available
                commitOffsetsSharedState.setSubscriptionHasAllFetchPositions(true);
                result.complete(true);
                return result;
            } else {
                commitOffsetsSharedState.setSubscriptionHasAllFetchPositions(false);
            }

            // Some positions are missing, so trigger requests to fetch offsets and update them.
            updatePositionsWithOffsets(deadlineMs).whenComplete((__, error) -> {
                if (error != null) {
                    result.completeExceptionally(error);
                } else {
                    result.complete(subscriptionState.hasAllFetchPositions());
                }
            });

        } catch (Exception e) {
            result.completeExceptionally(maybeWrapAsKafkaException(e));
        }
        return result;
    }

    private boolean maybeCompleteWithPreviousException(CompletableFuture<Boolean> result) {
        Throwable cachedException = commitOffsetsSharedState.getAndClearCachedUpdatePositionsException();
        if (cachedException != null) {
            result.completeExceptionally(cachedException);

Review comment (bug):
uhm updating the cached "subscriptionHasAllPositions" here will run before fetching. Wouldn't we have an outdated cache if the positions become invalid after the fetch result? (ex. fetch out of range that will reset offsets). 

In such cases, I imagine the positions would become invalid after this point, when processing the fetch result, but the cache will remain with hasAllPositions true and wrongly skip `updateFetchPositions` on the next poll. Am I missing something? I see we only update the cache here and when the assignment changes. 

Code after:
                        result.fetchedOffsets));
    }

    /**
     * Update fetch positions for assigned partitions that do not have a position. This will:
     * <ul>
     *     <li>check if all assigned partitions already have fetch positions and return right away if that's the case</li>
     *     <li>trigger an async request to validate positions (detect log truncation)</li>
     *     <li>fetch committed offsets if enabled, and use the response to update the positions</li>
     *     <li>fetch partition offsets for partitions that may still require a position, and use the response to
     *     update the positions</li>
     * </ul>
     *
     * @param deadlineMs Time in milliseconds when the triggering application event expires. Any error received after
     *                   this will be saved, and used to complete the result exceptionally on the next call to this
     *                   function.
     * @return Future that will complete with a boolean indicating if all assigned partitions have positions (based
     * on {@link SubscriptionState#hasAllFetchPositions()}). It will complete immediately, with true, if all positions
     * are already available. If some positions are missing, the future will complete once the offsets are retrieved and positions are updated.
     */
    public CompletableFuture<Void> updateFetchPositions(long deadlineMs) {
        CompletableFuture<Void> result = new CompletableFuture<>();

        try {
            if (maybeCompleteWithPreviousException(result)) {
                return result;
            }

            validatePositionsIfNeeded();

            if (subscriptionState.hasAllFetchPositions()) {
                // All positions are already available
                result.complete(null);
                return result;
            }

            // Some positions are missing, so trigger requests to fetch offsets and update them.
            updatePositionsWithOffsets(deadlineMs).whenComplete((__, error) -> {
                if (error != null) {
                    result.completeExceptionally(error);
                } else {
                    result.complete(null);
                }
            });

        } catch (Exception e) {
            result.completeExceptionally(maybeWrapAsKafkaException(e));
        }
        return result;
    }