KAFKA-17439: Make polling for new records an explicit action/event in the new consumer

[kirktrue]

Updated the FetchRequestManager to only create and enqueue fetch requests when signaled to do so by a FetchEvent.

The application thread and the background thread each contains logic that is performed if there is buffered data from a previous fetch. There's a race condition because the presence of buffered data could change between the two threads’ respective checks. Right now the window for the race condition to occur is wide open; this change aims to make the window ajar.

In the ClassicKafkaConsumer, the application thread will explicitly issue fetch requests (via the Fetcher class) at specific points in the Consumer.poll() cycle. Prior to this change, the AsyncKafkaConsumer would issue fetch requests independently from the user calling Consumer.poll(); the fetches would happen nearly continuously as soon as any assigned partition was fetchable. With this change, the AsyncKafkaConsumer introduces a FetchEvent that signals to the background thread that a fetch request should be issued. The specific points where this is done in the Consumer.poll() cycle of the AsyncKafkaConsumer now match the ClassicKafkaConsumer. In short: this makes AsyncKafkaConsumer act nearly identical to the ClassicKafkaConsumer in this regard.

As mentioned above, this change does not completely solve the problem related to fetch session eviction. Exactly how the window where the race condition can be shut completely is outside the scope of this change.

See KAFKA-17182.

Committer Checklist (excluded from commit message)

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

[AndrewJSchofield]

Thanks for the PR. I'm quite surprised to see how little code needed to change to make the fetching explicit. A couple of points.

First, I think it better not to overload the PollEvent because that's already used in the share consumer.

Second, it seems to me that there is still the potential for over-fetching, and this will still cause churn of the fetch session cache.

In the case where the consumer is only fetching a single partition, I think it works pretty well. The set of fetchable partitions will be empty if there's buffered data, and contain the only partition in the fetch session if there is not. So, you'll only send a Fetch request when there's a need for more data and the fetch session will not churn.

In the case where the consumer is fetching more than one partition on a particular node, if a subset of the partitions is fetchable, then the fetch session will be modified by sending a Fetch request and that seems to have the potential for a lot of churn.

Of course, all of this code is in common between the legacy consumer and the async consumer. The async consumer is still very keen on fetching so I don't properly grasp why this PR would make the fetch session behaviour better.

[kirktrue]

Hi @AndrewJSchofield!

Thanks for the review 👍

First, I think it better not to overload the PollEvent because that's already used in the share consumer.

Agreed. I've introduced a FetchEvent so that the two separate mechanisms won't step on each others' toes.

Second, it seems to me that there is still the potential for over-fetching, and this will still cause churn of the fetch session cache.

Agreed. It aims to lessen the churn. Preventing the churn completely is a future task.

In the case where the consumer is only fetching a single partition, I think it works pretty well. The set of fetchable partitions will be empty if there's buffered data, and contain the only partition in the fetch session if there is not. So, you'll only send a Fetch request when there's a need for more data and the fetch session will not churn.

Correct.

In the case where the consumer is fetching more than one partition on a particular node, if a subset of the partitions is fetchable, then the fetch session will be modified by sending a Fetch request and that seems to have the potential for a lot of churn.

Correct again!

Any partition with buffered data at the point where the fetch request is being generated will be marked as "removed" from the broker's fetch session cache. That's the crux of the problem 😞

Something that I tend to lose sight of is the fact that it's not a foregone conclusion that a fetch session will be evicted when it has partitions removed. Of course, it will increase its eligibility for eviction if the broker hosting the fetch session is resource-constrained and invokes the eviction process.

Of course, all of this code is in common between the legacy consumer and the async consumer.

I'm not sure I follow. This code is all specific to the AsyncKafkaConsumer. While the ClassicKafkaConsumer has a similar race condition, it is 2-4 orders of magnitude less likely to happen.

The async consumer is still very keen on fetching so I don't properly grasp why this PR would make the fetch session behaviour better.

Yep—the design of the AsyncKafkaConsumer fetching continuously in the background makes it very keen to cause this problem. With this change, the application thread now signals when to fetch, which results in the background thread creating and issuing the fetch requests much less often.

Thanks!

[kirktrue]

@AndrewJSchofield, et al.—it can be helpful to compare the flow of ClassicKafkaConsumer.poll() and AsyncKafkaConsumer.poll(), specifically how it invokes fetch. Note that the sendFetches() method name, as well as when it is invoked, comes from the ClassicKafkaConsumer.poll(). So this is really making the new consumer act much more like the old one.

[AndrewJSchofield]

Thanks for the updates and for the explanation of the mechanism.

I think it would be appropriate to test in FetchRequestManagerTest the mechanism of the pending fetch request future in the various permutations.

[lianetm]

Hey @kirktrue , thanks for the updates, some comments...

[lianetm]

Thanks for the updates!

[kirktrue]

@lianetm—tests are passing and all comments have been addressed. Can you make another review pass? Thanks!

[lianetm]

Thanks for the updates @kirktrue! Took another pass and left some comments for consideration.

[lianetm]

do we need to complete this future also on pollOnClose? there may be a pendingFetchRequestFuture there that won't be completed (not that I'm seeing how leaving that future uncompleted on close will cause a problem but seems safer to complete it, consistently with how we do it here after pollInternal)

