mjsax commented on code in PR #22783:
URL: https://github.com/apache/kafka/pull/22783#discussion_r3575133937
##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamThread.java:
##########
@@ -967,6 +967,10 @@ public void run() {
* @throws StreamsException if the store's change log does not
contain the partition
*/
boolean runLoop() {
+ // Populate the task-offset-sum snapshot before subscribing:
subscribing triggers the join heartbeat,
+ // which must already carry the offset sums of tasks discovered in the
local state directory, so that
+ // the broker-side sticky assignor can assign those tasks back to this
client on a cold start.
+ taskManager.maybeUpdateTaskOffsetSumSnapshot();
Review Comment:
I am just realizing that we need this only for "stream" case -- should we
make it a no-op for "classic"?
(If yes, we might put the corresponding code inside
`maybeUpdateTaskOffsetSumSnapshot()` such that the caller does not need to
think about it.)
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/streams/assignor/StickyTaskAssignor.java:
##########
@@ -133,6 +133,43 @@ private void initialize(final GroupSpec groupSpec, final
TopologyDescriber topol
}
}
}
+
+ maybeAddOffsetBasedPrevMembers(groupSpec);
+ }
+
+ /**
+ * A member that rejoins after a restart gets a fresh member ID and an
empty target assignment, so the previous
+ * ownership maps cannot capture that it owned any tasks before the
restart. Such members report cumulative
+ * offsets for tasks with state on local disk; treat them like previous
standbys of those tasks, so that the
+ * stickiness phases assign the tasks back to the local state, most
caught-up state first. Members already known
+ * as previous standbys keep precedence over offset-based candidates.
+ */
+ private void maybeAddOffsetBasedPrevMembers(final GroupSpec groupSpec) {
+ final Map<TaskId, ArrayList<MemberAndOffsetSum>>
tasksWithReportedOffsets = new HashMap<>();
+ for (final Map.Entry<String, AssignmentMemberSpec> memberEntry :
groupSpec.members().entrySet()) {
Review Comment:
It seem we could fold this loop into `initialize()` directly, as it does
iterator over `groupSpec.members().entrySet()` already, to loop over all
members only once instead of twice?
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/streams/assignor/StickyTaskAssignor.java:
##########
@@ -133,6 +133,43 @@ private void initialize(final GroupSpec groupSpec, final
TopologyDescriber topol
}
}
}
+
+ maybeAddOffsetBasedPrevMembers(groupSpec);
+ }
+
+ /**
+ * A member that rejoins after a restart gets a fresh member ID and an
empty target assignment, so the previous
+ * ownership maps cannot capture that it owned any tasks before the
restart. Such members report cumulative
+ * offsets for tasks with state on local disk; treat them like previous
standbys of those tasks, so that the
+ * stickiness phases assign the tasks back to the local state, most
caught-up state first. Members already known
+ * as previous standbys keep precedence over offset-based candidates.
+ */
+ private void maybeAddOffsetBasedPrevMembers(final GroupSpec groupSpec) {
+ final Map<TaskId, ArrayList<MemberAndOffsetSum>>
tasksWithReportedOffsets = new HashMap<>();
+ for (final Map.Entry<String, AssignmentMemberSpec> memberEntry :
groupSpec.members().entrySet()) {
+ final AssignmentMemberSpec memberSpec = memberEntry.getValue();
+ final Member member = new Member(memberSpec.processId(),
memberEntry.getKey());
+ for (final Map.Entry<TaskId, Long> taskOffset :
memberSpec.taskOffsets().entrySet()) {
+ if (taskOffset.getValue() < 0) {
+ continue;
+ }
+ tasksWithReportedOffsets
+ .computeIfAbsent(taskOffset.getKey(), task -> new
ArrayList<>())
+ .add(new MemberAndOffsetSum(member,
taskOffset.getValue()));
+ }
+ }
+
+ for (final Map.Entry<TaskId, ArrayList<MemberAndOffsetSum>> entry :
tasksWithReportedOffsets.entrySet()) {
+ final ArrayList<Member> candidates =
+
localState.standbyTaskToPrevMember.computeIfAbsent(entry.getKey(), task -> new
ArrayList<>());
Review Comment:
With the newly reported task-offsets, we might get offsets also for
"previously assigned standby tasks".
Thus, I think we can also sort those? -- And interesting case might be an
upgrade case, for which some clients report task-offset already, while others
don't, so we only get a partial picture -- for this case, we might want to put
"prev standbys" first for backward compatibility?
If I read the code correctly, we currently get the `candidates` (which are
"prev assigned tasks"), and add to the end of the all other candidates (in
descending offset order). -- This could also introduce duplicate entries? I
believe?
Maybe we can assemble all candidates (ie, "prev assigned" and task-offset
based) in one go in the first loop (and de-duplicate already?), and do a single
final sort at the end?
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/streams/assignor/StickyTaskAssignor.java:
##########
Review Comment:
```suggestion
ret.computeIfAbsent(taskId.subtopologyId(), () -> new HashSet<>())
.add(taskId.partition());
```
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/streams/assignor/StickyTaskAssignor.java:
##########
Review Comment:
Side improvement -- should be slightly more performant, and squeezing the
assignor which runs in the GC might be worth it.
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/streams/assignor/StickyTaskAssignor.java:
##########
Review Comment:
Same
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/streams/assignor/StickyTaskAssignor.java:
##########
Review Comment:
```suggestion
localState.standbyTaskToPrevMember
.computeIfAbsent(taskId, () -> new
ArrayList<>(localState.numStandbyReplicas))
.add(member);
```
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/streams/assignor/StickyTaskAssignor.java:
##########
Review Comment:
```suggestion
localState.processIdToState
.computeIfAbsent(processId, () -> new ProcessState(processId))
.addMember(memberId);
```
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/streams/assignor/StickyTaskAssignor.java:
##########
Review Comment:
Side improvement.
--
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]
