chickenchickenlove commented on code in PR #23742:
URL: https://github.com/apache/kafka/pull/23742#discussion_r4372916165
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/modern/TargetAssignmentBuilder.java:
##########
@@ -158,10 +158,13 @@ private Assignment newMemberAssignment(
String memberId
) {
MemberAssignment newMemberAssignment =
newGroupAssignment.members().get(memberId);
- if (newMemberAssignment != null) {
- return new Assignment(newMemberAssignment.partitions());
- } else {
+ if (newMemberAssignment == null) {
return Assignment.EMPTY;
}
+
+ // The output is only read by the coordinator. Assignor inputs are
+ // reconstructed from target assignment records during replay.
+ // So, we don't need defensive copy here.
+ return Assignment.of(newMemberAssignment.partitions());
Review Comment:
@squah-confluent @dajac
Thanks for pointing out the offloading case. I agree that
`Assignment.copyOf()` would protect against subsequent mutations after the copy
completes. I was wondering how far we would like to extend that protection in
this change.
With assignment offloading, another `assign()` invocation on the same
assignor may run before records are built from the previous result. If
invocations run concurrently, the original maps or sets could also be modified
while `Assignment.copyOf()` is copying them. Protecting against that case would
require copying and mutations to coordinate through the same synchronization
mechanism.
My understanding is that case (1) arises when an assignor retains references
to objects it has returned and reuses or modifies them in subsequent
invocations. If we require assignors not to modify returned maps or sets after
returning them, `Assignment.of()` should be sufficient here. Otherwise,
`Assignment.copyOf()` would provide additional protection, although it comes
with a copying cost and does not fully address concurrent mutations during
copying.
Separately, I think assignments reconstructed during replay still need to be
immutable. This prevents an assignor from modifying the coordinator’s in-memory
state through its inputs without corresponding records.
Would it make sense to clarify whether this change should focus on
protecting replayed state from mutation through assignor inputs, or also
provide protection against assignors modifying their outputs after returning
them?
What do you think?
(AFAIK, Assignment Offloading hasn't been integrated yet, so I'm responding
based on my understanding of how it might conceptually be applied in
`maybeUpdateTargetAssignment()`. 😅 )
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