[kirktrue]

I've moved the Future-handling code to pollInternal() for consistency. LMK what you think.

[lianetm]

The actual poll now happens in here (addAndGet that will complete when the background has had one run, called fetchMgr.poll), so should the log line on ln 1538 "Polling for fetches with timeout..." be right before this?

[kirktrue]

We're not polling for the incoming responses in sendFetches(), just enqueuing the outgoing requests. This mimics the ClassicKafkaConsumer in that the requests are enqueued in its sendFetches() but then toward the bottom of pollForFetches() client.poll() is invoked to wait for the results of the fetch requests.

[lianetm]

well the sendFetches blocks until the CreateFetchRequestsEvent completes, and that only happens on fetchMgr.poll

kafka/clients/src/main/java/org/apache/kafka/clients/consumer/internals/FetchRequestManager.java

Line 115 in aae7e97

pendingFetchRequestFuture.complete(null);

So when the sendFetches completes we did poll the manager right? (and depending on time, maybe we did poll the client.poll too, which happens in the background right after polling all managers). That's why the log for "Polling for fetches" made sense to me before the sendFetches, but am I missing another poll happening after the log line maybe? (where it is now)

[kirktrue]

The two ConsumerDelegate implementations work differently:

• AsyncKafkaConsumer: FetchRequestManager.poll() will complete the event's Future on the background thread before it exits, i.e. before the thread starts the network I/O. Completing the Future starts the application thread racing toward logging that message and the background thread racing toward starting network I/O. I'll admit—I haven't dug through the code to surmise the relative costs of each thread's work before either cross their respective finish lines to win.
• ClassicKafkaConsumer: Fetcher.sendFetchesInternal() calls ConsumerNetworkClient.send() to enqueue the request, but then it calls NetworkClient.wakeup(). Since the same ConsumerNetworkClient instance used by the consumer is also used by AbstractCoordinator.HeartbeatThread, it's technically possible that the heartbeat thread's run() method could start network I/O when it calls NetworkClient.pollNoWakeup(). Granted, that's a race that the application thread is much more likely to win given that the heartbeat thread runs much less frequently.

Here are some points to consider:

• The definition of the term "poll" as used in the log is open to interpretation. The term "poll" is everywhere, making its meaning ambiguous at any given point of use 😢
• I agree there is a race condition (for both consumers, but more likely for the new consumer) that could result the the log message being emitted after the network I/O has commenced
• For this to pose a problem to users, there needs to be other log entries that we're racing with, right?. We're trying to avoid the condition where the user is confused/mislead because the entries in the log are emitted in non-deterministic ordering.
• The log line in question is only output at level TRACE, which I assume is very rare for users to enable.

Given the above, I'm of the opinion that it's an exercise in hair splitting to alter the logging. However, I could also just change it which would have been way less effort than researching, thinking, and composing this response 🤣

If we leave the log line as it is, what would the effect be for the user?

[lianetm]

I surely didn't intend for you to put up that long response he he, sorry. It's not about the log line per-se, it's about the alignment on where the poll happens. The classic consumer logs "Polling for records", then calls client.poll. vs us here we do sendFetches (which triggers the client.poll async in the background thread because it blocks until we poll the fetch manager), then log "Polling for fetches...".

That's the diff I saw and just wanted to understand/align on where the poll happens: once we trigger sendFetches (blocking), the client.poll will happen in the background anytime, not controlled by the app thread. Agreed? If so I'm ok with leaving the log unchanged, understanding it could come out after the client.poll happened.

[kirktrue]

That's the diff I saw and just wanted to understand/align on where the poll happens: once we trigger sendFetches (blocking), the client.poll will happen in the background anytime, not controlled by the app thread. Agreed?

Agreed—the background thread is going to move from calling each of the RequestManager’s poll() method to NetworkClient.poll() method without the intervention of the application thread.

If so I'm ok with leaving the log unchanged, understanding it could come out after the client.poll happened.

Thanks!

[lianetm]

at this point we may already have records in hand to return (consumed position updated), so we should be very careful to not throw any error here. But this sendFetches could throw interrupted because of the addAndGet right?

Shouldn't we just do a best effort to pipeline the next requests using add instead of addAndGet? It would achieve what we want, removing the risk of errors, and it would actually align better with what the classic does on this sendFetches + transmitSends:

kafka/clients/src/main/java/org/apache/kafka/clients/consumer/internals/ConsumerNetworkClient.java

Lines 327 to 329 in 140d35c

	* Poll for network IO in best-effort only trying to transmit the ready-to-send request
	* Do not check any pending requests or metadata errors so that no exception should ever
	* be thrown, also no wakeups be triggered and no interrupted exception either.

[kirktrue]

Done.

[lianetm]

Thanks for the updates @kirktrue! Just one nit left, almost there.

[lianetm]

This is not true now for prefetching that uses .add instead of .addAndGet, should we remove this line?

[kirktrue]

Good catch. Reworded to state that it will not wait for confirmation.

[lianetm]

Thanks for all the updates @kirktrue! LGTM.