[
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.* 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.
was:
h2. Status
*Draft problem statement for KIP-1371.* The design and implementation are still
being validated. This issue does not commit to a final architecture.
KIP: [KIP-1371: Introduce a Consumer Reactor for State Management and Event
Processing|https://cwiki.apache.org/confluence/spaces/KAFKA/pages/449282795/KIP-1371%2BIntroduce%2Ba%2BConsumer%2BReactor%2Bfor%2BState%2BManagement%2Band%2BEvent%2BProcessing]
Discussion thread: TBD
h2. Problem
A request manager may be in one of three next-poll conditions: it can produce
work now, passage of time may make work possible, or an external input must
arrive first. The current deadline-based result records only when another poll
may occur. For an empty result, it does not identify which condition applies.
The async regular and share consumers already use a background event loop and
request managers. Waiting, wakeup, completion, and publication decisions can
still be made on separate paths using incomplete or differently timed views of
consumer state. This creates four recurring failure shapes:
* *Urgent work without progress.* A timer reaches zero while coordinator
availability, assignment, or in-flight state prevents a request from being
created. This can repeatedly poll the application and background threads
without producing work
([KAFKA-20253|https://issues.apache.org/jira/browse/KAFKA-20253],
[KAFKA-20426|https://issues.apache.org/jira/browse/KAFKA-20426];
[KAFKA-20970|https://issues.apache.org/jira/browse/KAFKA-20970] remains
proposed evidence).
* *Ambiguous empty manager results.* An empty result does not identify whether
time or an input can enable progress. A wakeup selected without corresponding
progress can create application/background ping-pong
([KAFKA-20854|https://issues.apache.org/jira/browse/KAFKA-20854]).
* *Publication and wait ordering races.* State, errors, completions, and
wakeups can become visible in different orders
([KAFKA-18641|https://issues.apache.org/jira/browse/KAFKA-18641];
[KAFKA-20397|https://issues.apache.org/jira/browse/KAFKA-20397] remains
proposed evidence).
* *Distributed lifecycle dependencies.* Coordinator discovery, commit,
leave-group, and shutdown progress can be started or stopped by different
components without one final view of outstanding work
([KAFKA-18569|https://issues.apache.org/jira/browse/KAFKA-18569],
[KAFKA-19357|https://issues.apache.org/jira/browse/KAFKA-19357]).
These issues do not prove that one call mechanism caused every bug. They
demonstrate that consumer-level timing, ordering, and application-visible
effects may depend on state distributed across managers and completion paths.
h2. Desired outcome
KIP-1371 makes the following behavior explicit and testable:
* Each request manager retains its mutable state and domain-specific rules.
* One {{PollResult}} reports produced {{NetworkCommand}} and {{ManagerEvent}}
values together with one typed {{NextPollCondition}}: immediate progress,
finite time-driven retry, or input-driven wait.
* Cross-manager observations are ordered and applied by one state owner against
the version used to create the request.
* {{ConsumerReactor}} publishes one immutable aggregate timing decision before
executing corresponding application-visible completions, notifications, or
wakeups.
* Regular, share, and Streams consumers reuse the execution mechanics while
retaining separate protocol rules.
* Migration remains phased, runnable, bounded, and diagnosable.
h2. Scope and POC
The initial migration is internal. It preserves Kafka protocols, public
{{Consumer}} and {{ShareConsumer}} APIs, callback thread guarantees, runtime
thread names, and the existing application/background thread topology.
POC branch:
[codex/async-consumer-reactor-poc|https://github.com/unknowntpo/kafka/tree/codex/async-consumer-reactor-poc]
Implementation phase map:
[consumer-reactor-poc-phase-map.md|https://github.com/unknowntpo/kafka/blob/cb7e5c4cb7619084d2e91f7efa494b35b1539d09/docs/design/consumer-reactor-poc-phase-map.md]
The Confluence KIP is the authoritative design. POC class placement and
compatibility adapters are implementation evidence, not part of the community
API commitment.
Summary: KIP-1371: Condition-Driven Consumer Request Manager Scheduling
(was: KIP-1371: Formalize Consumer Reactor Cross-Manager Coordination and
Publication)
> KIP-1371: Condition-Driven Consumer Request Manager Scheduling
> --------------------------------------------------------------
>
> 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.* 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.
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
