[
https://issues.apache.org/jira/browse/KAFKA-20995?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Eric Chang updated KAFKA-20995:
-------------------------------
Description:
h2. Status
*Draft for discussion.* Implementation acceptance remains pending.
KIP:
[KIP-1371|https://cwiki.apache.org/confluence/pages/viewpage.action?pageId=449282795]
Discussion thread: TBD
h2. Summary
Consumer request managers currently return outgoing requests together with one
numeric wait. Several busy-loop fixes share the same failure shape: a timer
reaches zero while an in-flight request, unavailable coordinator, or another
prerequisite prevents the related work from progressing. Existing fixes replace
that zero with another configured duration, which can make correctness depend
on the replacement value.
KIP-1371 introduces the internal {{ManagerPollCondition}} type. Each
request-manager {{poll()}} returns its outgoing requests and the condition for
the next manager pass:
* {{ready()}}: another local step can run immediately.
* {{after(now, delayMs)}}: time can make another pass useful.
* {{idle()}}: this work source has no autonomous deadline.
* {{either(first, second)}}: retain the earlier of independent obligations.
The request owner evaluates prerequisites before configuration-derived timing.
The same owner-local decision can govern request admission and network
scheduling. The network thread polls every manager once, combines the returned
conditions, and applies the existing network-poll bounds.
h2. Evidence
* [KAFKA-20253 / PR #22836|https://github.com/apache/kafka/pull/22836]
demonstrates an in-flight coordinator request whose elapsed retry timer
repeatedly requested immediate network polling.
* [KAFKA-21031 / PR #23357|https://github.com/apache/kafka/pull/23357]
demonstrates the same network-scheduling shape for an in-flight heartbeat with
an expired heartbeat timer.
* [KAFKA-20970 / PR #23227|https://github.com/apache/kafka/pull/23227] and
[KAFKA-21010 / PR #23348|https://github.com/apache/kafka/pull/23348] show the
related timeout-substitution hazard in the separate application-wait
calculation.
These issues have path-specific fixes or proposed fixes. The KIP addresses the
recurring programming model so the prerequisite and independent deadlines can
be reviewed together.
h2. Proposed changes
For request sources using {{RequestState}}, one {{sendCondition(now)}} decision
supplies both request admission and network scheduling. An in-flight request
returns {{idle()}} before consulting retry timing. Remaining backoff returns
{{after(...)}}, and eligible work returns {{ready()}}. Independent deadlines,
such as {{max.poll.interval.ms}} expiry, remain active through {{either(...)}}.
{{PollResult}} carries outgoing requests and the condition left after that
poll. The consumer network thread combines every manager condition and uses the
earliest remaining wait for network polling. It keeps the current manager
order, one poll per manager per pass, existing network-client bounds, input
wakeups, and unconditional {{pollOnClose(...)}} behavior.
The internal per-manager {{RequestManager.maximumTimeToWait(long)}} method
becomes {{applicationPollCondition(long)}} and uses the same condition
vocabulary. The application-facing
{{ApplicationEventHandler.maximumTimeToWait()}} path and application-wait
policy remain unchanged.
h2. Scope
The proposal covers the async consumer request-manager loop shared by regular,
share, and Streams consumers. Public consumer APIs, configuration, metrics,
wire protocols, thread topology, manager order, and existing wakeup paths
remain unchanged. {{ClassicKafkaConsumer}} keeps its current execution model.
Correctness still depends on each manager selecting the condition that matches
its state. Focused tests cover in-flight, backoff, missing-coordinator,
independent-expiry, completion, close, and network-loop aggregation behavior.
Performance is an acceptance constraint; the proposal does not claim a general
idle-CPU improvement.
was:
h2. Status
*Draft for discussion.* Implementation acceptance remains pending.
KIP:
[KIP-1371|https://cwiki.apache.org/confluence/pages/viewpage.action?pageId=449282795]
Discussion thread: TBD
h2. Summary
Consumer request managers currently return outgoing requests together with one
numeric wait. Several busy-loop fixes share the same failure shape: a timer
reaches zero while an in-flight request, unavailable coordinator, or another
prerequisite prevents the related work from progressing. The zero duration
immediately starts another iteration. Existing fixes repair each path with
local guards or substitute another configured timeout, leaving the same
relationship to be reconstructed for every manager state.
KIP-1371 introduces an internal {{ManagerPollCondition}} vocabulary. Each
request-manager {{poll()}} returns its work and the condition for the next
manager pass:
* {{ready()}}: another local step can run immediately.
* {{after(now, delayMs)}}: re-evaluate at the retained absolute deadline.
* {{idle()}}: this work source has no autonomous deadline.
* {{either(first, second)}}: retain the earlier of independent obligations.
The request owner evaluates eligibility before configuration-derived timing.
Request admission and scheduling use the same owner-local decision, while
independent operation and transport deadlines remain visible. The network
thread polls every manager once, aggregates the returned conditions, and bounds
network I/O by the earliest manager or transport deadline.
h2. Evidence
* [KAFKA-20253 / PR #22836|https://github.com/apache/kafka/pull/22836]
demonstrates an in-flight coordinator request whose elapsed retry timer
repeatedly requested immediate network polling.
* [KAFKA-21031 / PR #23357|https://github.com/apache/kafka/pull/23357]
demonstrates the same network-scheduling shape for an in-flight heartbeat with
an expired heartbeat timer.
* [KAFKA-20970 / PR #23227|https://github.com/apache/kafka/pull/23227] and
[KAFKA-21010 / PR #23348|https://github.com/apache/kafka/pull/23348] show the
related timeout-substitution hazard in the separate application-wait projection.
These issues have path-specific fixes or proposed fixes. The KIP addresses the
recurring programming model so the prerequisite and retained deadlines can be
reviewed together.
h2. Scope
The proposal covers the async consumer request-manager loop used by regular,
share, and Streams consumers. It changes internal scheduling types and the
internal per-manager application-wait projection. Public consumer APIs,
configuration, metrics, wire protocols, thread topology, manager order, and
input wakeups remain unchanged.
Transport contributes its earliest unsent or sent request timeout whenever a
manager returns {{idle()}}, so removing a manager deadline cannot suppress
request expiration. Performance is an acceptance constraint; current
measurements do not establish general idle CPU savings or throughput
non-regression.
h2. Validation
The prototype migrates coordinator, commit, topic metadata, share
acknowledgement, Streams topology, and regular/share heartbeat scheduling to
shared owner-local decisions. Focused tests cover retry, in-flight, timeout,
throttle, leave, close, and independent-expiry behavior. Before merge,
validation must complete the transport-deadline integration,
zero/positive/maximum timer sweep, loop-level liveness coverage, and stable
performance comparison.
> KIP-1371: Explicit Wait Conditions for Consumer Request Managers
> ----------------------------------------------------------------
>
> Key: KAFKA-20995
> URL: https://issues.apache.org/jira/browse/KAFKA-20995
> Project: Kafka
> Issue Type: Improvement
> Components: consumer
> Reporter: Eric Chang
> Assignee: Eric Chang
> Priority: Major
>
> h2. Status
> *Draft for discussion.* Implementation acceptance remains pending.
> KIP:
> [KIP-1371|https://cwiki.apache.org/confluence/pages/viewpage.action?pageId=449282795]
> Discussion thread: TBD
> h2. Summary
> Consumer request managers currently return outgoing requests together with
> one numeric wait. Several busy-loop fixes share the same failure shape: a
> timer reaches zero while an in-flight request, unavailable coordinator, or
> another prerequisite prevents the related work from progressing. Existing
> fixes replace that zero with another configured duration, which can make
> correctness depend on the replacement value.
> KIP-1371 introduces the internal {{ManagerPollCondition}} type. Each
> request-manager {{poll()}} returns its outgoing requests and the condition
> for the next manager pass:
> * {{ready()}}: another local step can run immediately.
> * {{after(now, delayMs)}}: time can make another pass useful.
> * {{idle()}}: this work source has no autonomous deadline.
> * {{either(first, second)}}: retain the earlier of independent obligations.
> The request owner evaluates prerequisites before configuration-derived
> timing. The same owner-local decision can govern request admission and
> network scheduling. The network thread polls every manager once, combines the
> returned conditions, and applies the existing network-poll bounds.
> h2. Evidence
> * [KAFKA-20253 / PR #22836|https://github.com/apache/kafka/pull/22836]
> demonstrates an in-flight coordinator request whose elapsed retry timer
> repeatedly requested immediate network polling.
> * [KAFKA-21031 / PR #23357|https://github.com/apache/kafka/pull/23357]
> demonstrates the same network-scheduling shape for an in-flight heartbeat
> with an expired heartbeat timer.
> * [KAFKA-20970 / PR #23227|https://github.com/apache/kafka/pull/23227] and
> [KAFKA-21010 / PR #23348|https://github.com/apache/kafka/pull/23348] show the
> related timeout-substitution hazard in the separate application-wait
> calculation.
> These issues have path-specific fixes or proposed fixes. The KIP addresses
> the recurring programming model so the prerequisite and independent deadlines
> can be reviewed together.
> h2. Proposed changes
> For request sources using {{RequestState}}, one {{sendCondition(now)}}
> decision supplies both request admission and network scheduling. An in-flight
> request returns {{idle()}} before consulting retry timing. Remaining backoff
> returns {{after(...)}}, and eligible work returns {{ready()}}. Independent
> deadlines, such as {{max.poll.interval.ms}} expiry, remain active through
> {{either(...)}}.
> {{PollResult}} carries outgoing requests and the condition left after that
> poll. The consumer network thread combines every manager condition and uses
> the earliest remaining wait for network polling. It keeps the current manager
> order, one poll per manager per pass, existing network-client bounds, input
> wakeups, and unconditional {{pollOnClose(...)}} behavior.
> The internal per-manager {{RequestManager.maximumTimeToWait(long)}} method
> becomes {{applicationPollCondition(long)}} and uses the same condition
> vocabulary. The application-facing
> {{ApplicationEventHandler.maximumTimeToWait()}} path and application-wait
> policy remain unchanged.
> h2. Scope
> The proposal covers the async consumer request-manager loop shared by
> regular, share, and Streams consumers. Public consumer APIs, configuration,
> metrics, wire protocols, thread topology, manager order, and existing wakeup
> paths remain unchanged. {{ClassicKafkaConsumer}} keeps its current execution
> model.
> Correctness still depends on each manager selecting the condition that
> matches its state. Focused tests cover in-flight, backoff,
> missing-coordinator, independent-expiry, completion, close, and network-loop
> aggregation behavior. Performance is an acceptance constraint; the proposal
> does not claim a general idle-CPU improvement.
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
