Ziyun Fu created KAFKA-20846:
--------------------------------
Summary: Streams TaskAssignor receives mutable nested collections
in its GroupSpec
Key: KAFKA-20846
URL: https://issues.apache.org/jira/browse/KAFKA-20846
Project: Kafka
Issue Type: Bug
Reporter: Ziyun Fu
KAFKA-20790 promoted the broker-side \{{streams}} task assignor API to
\{{group-coordinator-api}} and marked it \{{@InterfaceAudience.Public}}, so a
\{{GroupSpec}} is now passed to third-party code running inside the group
coordinator.
The collections reachable from that spec are only shallowly protected, and
several of them alias live coordinator state. As a result, a custom assignor
can accidentally mutate coordinator-owned state through nested mutable
collections.
h2. Problem
{\{MemberMetadataAndStateImpl}} wraps each field with
\{{Collections.unmodifiableMap()}}, but several fields are now nested
collections such as \{{Map<String, Set<Integer>>}} and \{{Map<String,
Map<Integer, Long>>}}. This only protects the outer map. Calling \{{get()}}
still returns the inner mutable \{{HashSet}} or \{{HashMap}}.
This used to be safe enough when task offsets were a flat \{{Map<TaskId,
Long>}}, because the values were immutable \{{Long}} instances. That assumption
no longer holds.
In addition, \{{TargetAssignmentBuilder#createMemberMetadataAndState}} passes
several maps through without copying. For \{{standbyTasks}}, \{{warmupTasks}},
\{{taskOffsets}}, and \{{taskEndOffsets}}, the collections exposed through
\{{GroupSpec}} can alias live coordinator state.
That means code like the following can silently mutate internal coordinator
state:
{code:java}
spec.memberAssignmentState(memberId).standbyTasks().get("sub-1").remove(0);
{code}
Separately, \{{GroupSpecImpl}} does not wrap \{{configs}} at all, so calls such
as \{{groupSpec.configs().put(..)}} or \{{groupSpec.configs().clear()}} can
mutate the coordinator's assignment configs directly. Only \{{members}} is
protected today, and that protection is applied at the call site rather than
enforced by the type itself.
h2. Impact
Mutating \{{standbyTasks}} or \{{warmupTasks}} can corrupt the coordinator's
view of what a member currently owns, causing it to diverge from the log and
affecting subsequent reconciliation.
Mutating task offsets can skew the lag estimate used for warm-up promotion and
also affect concurrent describe-group responses. This corruption is transient
because the next heartbeat overwrites it, but it is still externally observable
and can influence assignment behavior in the meantime.
This is not just theoretical. \{{MockAssignor}} already relies on the inner
sets being mutable: it shallow-copies the outer map and then mutates a shared
inner \{{Set}}. No test caught this because the assertions only inspect the
output, and the output shares the mutated object with the input.
The pattern dates back to KAFKA-18319. It is harmless for \{{activeTasks}}
today only by accident, because \{{createMemberMetadataAndState}} rebuilds that
field on each cycle to strip assignment epochs, so writes land on a throwaway
object. Before KAFKA-20790, the same pattern mutated the stored target
assignment in place, and the same behavior applied to \{{standbyTasks}} or
\{{warmupTasks}} would corrupt group state today.
h2. Proposed fix
* Wrap nested collections in \{{MemberMetadataAndStateImpl}} so that every
layer handed to an assignor is unmodifiable.
* Wrap \{{configs}} and \{{members}} inside \{{GroupSpecImpl}} so the
immutability invariant is enforced by the type rather than by the call site.
* Deep-copy in \{{MockAssignor}} instead of relying on mutable input
collections.
* Document the contract in the \{{GroupSpec}} and \{{MemberAssignmentState}}
javadocs so assignor implementors know they must copy before mutating.
Doing this at the boundary rather than in \{{TasksTupleWithEpochs}} /
\{{MemberTaskOffsets}} is deliberate. \{{MemberTaskOffsets#update}} feeds an
already wrapped map back through its own constructor whenever one side is
unchanged, so wrapping there would add another \{{UnmodifiableMap}} layer per
heartbeat. Avoiding that would require changing two hot reconciliation types
from view semantics to copy semantics, which is a much larger change without
improving protection at the actual trust boundary.
The goal is not to stop a malicious assignor. Anyone who can load arbitrary
code into the broker can do much worse. The goal is to turn an easy, silent
state corruption bug into a safe read-only API boundary for normal third-party
assignor implementations.
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
