[
https://issues.apache.org/jira/browse/KAFKA-20870?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=18100616#comment-18100616
]
Matthias J. Sax commented on KAFKA-20870:
-----------------------------------------
I would first like to hear from [~dajac] [~squah-confluent] [~lianetm]
[~lucasbru] about this ticket and what they think. – We should first agree on a
direction forward...
> 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
>
> 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)
