[
https://issues.apache.org/jira/browse/KAFKA-20870?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=18101488#comment-18101488
]
Matthias J. Sax commented on KAFKA-20870:
-----------------------------------------
But Claude found something else... There is a code path to handle stale epochs,
which seems to be broken if all three lists are `null` (which is allowed and
won't trigger the guard you found).
{code:java}
GroupMetadataManager:1872-1882, the stale-epoch branch whose own comment says
"we accept it because the response with the bumped epoch may have been lost":
|| !areOwnedTasksContainedInAssignedTasksWithEpochs(ownedActiveTasks, ...)
|| !areOwnedTasksContainedInAssignedTasks(ownedStandbyTasks, ...)
and both helpers open with if (ownedTasks == null) return false; (:1539,
:1565). So null → false → negated → fenced. A member retrying with its previous
epoch and "nothing changed" gets FencedMemberEpochException instead of the
acceptance that branch exists to provide.
The AK client is shielded, but by accident rather than design:
heartbeatState.reset() fires on every error response and every transport
failure (:470, :729, :848), clearing lastSentFields.assignment, so a retry
after a lost response resends all three lists. A third-party client that keeps
its epoch and omits unchanged lists — exactly what the per-field wording
invites — gets fenced.{code}
> Fix owned-task reports in StreamsGroupHeartbeat
> -----------------------------------------------
>
> Key: KAFKA-20870
> URL: https://issues.apache.org/jira/browse/KAFKA-20870
> Project: Kafka
> Issue Type: Bug
> Components: group-coordinator, streams
> Reporter: Matthias J. Sax
> Priority: Major
> Labels: needs-kip
>
> This ticket is partially a bug report, and mainly an improvement idea.
> _Summary:_ Broker discards the entire StreamsGroupHeartbeat owned-task report
> unless all three task lists are non-null
> h3. 1) What the protocol documents
> `StreamsGroupHeartbeatRequest` has three _independently_ nullable owned-task
> lists, each documented per field:
> - `ActiveTasks` — *"Currently owned active tasks for this client. Null if
> unchanged since last heartbeat."*
> - `StandbyTasks` — same wording
> - `WarmupTasks` — same wording
> Nullable arrays distinguish null from empty on the wire (compact encoding:
> `0x00` = null, `0x01` = empty), and both values are meaningful here. An empty
> list is how a member reports that it has released a role's tasks; null is
> documented to mean "unchanged". Nothing in the schema says the three fields
> must be sent together.
> h3. 2) What the broker does
> -In the streams reconciliation path, `GroupMetadataManager` converts the
> request into a `TasksTuple` only when all three lists are non-null:-
> {code:java}
> TasksTuple ownedTasks = null;
> if (ownedActiveTasks != null && ownedStandbyTasks != null && ownedWarmupTasks
> != null) {
> ownedTasks = TasksTuple.fromHeartbeatRequest(ownedActiveTasks,
> ownedStandbyTasks, ownedWarmupTasks);
> } {code}
> -`ownedTasks` is put into an `Optional`.-
> -Otherwise the whole report is dropped (an empty `Optional` is create, even
> if some field are not-null), and `CurrentAssignmentBuilder.hasNotReleased`
> reads an absent report conservatively:-
> {code:java}
> return ownedTasks.isEmpty() ||
> ownedTasks.get().containsAny(tasksPendingRevocation); ```{code}
> -(`ownedTasks` is the `Optional<TasksTuple>`; `isEmpty()` here means "no
> report", not "owns nothing".)-
> -So a heartbeat reporting two of the three lists is treated identically to
> one reporting none — the lists that *were* sent are discarded.-
> There is actually an upfront guard, rejecting such a HB inside
> `GroupCoordinatorService#throwIfStreamsGroupHeartbeatRequestIsInvalid`.
> h3. 3) Consequence
> A member that reports a subset of the three lists never has _any_ revocation
> acknowledged, including for the roles it did report. The acknowledgement
> never lands, so the member stays in its revoking state and the group never
> leaves rebalancing. No error is returned and nothing is logged, so the group
> simply stops making progress.
> The failure mode is {_}liveness only, never safety{_}: the conservative
> reading means the broker never concludes a task was released when it wasn't,
> so it cannot grant a task to a new owner while the previous owner still runs
> it.
> A second site makes the same assumption:
> `throwIfStreamsGroupMemberEpochIsInvalid`. In the stale-epoch branch
> (`receivedMemberEpoch < member.memberEpoch()`),
> `areOwnedTasksContainedInAssignedTasks` returns `false` for a null list, so a
> member retrying with its previous epoch and any null owned-task list is
> fenced with `FencedMemberEpochException`.
> h3. 4) Root cause
> `TasksTuple`'s record constructor requires all three maps to be non-null, so
> "this role was not reported" has no representation in the type the broker
> converts the request into. Three independent absences on the wire are
> collapsed into a single `Optional<TasksTuple>`, whose only lossless inputs
> are all-present and all-absent; everything in between is rounded to absent.
> `TasksTuple` is the right type for an *assignment* (where all three roles are
> always fully known) and the wrong type for a wire-side {*}report{*}.
> h3. 5) Not reachable with the Apache Java client
> The client is itself all-or-nothing:
> - The JOINING heartbeat sets all three lists to empty lists (an explicit "I
> own nothing").
> - Otherwise all three are set together, and only when the reconciled
> assignment changed.
> - `HeartbeatState.reset()` clears the recorded assignment and is called
> unconditionally on every error response and on every transport failure, so a
> full report is resent after a coordinator change.
> This is therefore latent, and it affects third-party protocol implementations
> — which the per-field wording actively invites. An implementer who reads the
> schema and resends only the list that changed loses all revocation
> acknowledgements, not just the omitted role's.
> h3. 6) Contrast with two places that get this right
> - _KIP-848._ `ConsumerGroupHeartbeatRequest` has a *single* nullable
> `TopicPartitions` field with the same convention ({*}"null if it didn't
> change since the last heartbeat"{*}), and the consumer-side
> `CurrentAssignmentBuilder` applies the same conservative reading (`if
> (ownedTopicPartitions == null) return true;`). With one field the partial
> state is unrepresentable, so this cannot arise. The streams protocol
> implements the same mechanism, split across three fields.
> - {_}`ProcessId`{_}, in this same request, carries the identical *"Null if
> unchanged since last heartbeat"* wording and {*}{{*}}is{{*}}{*} honoured per
> field (`maybeUpdateProcessId(Optional.ofNullable(processId))`) — because that
> value is persisted in the member record and so survives a coordinator
> failover. The convention works for persisted fields and breaks for transient
> ones; owned tasks are transient.
> h2. Options considered
> Relevant compatibility fact: `StreamsGroupHeartbeatRequest` is
> `validVersions: 0-1` with no `latestVersionUnstable`, so both versions are
> shipped and stable (talking AK 4.4 already even if not release yet).
> _A. Persist partial revocation progress._ Make the acknowledgement per-role:
> subtract the released tasks of each reported role from
> `tasksPendingRevocation` and persist the shrunken set. Failover is then a
> non-issue because progress is in the log. Note that today a partial release
> records *nothing* — `CurrentAssignmentBuilder` returns the same member object
> while `hasNotReleased` is true, and `maybeReconcile` only appends a record
> when the member actually changed — so this adds current-assignment record
> writes on the heartbeat path, up to one per role per revocation instead of
> one per revocation. Steady-state cost increase; likely the hardest to land.
> _B. Accumulate partial reports in memory and persist only on completion._
> Avoids A's extra writes, but the accumulated progress is transient and is
> lost on coordinator failover, with no event the client can observe (its own
> assignment has not changed, so it has no reason to resend). Recovering
> therefore needs a broker→client "resend the full report" signal — the shape
> already used by `TopologyDescriptionRequired` in the heartbeat response. Adds
> transient broker state *and* protocol surface to serve a case no client
> produces.
> _C. Document the all-or-nothing contract and reject violations._ Reword the
> three owned-task fields to say they are reported together, all or none, and
> have the broker reject a partial report with an explicit error instead of
> absorbing it into a rebalance that never finishes. No format change; makes
> the prose match the implementation and both clients; turns a silent hang into
> something diagnosable. Leaves a wire format that still *permits* the illegal
> state, i.e. the contract is prose rather than structure.
> _D. Change the request shape in a new version._ Replace the three lists with
> a single nullable outer field containing three non-nullable inner lists, so
> "nothing changed" is the outer null and a partial report is unrepresentable —
> restoring the property KIP-848 has. This is the correct shape, but since v0
> and v1 are stable it is a {_}v2 addition{_}: the flat three-field path has to
> be retained for v0/v1 clients, so the rounding behaviour survives for them
> regardless.
> C and D compose: fix the wording and reject now; if the request shape is ever
> revised for another reason, adopt the nested form and C's prose becomes
> redundant.
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
