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

Code before:
        } else {
            this.autoCommitState = Optional.empty();
        }
        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) {
            handleClosingWithoutCoordinator();
            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;

        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
    public void signalClose() {

Review comment (bug):
Perhaps it could be written like this for better readability:
```suggestion
            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));
            }
```

Code after:
        } else {
            this.autoCommitState = Optional.empty();
        }
        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);
    }