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

Code before:
        }
        this.coordinatorRequestManager = coordinatorRequestManager;
        this.groupId = groupId;
        this.groupInstanceId = groupInstanceId;
        this.subscriptions = subscriptions;
        this.metadata = metadata;
        this.retryBackoffMs = retryBackoffMs;
        this.retryBackoffMaxMs = retryBackoffMaxMs;
        this.jitter = jitter;
        this.throwOnFetchStableOffsetUnsupported = config.getBoolean(THROW_ON_FETCH_STABLE_OFFSET_UNSUPPORTED);
        this.memberInfo = new MemberInfo();
        this.metricsManager = new OffsetCommitMetricsManager(metrics);
        this.offsetCommitCallbackInvoker = offsetCommitCallbackInvoker;
        this.lastEpochSentOnCommit = Optional.empty();
    }

    /**
     * Poll for the {@link OffsetFetchRequest} and {@link OffsetCommitRequest} request if there's any.
     */
    @Override
    public NetworkClientDelegate.PollResult poll(final long currentTimeMs) {
        if (coordinatorRequestManager.coordinator().isEmpty() && closing) {
            CommitFailedException exception = new CommitFailedException(
                    "Failed to commit offsets: Coordinator unknown and consumer is closing");

            List<NetworkClientDelegate.UnsentRequest> pendingCommits = pendingRequests.drainPendingCommits();

            // Complete each commit request exceptionally
            pendingCommits.forEach(request -> {
                if (request.future() != null) {
                    request.future().completeExceptionally(exception);
                }
            });

            pendingRequests.clearAll();
            return EMPTY;
        }

        // poll when the coordinator node is known and fatal error is not present
        if (coordinatorRequestManager.coordinator().isEmpty()) {
            pendingRequests.maybeFailOnCoordinatorFatalError();
            return EMPTY;
        }

        if (closing) {
            return drainPendingOffsetCommitRequests();
        }

        if (!pendingRequests.hasUnsentRequests())
            return EMPTY;


Review comment (bug):
We could inline this variable.
`pendingRequests.drainPendingCommits().forEach...`


Code after:
        }
        this.coordinatorRequestManager = coordinatorRequestManager;
        this.groupId = groupId;
        this.groupInstanceId = groupInstanceId;
        this.subscriptions = subscriptions;
        this.metadata = metadata;
        this.retryBackoffMs = retryBackoffMs;
        this.retryBackoffMaxMs = retryBackoffMaxMs;
        this.jitter = jitter;
        this.throwOnFetchStableOffsetUnsupported = config.getBoolean(THROW_ON_FETCH_STABLE_OFFSET_UNSUPPORTED);
        this.memberInfo = new MemberInfo();
        this.metricsManager = new OffsetCommitMetricsManager(metrics);
        this.offsetCommitCallbackInvoker = offsetCommitCallbackInvoker;
        this.lastEpochSentOnCommit = Optional.empty();
    }

    /**
     * Poll for the {@link OffsetFetchRequest} and {@link OffsetCommitRequest} request if there's any.
     */
    @Override
    public NetworkClientDelegate.PollResult poll(final long currentTimeMs) {
        // poll when the coordinator node is known and fatal error is not present
        if (coordinatorRequestManager.coordinator().isEmpty()) {
            pendingRequests.maybeFailOnCoordinatorFatalError();

            if (closing && pendingRequests.hasUnsentRequests()) {
                CommitFailedException exception = new CommitFailedException(
                        "Failed to commit offsets: Coordinator unknown and consumer is closing");
                pendingRequests.drainPendingCommits()
                        .forEach(request -> request.future().completeExceptionally(exception));
            }

            return EMPTY;
        }

        if (closing) {
            return drainPendingOffsetCommitRequests();
        }

        if (!pendingRequests.hasUnsentRequests())
            return EMPTY;

        List<NetworkClientDelegate.UnsentRequest> requests = pendingRequests.drain(currentTimeMs);
        // min of the remainingBackoffMs of all the request that are still backing off
        final long timeUntilNextPoll = Math.min(
            findMinTime(unsentOffsetCommitRequests(), currentTimeMs),
            findMinTime(unsentOffsetFetchRequests(), currentTimeMs));
        return new NetworkClientDelegate.PollResult(timeUntilNextPoll, requests);
    }

    @Override