Matthias J. Sax created KAFKA-20870:
---------------------------------------
Summary: 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
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
## Description
### 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.
### What the broker does
In the streams reconciliation path, `GroupMetadataManager` converts the request
into a `TasksTuple` only when all three lists are non-null:
```java
TasksTuple ownedTasks = null;
if (ownedActiveTasks != null && ownedStandbyTasks != null && ownedWarmupTasks
!= null) {
ownedTasks = TasksTuple.fromHeartbeatRequest(ownedActiveTasks,
ownedStandbyTasks, ownedWarmupTasks);
}
```
`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:
```java
return ownedTasks.isEmpty() ||
ownedTasks.get().containsAny(tasksPendingRevocation);
```
(`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.
### 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`.
### 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*.
### 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.
### 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.
---
## 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)
