[
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. 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.
was:
h2. Status
*Draft for discussion.* This issue records the narrowed scheduling proposal for
KIP-1371. Implementation acceptance remains pending.
KIP:
[KIP-1371|https://cwiki.apache.org/confluence/pages/viewpage.action?pageId=449282795]
Discussion thread: TBD
The linked Confluence page still contains the earlier, broader proposal and has
not yet been synchronized with the scope described here.
h2. Summary
The consumer network thread currently invokes each request manager to perform
work and obtain a delay for the next network poll. An expired timer may return
zero even when a response or another prerequisite is still needed before work
can proceed. Authors must account for this relationship in each manager;
missing it can cause polling without useful progress.
This proposal introduces a programming model that expresses wait conditions
separately from executing work. Each manager declares whether it can run now,
must wait until a retained deadline, or needs an input that changes its state.
The loop applies that declaration before invoking the manager. The goal is to
make wait conditions easier to express, review, compose, and test, reducing the
risk of busy loops.
The change retains deadline-driven network polling, the ordered manager scan,
existing threads and input wakeups. Managers retain ownership of their state
and domain policy. Performance is an acceptance constraint and a possible
benefit, not the primary justification.
h2. Existing evidence
* [KAFKA-20253: High CPU loop on consumer after failed
re-authentication|https://issues.apache.org/jira/browse/KAFKA-20253] — the
merged [PR #22836|https://github.com/apache/kafka/pull/22836] changes
coordinator discovery to return {{PollResult.EMPTY}} while a request is in
flight, instead of letting expired backoff request immediate network polling.
This directly demonstrates the timer/prerequisite mismatch.
* [KAFKA-20426: Using both group.id and assign() causes a busy loop in
AsyncKafkaConsumer|https://issues.apache.org/jira/browse/KAFKA-20426] and
[KAFKA-20970: Another busy loop happens if auto.commit.interval.ms is less than
bootstrap.resolve.timeout.ms|https://issues.apache.org/jira/browse/KAFKA-20970]
have merged fixes in [PR #22018|https://github.com/apache/kafka/pull/22018] and
[PR #23227|https://github.com/apache/kafka/pull/23227]. They illustrate related
prerequisite checks in application-wait calculations. Those application-wait
changes are supporting evidence, not fixes delivered by this proposal.
These cases are already fixed. The proposal addresses the recurring programming
pattern rather than claiming to fix those bugs again. The existing interface
permits correct implementations; the improvement is an explicit scheduling
boundary that consistently applies the conditions reported by managers.
h2. Proposed changes
Introduce an internal {{nextPollCondition(long currentTimeMs)}} query, separate
from {{poll(long currentTimeMs)}}:
* {{ready()}}: a local step can be attempted now, whether or not it produces a
network request.
* {{after(now, delayMs)}}: recheck at a deadline derived from retained owner
state. Repeated queries must not postpone it.
* {{idle()}}: no autonomous step is due; an existing input path must change
state before further work is eligible.
The query does not produce requests, complete futures, or advance domain state.
{{poll()}} retains state changes and request production. Once migration is
complete, {{PollResult}} carries outgoing requests without
{{timeUntilNextPollMs}}.
{{either(first, second)}} combines independent conditions; it is not a fourth
form. For example, a metadata request waiting for a response may retain its
operation-expiry deadline while a separate request retains its retry deadline.
The combined condition is ready when either is ready and keeps the shorter wait.
Before each manager's turn, the loop queries its condition and invokes it only
when eligible. After the ordered pass, the loop re-queries conditions to
observe newly enabled work and bound the network wait, retaining the existing
maximum poll interval. One manager's urgency does not make every manager
eligible.
For coordinator discovery already in flight, {{idle()}} removes unnecessary
manager invocation while the network client continues processing the response
and transport timeout. If the manager owns an independent operation deadline,
that deadline must still be represented.
h2. Scope
This proposal covers request-manager scheduling on {{ConsumerNetworkThread}},
including the consumer, share consumer, and Streams paths using that loop.
It preserves public APIs, configuration, metrics, wire protocols, thread
topology, and the {{ConsumerNetworkThread}} name. The separate duration-based
{{maximumTimeToWait()}} interface and application-wait publication remain
unchanged. Shutdown {{pollOnClose()}} remains unconditional.
Cross-manager state ownership, completion/publication ordering, lifecycle
trackers, general retry or recovery policies, and application-thread wakeup
policy are outside this proposal. The earlier reactor and lifecycle POC is not
the implementation scope of this issue.
h2. Validation and open questions
* Verify deadlines, enabling inputs, request ordering, and shutdown behavior
across managers. Owner-level mixed retry/expiry tests provide focused evidence,
but do not replace loop-level or broker-backed validation.
* Resolve migration compatibility for non-urgent invocation opportunities,
particularly Streams reconciliation. The experimental {{onLoop}} option is not
an adopted fourth form, and complete coverage by the three core forms is not
yet established.
* Establish performance without material regression. Current measurements do
not establish general idle CPU savings or throughput non-regression. Allocation
representation remains an implementation question.
The mechanism applies declared conditions; it does not infer correct domain
policy. Incorrectly returning {{ready()}} can still cause a busy loop, and
omitting an expiry from {{idle()}} can delay completion. State-transition tests
remain necessary.
Summary: KIP-1371: Explicit Wait Conditions for Consumer Request
Managers (was: KIP-1371: Condition-Driven Consumer Request Manager Scheduling)
> 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. 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.
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
