[
https://issues.apache.org/jira/browse/KAFKA-20870?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=18101487#comment-18101487
]
Matthias J. Sax commented on KAFKA-20870:
-----------------------------------------
Thanks for taking a look. You are right. I did miss this guard. – That's
actually good news, as it make the issue less pressing to address.
> 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.
> 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)
