This is an automated email from the ASF dual-hosted git repository.
mjsax pushed a commit to branch trunk
in repository https://gitbox.apache.org/repos/asf/kafka.git
The following commit(s) were added to refs/heads/trunk by this push:
new 553f6635a93 KAFKA-20665: Promote a warm-up task to active in place in
the reconciler (#22923)
553f6635a93 is described below
commit 553f6635a934f76e96570341b3acb2a257aa7055
Author: Matthias J. Sax <[email protected]>
AuthorDate: Tue Jul 28 10:52:48 2026 -0700
KAFKA-20665: Promote a warm-up task to active in place in the reconciler
(#22923)
When the target assignment moves a task to a member that currently holds
it as a warm-up, the reconciler now promotes the warm-up to active in
place instead of revoking it and reassigning the active in a later step.
This lets the member recycle the task's (in-memory) state store instead
of closing it and restoring from the changelog (cf. KAFKA-9501).
CurrentAssignmentBuilder:
- The active-task "unreleased" predicate no longer treats a warm-up
held by *this* member for the same task as a blocker (that is the
promotion).
It still blocks when a *different* member on the same process holds
the warm-up, so a process never runs a task as both active and warm-up.
- applyWarmupPromotions() keeps the warm-up as an assigned task
while the active is still owned elsewhere, or while its grant is deferred
by another pending revocation (buildNewMember gives revocation
priority), and drops it only in the step in which the active is actually
granted.
Reviewers: Sean Quah <[email protected]>
---
.../group/streams/CurrentAssignmentBuilder.java | 122 +++++++++++--
.../streams/CurrentAssignmentBuilderTest.java | 199 +++++++++++++++++++++
2 files changed, 306 insertions(+), 15 deletions(-)
diff --git
a/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/streams/CurrentAssignmentBuilder.java
b/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/streams/CurrentAssignmentBuilder.java
index 67c76616e73..8544bed067b 100644
---
a/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/streams/CurrentAssignmentBuilder.java
+++
b/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/streams/CurrentAssignmentBuilder.java
@@ -178,12 +178,10 @@ public class CurrentAssignmentBuilder {
// owned tasks set in the StreamsGroupHeartbeat API.
// If the member provides its owned tasks, we verify if it
still
- // owns any of the revoked tasks. If it did not provide it's
+ // owns any of the revoked tasks. If it did not provide its
// owned tasks, or we still own some of the revoked tasks, we
// cannot progress.
- if (
- ownedTasks.isEmpty() ||
ownedTasks.get().containsAny(member.tasksPendingRevocation())
- ) {
+ if (hasNotReleased(member.tasksPendingRevocation())) {
return member;
}
@@ -426,12 +424,20 @@ public class CurrentAssignmentBuilder {
newActiveAssignedTasks,
newActiveTasksPendingRevocation,
newActiveTasksPendingAssignment,
+ // In general, an active task can only be assigned once its
previous owner has released it.
+ // In addition, we cannot assign an active task to a _process_
that still holds a corresponding
+ // standby or warm-up task, because a single process must not run
the same task twice.
+ // The one exception is an in-place warm-up promotion: if THIS
MEMBER holds the task as a warm-up,
+ // we convert it to active in a single step (drop the warm-up,
grant the active)
(subtopologyId, partitionId) ->
+ // still owned as active by its previous owner
currentActiveTaskProcessId.apply(subtopologyId, partitionId)
!= null ||
- currentStandbyTaskProcessIds.apply(subtopologyId,
partitionId)
- .contains(member.processId()) ||
- currentWarmupTaskProcessIds.apply(subtopologyId,
partitionId)
- .contains(member.processId())
+ // this member's process still holds it as a standby
+ currentStandbyTaskProcessIds.apply(subtopologyId,
partitionId).contains(member.processId()) ||
+ // this member's process holds it as a warm-up via a
*different* member (if this member owns
+ // the warm-up it is a promotion, which is allowed)
+ (currentWarmupTaskProcessIds.apply(subtopologyId,
partitionId).contains(member.processId())
+ &&
!memberAssignedTasks.warmupTasks().getOrDefault(subtopologyId,
Set.of()).contains(partitionId))
);
boolean hasUnreleasedStandbyTasks = computeAssignmentDifference(
@@ -464,13 +470,18 @@ public class CurrentAssignmentBuilder {
.contains(member.processId())
);
+ TasksTupleWithEpochs newTasksPendingRevocation = applyWarmupPromotions(
+ memberAssignedTasks,
+ newActiveTasksPendingRevocation,
+ newStandbyTasksPendingRevocation,
+ newWarmupTasksPendingRevocation,
+ newActiveTasksPendingAssignment,
+ newWarmupAssignedTasks // modified in-place by
applyWarmupPromotions
+ );
+
return buildNewMember(
memberEpoch,
- new TasksTupleWithEpochs(
- newActiveTasksPendingRevocation,
- newStandbyTasksPendingRevocation,
- newWarmupTasksPendingRevocation
- ),
+ newTasksPendingRevocation,
new TasksTupleWithEpochs(
newActiveAssignedTasks,
newStandbyAssignedTasks,
@@ -485,6 +496,87 @@ public class CurrentAssignmentBuilder {
);
}
+ /**
+ * Applies warm-up promotions to the freshly computed reconciliation
result. A task the member holds as a
+ * warm-up that the target wants it to own as an active task is promoted
in place rather than
+ * revoked-then-reassigned, so its recyclable state store survives instead
of being closed and restored from
+ * the changelog (cf. KAFKA-9501). Such a warm-up is never routed through
the (ack-based) revocation path.
+ *
+ * @return the member's pending revocation after the promoted warm-ups
have been removed from it.
+ */
+ private TasksTupleWithEpochs applyWarmupPromotions(TasksTupleWithEpochs
memberAssignedTasks,
+ Map<String,
Map<Integer, Integer>> newActiveTasksPendingRevocation,
+ Map<String,
Set<Integer>> newStandbyTasksPendingRevocation,
+ Map<String,
Set<Integer>> newWarmupTasksPendingRevocation,
+ Map<String,
Map<Integer, Integer>> newActiveTasksPendingAssignment,
+ Map<String,
Set<Integer>> newWarmupAssignedTasks) {
+ // If we promote a warm-up to active, the warm-up does not need an
explicit client-side revocation.
+ // Thus, the warm-up can be removed from
`newWarmupTasksPendingRevocation` -- we don't expect a client ack back.
+ // We need to remove it regardless of wether the warm-up to active
promotion happens now, or is still pending on
+ // the active task revocation.
+ for (Map.Entry<String, Set<Integer>> warmup :
memberAssignedTasks.warmupTasks().entrySet()) {
+ String subtopologyId = warmup.getKey();
+ Set<Integer> targetActiveTasks =
targetAssignment.activeTasks().getOrDefault(subtopologyId, Set.of());
+ for (Integer partitionId : warmup.getValue()) {
+ if (targetActiveTasks.contains(partitionId)) {
+ removeFromTaskSet(newWarmupTasksPendingRevocation,
subtopologyId, partitionId);
+ }
+ }
+ }
+
+ TasksTupleWithEpochs newTasksPendingRevocation = new
TasksTupleWithEpochs(
+ newActiveTasksPendingRevocation,
+ newStandbyTasksPendingRevocation,
+ newWarmupTasksPendingRevocation
+ );
+ boolean hasTasksToBeRevoked = !newTasksPendingRevocation.isEmpty()
+ && hasNotReleased(newTasksPendingRevocation);
+
+ // keep warm-up task and don't revoke if we are not ready for warm-up
to active promotion
+ // - this member still needs to complete its own revocation of other
tasks
+ // (which must complete before any assignment can happen)
+ // - the active task was not released by its previous owner yet
+ for (Map.Entry<String, Set<Integer>> warmup :
memberAssignedTasks.warmupTasks().entrySet()) {
+ String subtopologyId = warmup.getKey();
+ Set<Integer> targetActiveTasks =
targetAssignment.activeTasks().getOrDefault(subtopologyId, Set.of());
+ Map<Integer, Integer> grantedActiveTasks =
newActiveTasksPendingAssignment.getOrDefault(subtopologyId, Map.of());
+ for (Integer partitionId : warmup.getValue()) {
+ boolean promotedThisStep = !hasTasksToBeRevoked &&
grantedActiveTasks.containsKey(partitionId);
+ if (targetActiveTasks.contains(partitionId) &&
!promotedThisStep) {
+ newWarmupAssignedTasks.computeIfAbsent(subtopologyId, __
-> new HashSet<>()).add(partitionId);
+ }
+ }
+ }
+
+ return newTasksPendingRevocation;
+ }
+
+ /**
+ * Removes a single subtopology/partition from a task set, dropping the
subtopology entry entirely if it
+ * becomes empty (so the resulting map matches what the difference helpers
would have produced).
+ */
+ private static void removeFromTaskSet(Map<String, Set<Integer>> tasks,
String subtopologyId, Integer partitionId) {
+ Set<Integer> partitions = tasks.get(subtopologyId);
+ if (partitions != null && partitions.remove(partitionId) &&
partitions.isEmpty()) {
+ tasks.remove(subtopologyId);
+ }
+ }
+
+ /**
+ * Checks whether the member has not yet confirmed the release of the
given tasks pending revocation.
+ * Revocation is ack-based: it is confirmed only once the member reports
its currently owned tasks in a
+ * heartbeat and none of the given tasks appear among them.
+ *
+ * @param tasksPendingRevocation The tasks whose revocation we are waiting
for.
+ * @return true if the release of any of those tasks is not yet confirmed.
+ */
+ private boolean hasNotReleased(TasksTupleWithEpochs
tasksPendingRevocation) {
+ // Note that {@code ownedTasks} being empty means the member did not
report its owned tasks in this
+ // heartbeat -- it is an {@link Optional}, not an empty task set.
Without such a report we cannot prove
+ // the tasks were released, so we conservatively treat them as still
held.
+ return ownedTasks.isEmpty() ||
ownedTasks.get().containsAny(tasksPendingRevocation);
+ }
+
private StreamsGroupMember buildNewMember(final int memberEpoch,
final TasksTupleWithEpochs
newTasksPendingRevocation,
final TasksTupleWithEpochs
newAssignedTasks,
@@ -492,8 +584,8 @@ public class CurrentAssignmentBuilder {
final boolean
hasUnreleasedTasks) {
final boolean hasTasksToBeRevoked =
- (!newTasksPendingRevocation.isEmpty())
- && (ownedTasks.isEmpty() ||
ownedTasks.get().containsAny(newTasksPendingRevocation));
+ !newTasksPendingRevocation.isEmpty()
+ && hasNotReleased(newTasksPendingRevocation);
if (hasTasksToBeRevoked) {
// If there are tasks to be revoked, the member remains in its
current
diff --git
a/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/streams/CurrentAssignmentBuilderTest.java
b/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/streams/CurrentAssignmentBuilderTest.java
index b383a35f786..3473d90f4ac 100644
---
a/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/streams/CurrentAssignmentBuilderTest.java
+++
b/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/streams/CurrentAssignmentBuilderTest.java
@@ -974,4 +974,203 @@ public class CurrentAssignmentBuilderTest {
updatedMember
);
}
+
+ @Test
+ public void
testWarmupPromotionHoldsWarmupWhileActiveOwnedByAnotherMember() {
+ final int memberEpoch = 10;
+ final String otherProcessId = "other_process_id";
+
+ StreamsGroupMember member = new StreamsGroupMember.Builder(MEMBER_NAME)
+ .setState(MemberState.STABLE)
+ .setProcessId(PROCESS_ID)
+ .setMemberEpoch(memberEpoch)
+ .setPreviousMemberEpoch(memberEpoch)
+ .setAssignedTasks(mkTasksTupleWithCommonEpoch(TaskRole.WARMUP,
memberEpoch,
+ mkTasks(SUBTOPOLOGY_ID1, 0)))
+ .setTasksPendingRevocation(TasksTupleWithEpochs.EMPTY)
+ .build();
+
+ // The target wants this member to own the task as an active task, but
another member still owns it
+ // active. The member keeps the warm-up (so it stays caught up) and
waits in UNRELEASED_TASKS -- it is
+ // not asked to revoke the warm-up.
+ StreamsGroupMember updatedMember = new CurrentAssignmentBuilder(member)
+ .withTargetAssignment(memberEpoch + 1,
mkTasksTuple(TaskRole.ACTIVE,
+ mkTasks(SUBTOPOLOGY_ID1, 0)))
+ .withCurrentActiveTaskProcessId((subtopologyId, partitionId) ->
otherProcessId)
+ .withCurrentStandbyTaskProcessIds((subtopologyId, partitionId) ->
Set.of())
+ .withCurrentWarmupTaskProcessIds((subtopologyId, partitionId) ->
Set.of(PROCESS_ID))
+ .build();
+
+ assertEquals(
+ new StreamsGroupMember.Builder(MEMBER_NAME)
+ .setState(MemberState.UNRELEASED_TASKS)
+ .setProcessId(PROCESS_ID)
+ .setMemberEpoch(memberEpoch + 1)
+ .setPreviousMemberEpoch(memberEpoch)
+ .setAssignedTasks(mkTasksTupleWithCommonEpoch(TaskRole.WARMUP,
memberEpoch,
+ mkTasks(SUBTOPOLOGY_ID1, 0)))
+ .setTasksPendingRevocation(TasksTupleWithEpochs.EMPTY)
+ .build(),
+ updatedMember
+ );
+ }
+
+ @Test
+ public void testWarmupPromotedToActiveWhenActiveIsReleased() {
+ final int memberEpoch = 10;
+
+ StreamsGroupMember member = new StreamsGroupMember.Builder(MEMBER_NAME)
+ .setState(MemberState.STABLE)
+ .setProcessId(PROCESS_ID)
+ .setMemberEpoch(memberEpoch)
+ .setPreviousMemberEpoch(memberEpoch)
+ .setAssignedTasks(mkTasksTupleWithCommonEpoch(TaskRole.WARMUP,
memberEpoch,
+ mkTasks(SUBTOPOLOGY_ID1, 0)))
+ .setTasksPendingRevocation(TasksTupleWithEpochs.EMPTY)
+ .build();
+
+ // The target wants this member to own the task active and no other
member owns it active any more, so
+ // the warm-up is promoted in place: the active is granted and the
warm-up is dropped in one step
+ // (no revoke-then-reassign round trip).
+ StreamsGroupMember updatedMember = new CurrentAssignmentBuilder(member)
+ .withTargetAssignment(memberEpoch + 1,
mkTasksTuple(TaskRole.ACTIVE,
+ mkTasks(SUBTOPOLOGY_ID1, 0)))
+ .withCurrentActiveTaskProcessId((subtopologyId, partitionId) ->
null)
+ .withCurrentStandbyTaskProcessIds((subtopologyId, partitionId) ->
Set.of())
+ .withCurrentWarmupTaskProcessIds((subtopologyId, partitionId) ->
Set.of(PROCESS_ID))
+ .build();
+
+ assertEquals(
+ new StreamsGroupMember.Builder(MEMBER_NAME)
+ .setState(MemberState.STABLE)
+ .setProcessId(PROCESS_ID)
+ .setMemberEpoch(memberEpoch + 1)
+ .setPreviousMemberEpoch(memberEpoch)
+ .setAssignedTasks(mkTasksTupleWithCommonEpoch(TaskRole.ACTIVE,
memberEpoch + 1,
+ mkTasks(SUBTOPOLOGY_ID1, 0)))
+ .setTasksPendingRevocation(TasksTupleWithEpochs.EMPTY)
+ .build(),
+ updatedMember
+ );
+ }
+
+ @Test
+ public void testWarmupPromotedToActiveFromUnreleasedState() {
+ final int memberEpoch = 11;
+
+ // The member already advanced to the target epoch holding the
warm-up, waiting for the previous owner
+ // to release the active task.
+ StreamsGroupMember member = new StreamsGroupMember.Builder(MEMBER_NAME)
+ .setState(MemberState.UNRELEASED_TASKS)
+ .setProcessId(PROCESS_ID)
+ .setMemberEpoch(memberEpoch)
+ .setPreviousMemberEpoch(memberEpoch - 1)
+ .setAssignedTasks(mkTasksTupleWithCommonEpoch(TaskRole.WARMUP,
memberEpoch,
+ mkTasks(SUBTOPOLOGY_ID1, 0)))
+ .setTasksPendingRevocation(TasksTupleWithEpochs.EMPTY)
+ .build();
+
+ StreamsGroupMember updatedMember = new CurrentAssignmentBuilder(member)
+ .withTargetAssignment(memberEpoch, mkTasksTuple(TaskRole.ACTIVE,
+ mkTasks(SUBTOPOLOGY_ID1, 0)))
+ .withCurrentActiveTaskProcessId((subtopologyId, partitionId) ->
null)
+ .withCurrentStandbyTaskProcessIds((subtopologyId, partitionId) ->
Set.of())
+ .withCurrentWarmupTaskProcessIds((subtopologyId, partitionId) ->
Set.of(PROCESS_ID))
+ .build();
+
+ assertEquals(
+ new StreamsGroupMember.Builder(MEMBER_NAME)
+ .setState(MemberState.STABLE)
+ .setProcessId(PROCESS_ID)
+ .setMemberEpoch(memberEpoch)
+ .setPreviousMemberEpoch(memberEpoch)
+ .setAssignedTasks(mkTasksTupleWithCommonEpoch(TaskRole.ACTIVE,
memberEpoch,
+ mkTasks(SUBTOPOLOGY_ID1, 0)))
+ .setTasksPendingRevocation(TasksTupleWithEpochs.EMPTY)
+ .build(),
+ updatedMember
+ );
+ }
+
+ @Test
+ public void
testActiveNotGrantedWhileAnotherMemberOnSameProcessHoldsWarmup() {
+ final int memberEpoch = 10;
+
+ // This member does not hold the warm-up itself; a *different* member
on the same process does (reflected
+ // by currentWarmupTaskProcessIds returning this member's process).
The active must not be granted -- the
+ // process would otherwise run the task as both active (this member)
and warm-up (the sibling). Only the
+ // member that itself holds the warm-up is promoted; this one waits.
+ StreamsGroupMember member = new StreamsGroupMember.Builder(MEMBER_NAME)
+ .setState(MemberState.STABLE)
+ .setProcessId(PROCESS_ID)
+ .setMemberEpoch(memberEpoch)
+ .setPreviousMemberEpoch(memberEpoch)
+ .setAssignedTasks(TasksTupleWithEpochs.EMPTY)
+ .setTasksPendingRevocation(TasksTupleWithEpochs.EMPTY)
+ .build();
+
+ StreamsGroupMember updatedMember = new CurrentAssignmentBuilder(member)
+ .withTargetAssignment(memberEpoch + 1,
mkTasksTuple(TaskRole.ACTIVE,
+ mkTasks(SUBTOPOLOGY_ID1, 0)))
+ .withCurrentActiveTaskProcessId((subtopologyId, partitionId) ->
null)
+ .withCurrentStandbyTaskProcessIds((subtopologyId, partitionId) ->
Set.of())
+ .withCurrentWarmupTaskProcessIds((subtopologyId, partitionId) ->
Set.of(PROCESS_ID))
+ .build();
+
+ assertEquals(
+ new StreamsGroupMember.Builder(MEMBER_NAME)
+ .setState(MemberState.UNRELEASED_TASKS)
+ .setProcessId(PROCESS_ID)
+ .setMemberEpoch(memberEpoch + 1)
+ .setPreviousMemberEpoch(memberEpoch)
+ .setAssignedTasks(TasksTupleWithEpochs.EMPTY)
+ .setTasksPendingRevocation(TasksTupleWithEpochs.EMPTY)
+ .build(),
+ updatedMember
+ );
+ }
+
+ @Test
+ public void
testWarmupNotDroppedWhilePromotionIsBlockedByAnotherRevocation() {
+ final int memberEpoch = 10;
+
+ // The member holds a warm-up for one task (which the target wants it
to own active, and that active is
+ // free -> promotable) AND an active task the target no longer wants
(-> pending revocation). Because any
+ // pending revocation keeps the member in UNREVOKED_TASKS, the
promotion cannot be granted this step; the
+ // warm-up must be KEPT (its recyclable state preserved), not dropped
in anticipation of a grant that does
+ // not happen.
+ StreamsGroupMember member = new StreamsGroupMember.Builder(MEMBER_NAME)
+ .setState(MemberState.STABLE)
+ .setProcessId(PROCESS_ID)
+ .setMemberEpoch(memberEpoch)
+ .setPreviousMemberEpoch(memberEpoch)
+ .setAssignedTasks(new TasksTupleWithEpochs(
+ Map.of(SUBTOPOLOGY_ID1, Map.of(1, memberEpoch)), // active
task the target no longer wants
+ Map.of(),
+ Map.of(SUBTOPOLOGY_ID1, Set.of(0)))) // warm-up
the target wants promoted to active
+ .setTasksPendingRevocation(TasksTupleWithEpochs.EMPTY)
+ .build();
+
+ StreamsGroupMember updatedMember = new CurrentAssignmentBuilder(member)
+ .withTargetAssignment(memberEpoch + 1,
mkTasksTuple(TaskRole.ACTIVE,
+ mkTasks(SUBTOPOLOGY_ID1, 0)))
+ .withCurrentActiveTaskProcessId((subtopologyId, partitionId) ->
partitionId == 1 ? PROCESS_ID : null)
+ .withCurrentStandbyTaskProcessIds((subtopologyId, partitionId) ->
Set.of())
+ .withCurrentWarmupTaskProcessIds((subtopologyId, partitionId) ->
partitionId == 0 ? Set.of(PROCESS_ID) : Set.of())
+ .build();
+
+ assertEquals(
+ new StreamsGroupMember.Builder(MEMBER_NAME)
+ .setState(MemberState.UNREVOKED_TASKS)
+ .setProcessId(PROCESS_ID)
+ .setMemberEpoch(memberEpoch)
+ .setPreviousMemberEpoch(memberEpoch)
+ .setAssignedTasks(mkTasksTupleWithCommonEpoch(TaskRole.WARMUP,
memberEpoch,
+ mkTasks(SUBTOPOLOGY_ID1, 0))) //
warm-up kept, not dropped
+
.setTasksPendingRevocation(mkTasksTupleWithCommonEpoch(TaskRole.ACTIVE,
memberEpoch,
+ mkTasks(SUBTOPOLOGY_ID1, 1)))
+ .build(),
+ updatedMember
+ );
+ }
}
