mjsax commented on code in PR #23488:
URL: https://github.com/apache/kafka/pull/23488#discussion_r4049941401
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/streams/AssignmentRefinerImpl.java:
##########
@@ -221,46 +262,112 @@ static TaskDecisions analyzeTasks(
// was removed from the group in the meantime. For this case, all
previously owned tasks of this member
// (which did not get move to a new owner) will be "dandling"
which will be fixed by the next assignor run.
// Furthermore, we stage all tasks the assignor moves to this
member on their old owners to keep them
- // "online".
+ // "online". A task nobody holds as an active task is put on a
caught-up copy holder instead, which is the
+ // only way it runs at all before the next assignor run.
final StreamsGroupMember targetMember = members.get(targetOwner);
if (targetMember == null) {
- if (holder != null) {
- stagedMigrations.add(new StagedMigration(
- task,
- holder.memberId(),
- targetOwner,
- Optional.empty(),
- Optional.empty()
- ));
- }
+ final Optional<String> currentOwner = holder != null
+ ? Optional.of(holder.memberId())
+ : bestCopyToPromote(currentAssignment, task,
targetStandbyHolders, processLoad);
+ currentOwner.ifPresent(owner -> stagedMigrations.add(
+ stagedMigration(currentAssignment, task, owner,
targetOwner, Optional.empty())));
continue;
}
final String targetProcessId = targetMember.processId();
+ final Optional<String> currentProcessId =
Optional.ofNullable(holder)
+ .map(activeHolder ->
members.get(activeHolder.memberId()).processId());
- // The task moves now, for any of three reasons.
- // 1. Nobody holds it.
- // 2. Somebody holds it but is still restoring it.
- // 3. Somebody is processing it and the target owner is caught up
- if (holder == null
- || !holder.processing()
- || isReady(currentAssignment, task,
members.get(holder.memberId()).processId(), targetProcessId)) {
+ if (isReady(currentAssignment, task, currentProcessId,
targetProcessId)) {
grantedTasks.add(new TaskGrant(task, targetOwner));
continue;
}
- stagedMigrations.add(new StagedMigration(
- task,
- holder.memberId(),
- targetOwner,
- Optional.of(targetProcessId),
- findCopyOnProcess(currentAssignment, task, targetProcessId)
- ));
+ // Which member the task keeps running on until the target owner
is warm. Empty means nobody has state
+ // worth keeping the task on, so the target owner takes it over
cold.
+ final Optional<String> currentOwner;
+ if (holder != null && holder.hot()) {
+ currentOwner = Optional.of(holder.memberId());
+ } else if (holder == null && onDisk(currentAssignment,
targetProcessId, task)) {
+ // The target owner's process left this task's state on disk
and can reopen it. How far behind that
+ // state is cannot be measured -- a member reports an end
offset only for a task it is restoring --
Review Comment:
Yes we could
--
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]
