[
https://issues.apache.org/jira/browse/KAFKA-8635?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Denis Washington updated KAFKA-8635:
------------------------------------
Description:
In our Kafka Streams applications (with EOS enabled), we were seeing mysterious
long delays between records being produced by a stream task and the same
records being consumed by the next task. These delays turned out to always be
around {{retry.backoff.ms}} long; reducing that value reduced the delays by
about the same amount.
After digging further, I pinned down the problem to the following lines in
{{org.apache.kafka.clients.producer.internals.Sender#runOnce}}:
{{} else if (transactionManager.hasInFlightTransactionalRequest() ||
maybeSendTransactionalRequest()) {}}
{{ // as long as there are outstanding transactional requests, we simply
wait for them to return}}}}
{{ client.poll(retryBackoffMs, time.milliseconds());}}}}}}
{{ return;}}
{{}}}
This code seems to assume that, if {{maybeSendTransactionalRequest}} returns
true, a transactional request has been sent out that should be waited for.
However, this is not true if the request requires a coordinator lookup:
{{if (nextRequestHandler.needsCoordinator()) {}}
{{ targetNode =
transactionManager.coordinator(nextRequestHandler.coordinatorType());}}
{{ if (targetNode == null) {}}
{{ transactionManager.lookupCoordinator(nextRequestHandler); }}}}}}
{{ break;}}
{{ }}}
{{ ...}}
{{lookupCoordinator()}} does not actually send anything, but just enqueues a
coordinator lookup request for the {{Sender}}'s next run loop iteration.
{{maybeSendTransactionalRequest}} still returns true, though (the {{break}}
jumps to a {{return true}} at the end of the method), leading the {{Sender}} to
needlessly wait via {{client.poll()}} although there is actually no request
in-flight.
I _think_ the fix is to let {{maybeSendTransactionalRequest}} return false if
it merely enqueues the coordinator lookup instead of actually sending anything.
But I'm not sure, hence the bug report instead of a pull request.
was:
In our Kafka Streams applications (with EOS enabled), we were seeing mysterious
long delays between records being produced by a stream task and the same
records being consumed by the next task. These delays turned out to always be
around {{retry.backoff.ms}} long; reducing that value reduced the delays by
about the same amount.
After digging further, I pinned down the problem to the following lines in
{{org.apache.kafka.clients.producer.internals.Sender#runOnce}}:
{{} else if (transactionManager.hasInFlightTransactionalRequest() ||
maybeSendTransactionalRequest()) {}}
{{ // as long as there are outstanding transactional requests, we simply
wait for them to return}}}}
{{ client.poll(retryBackoffMs, time.milliseconds());}}}}}}
{{ return;}}
{{}}}
This code seems to assume that, if {{maybeSendTransactionalRequest}} returns
true, a transactional request has been sent out that should be waited for.
However, this is not true if the request requires a coordinator lookup:
{{if (nextRequestHandler.needsCoordinator()) {}}
{{ targetNode =
transactionManager.coordinator(nextRequestHandler.coordinatorType());}}
{{ if (targetNode == null) {}}
\{{ transactionManager.lookupCoordinator(nextRequestHandler); }}}}
\{{ break;}}
\{{ }}}
{{ ...}}
{{lookupCoordinator()}} does not actually send anything, but just enqueues a
coordinator lookup request for the {{Sender}}'s next run loop iteration.
{{maybeSendTransactionalRequest}} still returns true, though (the {{break}}
jumps to a {{return true}} at the end of the method), leading the {{Sender}} to
needlessly wait via {{client.poll()}} although there is actually no request
in-flight.
I _think_ the fix is to let {{maybeSendTransactionalRequest}} return false if
it merely enqueues the coordinator lookup instead of actually sending anything.
But I'm not sure, hence the bug report instead of a pull request.
> Unnecessary wait when looking up coordinator before transactional request
> -------------------------------------------------------------------------
>
> Key: KAFKA-8635
> URL: https://issues.apache.org/jira/browse/KAFKA-8635
> Project: Kafka
> Issue Type: Bug
> Components: clients
> Affects Versions: 2.3.0, 2.2.1
> Reporter: Denis Washington
> Priority: Major
>
> In our Kafka Streams applications (with EOS enabled), we were seeing
> mysterious long delays between records being produced by a stream task and
> the same records being consumed by the next task. These delays turned out to
> always be around {{retry.backoff.ms}} long; reducing that value reduced the
> delays by about the same amount.
> After digging further, I pinned down the problem to the following lines in
> {{org.apache.kafka.clients.producer.internals.Sender#runOnce}}:
> {{} else if (transactionManager.hasInFlightTransactionalRequest() ||
> maybeSendTransactionalRequest()) {}}
> {{ // as long as there are outstanding transactional requests, we simply
> wait for them to return}}}}
> {{ client.poll(retryBackoffMs, time.milliseconds());}}}}}}
> {{ return;}}
> {{}}}
> This code seems to assume that, if {{maybeSendTransactionalRequest}} returns
> true, a transactional request has been sent out that should be waited for.
> However, this is not true if the request requires a coordinator lookup:
> {{if (nextRequestHandler.needsCoordinator()) {}}
> {{ targetNode =
> transactionManager.coordinator(nextRequestHandler.coordinatorType());}}
> {{ if (targetNode == null) {}}
> {{ transactionManager.lookupCoordinator(nextRequestHandler); }}}}}}
> {{ break;}}
> {{ }}}
> {{ ...}}
> {{lookupCoordinator()}} does not actually send anything, but just enqueues a
> coordinator lookup request for the {{Sender}}'s next run loop iteration.
> {{maybeSendTransactionalRequest}} still returns true, though (the {{break}}
> jumps to a {{return true}} at the end of the method), leading the {{Sender}}
> to needlessly wait via {{client.poll()}} although there is actually no
> request in-flight.
> I _think_ the fix is to let {{maybeSendTransactionalRequest}} return false if
> it merely enqueues the coordinator lookup instead of actually sending
> anything. But I'm not sure, hence the bug report instead of a pull request.
--
This message was sent by Atlassian JIRA
(v7.6.3#76005)
