philipnee commented on code in PR #13380:
URL: https://github.com/apache/kafka/pull/13380#discussion_r1139276421
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/CommitRequestManager.java:
##########
@@ -80,16 +97,18 @@ public CommitRequestManager(
*/
@Override
public NetworkClientDelegate.PollResult poll(final long currentTimeMs) {
- maybeAutoCommit(currentTimeMs);
+ maybeAutoCommit();
+ List<NetworkClientDelegate.UnsentRequest> unsentRequests = new
ArrayList<>();
- if (stagedCommits.isEmpty()) {
- return new NetworkClientDelegate.PollResult(Long.MAX_VALUE, new
ArrayList<>());
+ if (!stagedCommits.isEmpty()) {
Review Comment:
One comment here, I think the intention of the RequestState is that we
should retry these requests by the RequestManager instead of the NetworkClient,
so this is how I think I'm going to write it:
1. `class UnsentOffsetFetchRequest extends Request State`
2. `class UnsentOffsetCommitRequest extends Request State`
3. `class UnsentRequests` holding a list of `UnsentOffsetFetchRequest` and
`UnsentOffsetCommitRequest`
4. So requestManager.poll() will check a. if there are any unsent offset
commits and b. if there's any sendable offsetFetch.
For inflight requests, I guess we will need to create another collection to
track it.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscr...@kafka.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
