KAFKA-14468: Implement CommitRequestManager to manage the commit and autocommit requests by philipnee · Pull Request #13021 · apache/kafka

philipnee · 2022-12-19T18:08:51Z

This pull request introduces a CommitRequestManager to efficiently manage commit requests from clients and the autocommit state. The manager utilizes a "staged" commit queue to store commit requests made by clients. A background thread regularly polls the CommitRequestManager, which then checks the queue for any outstanding commit requests. When permitted, the CommitRequestManager generates a PollResult which contains a list of UnsentRequests that are subsequently processed by the NetworkClientDelegate.

In addition, a RequestManagerRegistry has been implemented to hold all request managers, including the new CommitRequestManager and the CoordinatorRequestManager. The registry is regularly polled by a background thread in each event loop, ensuring that all request managers are kept up to date and able to handle incoming requests

Committer Checklist (excluded from commit message)

Verify design and implementation
Verify test coverage and CI build status
Verify documentation (including upgrade notes)

philipnee · 2022-12-20T04:08:29Z

@kirktrue - Could you take a quick pass?

philipnee · 2022-12-20T04:12:51Z

need to refactor this sad toString method.

philipnee · 2022-12-20T04:27:03Z

we should test the order commit when calling poll. The commit sequence order should be guarantee.

philipnee · 2022-12-20T19:13:18Z

Not sure if it is worth writing a builder.

vvcephei · 2022-12-29T16:37:37Z

Thanks @philipnee !

I just made a quick pass. I'm sorry; I haven't been keeping up with the progress of your project, so I'm a little hazy on the details.

It looks like the new CommitRequestManager is a component that lives in the background part of the system, basically to help the background thread keep track of commit requests that have been/need to be sent to the network?

I think that makes sense, but in that case, it is mildly surprising to see autocommit living where it lives. I initially would have expected it to originate from the foreground, but on second though, maybe it's because the background thread runs on a reliable timer? The caller is required to call poll on a regular basis, but maybe it's not frequent enough for your purposes?

It would be good to understand the general principle for the division of responsibilities between the foreground and background. Do you have this documented somewhere that we can use for a north star?

philipnee · 2023-01-05T19:13:44Z

We should finalize the plan for subscriptionState, whether we will reuse the same class or implement a new one (and deprecate the old one)

I think the main problem with using the current SubscriptionState is that it couples together the commit position and the fetch position. With the move to the background thread, I don't think we can depend on these advancing in lock-step anymore. The fetch position can advance after we have received a fetch position and the commit position will be advanced when the application calls poll(). With that said, it is probably simpler to build this into the current SubscriptionState instead of creating something new.

+1. I checked the current SusbscriptionState class and feel it's still resuable.

hachikuji · 2023-01-18T23:43:42Z

I think the main problem with using the current SubscriptionState is that it couples together the commit position and the fetch position. With the move to the background thread, I don't think we can depend on these advancing in lock-step anymore. The fetch position can advance after we have received a fetch position and the commit position will be advanced when the application calls poll(). With that said, it is probably simpler to build this into the current SubscriptionState instead of creating something new.

hachikuji · 2023-01-18T23:54:13Z

nit: this and config are unused

hachikuji · 2023-01-19T00:07:28Z

I think we can simplify this API by removing the callback argument. Instead, if we let this API return CompletableFuture as suggested in another comment, then we can chain the callback logic to the future when invoked from KafkaConsumer.

hachikuji · 2023-01-19T00:17:20Z

Can we use a normal CompletableFuture instead of RequestFuture. The latter is bound up with ConsumerNetworkClient which we are not using in this implementation.

Also, I'm surprised to see this return something as low level as ClientResponse. I think we can leave the response handling confined to this class and just expose CompletableFuture<Void> or something like that.

hachikuji · 2023-01-19T00:18:54Z

Stuff like this is better to just chain on top of a future.

hachikuji · 2023-01-19T00:20:56Z

Why would this be optional? We cannot commit offsets unless we have a coordinator.

hachikuji · 2023-01-19T00:21:30Z

The optional is kind of annoying. I wonder if we could treat auto-commit disabled as having an auto-commit interval of Long.MaxVAlue.

It makes sense to use a MaxValue as well, my counter argument is, i think explicitly disabling autoCommitState makes the logic more straightforward.

commit manager clean up clean up clean up client poll to autocommit wip Addressed some PR comments Init commit for the request manager

guozhangwang

One major question is that stagedCommits queue seems not being cleared, is that intentional? All others are minor.

guozhangwang · 2023-02-03T22:40:21Z

