[
https://issues.apache.org/jira/browse/KAFKA-20846?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Ziyun Fu updated KAFKA-20846:
-----------------------------
Component/s: streams
> 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
> Components: streams
> Reporter: Ziyun Fu
> Assignee: Ziyun Fu
> Priority: Minor
>
> 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)