+import java.util.Objects;
+import java.util.Optional;
+
+public class GroupStateManager {


Maybe just call it GroupState since it does not do any management, but more of a grouping of multiple primitive states?

thanks for suggestions, changed.

guozhangwang · 2023-02-03T22:41:02Z

+    private final Logger log;
+    private final Optional<AutoCommitState> autoCommitState;
+    private final CoordinatorRequestManager coordinatorRequestManager;
+    private final GroupStateManager groupState;


I'd suggest we either change the class name to just GroupState or name this field groupStateManager? Personally I prefer the former.

Agreed. makes sense. thanks.

guozhangwang · 2023-02-03T22:43:38Z

+1. I checked the current SusbscriptionState class and feel it's still resuable.

guozhangwang · 2023-02-03T22:49:07Z

+
+        Map<TopicPartition, OffsetAndMetadata> allConsumedOffsets = subscriptionState.allConsumed();
+        log.debug("Auto-committing offsets {}", allConsumedOffsets);
+        sendAutoCommit(allConsumedOffsets);


I'm wondering if we should reset immediately or should we have a flag of autoCommitInFlight and only reset to re-enable canSendAutocommit, because I vaguely remember in the past, we have seen issues where auto commit keeps being triggered while there are still auto commits inflight due to network partition, causing OOM and other issues. And that's why we ended up adding this flag in the old code.

If we want to go very fancy, we can potentially just update the unsent auto commit request inside the stagedCommits when we want to send another..

Hmm, I see your point. Let's track the inflight status in the autocommit state and reset them upon completing the future (in the callback).

guozhangwang · 2023-02-03T22:49:57Z

+                log.debug("Completed asynchronous auto-commit of offsets {}", allConsumedOffsets);
+            }
+
+            if (throwable instanceof RetriableCommitFailedException) {


merge this as else if along with line 124?

I moved it to the exceptionally block, it think it would be clearer this way.

guozhangwang · 2023-02-03T22:53:55Z

+    public NetworkClientDelegate.PollResult poll(final long currentTimeMs) {
+        maybeAutoCommit(currentTimeMs);
+
+        if (stagedCommits.isEmpty()) {


The stagedCommits seems never cleared?

thanks, addressed.

guozhangwang · 2023-02-03T22:55:36Z

-                            final CoordinatorRequestManager coordinatorManager) {
+                            final GroupStateManager groupState,
+                            final CoordinatorRequestManager coordinatorManager,
+                            CommitRequestManager commitRequestManager) {


Why this is not final?

guozhangwang · 2023-02-03T23:01:40Z

+    private boolean process(final PollApplicationEvent event) {
+        Optional<RequestManager> commitRequestManger = registry.get(RequestManager.Type.COMMIT);
+        if (!commitRequestManger.isPresent()) {
+            return false;


Why return false here? If the user did not set group.id and never use auto or manual commits, then here we should just skip right?

Address comments from the PR review

guozhangwang · 2023-02-07T19:31:42Z

Made another look, LGTM. Leaving for @hachikuji for a final look.

philipnee · 2023-02-07T19:37:14Z

much thanks @guozhangwang

philipnee · 2023-02-10T19:51:44Z

There's still a couple things to be hashed out in the future, in particular

how do we plan to structure the subscription state and
what are the API calls for the subscription state in the background thread

For this specific PR, I think let's use the existing API, and change it later. There's really not a lot of interaction to the subscription state for this model for now, so I think it's fine. In the new model, we might need to update the commit positions in the callback, but we can add that after this.

@hachikuji @kirktrue

philipnee force-pushed the consumer-refactor-commit-request-manager branch from 505c362 to 8b7eb65 Compare December 20, 2022 04:06

philipnee marked this pull request as ready for review December 20, 2022 04:08

philipnee commented Dec 20, 2022

View reviewed changes

Comment thread clients/src/main/java/org/apache/kafka/clients/consumer/internals/CommitRequestManager.java Outdated

philipnee commented Dec 20, 2022

View reviewed changes

philipnee commented Jan 5, 2023

View reviewed changes

philipnee commented Jan 9, 2023

View reviewed changes

Comment thread clients/src/main/java/org/apache/kafka/clients/consumer/internals/CommitRequestManager.java Outdated

hachikuji reviewed Jan 19, 2023

View reviewed changes

Add commit manager to handle commits

40abf2e

commit manager clean up clean up clean up client poll to autocommit wip Addressed some PR comments Init commit for the request manager

philipnee force-pushed the consumer-refactor-commit-request-manager branch from 0538244 to 40abf2e Compare January 24, 2023 21:57

Philip Nee added 2 commits January 24, 2023 16:05

dead store

a3d87af

Addressed previous comments and add more test cases to improve coverage

49aa05d

guozhangwang reviewed Feb 3, 2023

View reviewed changes

Philip Nee added 3 commits February 6, 2023 13:10

Addressed some comments

9813dcf

Address comments from the PR review

ensure staged commit is purged (testing)

d98695f

make commit manager final

da218d0

guozhangwang merged commit 62431dc into apache:trunk Feb 24, 2023

philipnee added KIP-848 The Next Generation of the Consumer Rebalance Protocol ctr Consumer Threading Refactor (KIP-848) labels May 23, 2023

Uh oh!

Conversation

philipnee commented Dec 19, 2022 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Committer Checklist (excluded from commit message)

Uh oh!

philipnee commented Dec 20, 2022

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

vvcephei commented Dec 29, 2022

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

guozhangwang left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

guozhangwang Feb 3, 2023 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

guozhangwang commented Feb 7, 2023

Uh oh!

philipnee commented Feb 7, 2023

Uh oh!

philipnee commented Feb 10, 2023

Uh oh!

Reviewers

Assignees

Labels

philipnee commented Dec 19, 2022 •

edited

Loading

guozhangwang Feb 3, 2023 •

edited

Loading