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 07e1f099aca KAFKA-20665: Change task roles in place in the reconciler
(#22981)
07e1f099aca is described below
commit 07e1f099acad26200b2681411e05616bd3fe7879
Author: Matthias J. Sax <[email protected]>
AuthorDate: Thu Jul 30 07:28:33 2026 -0700
KAFKA-20665: Change task roles in place in the reconciler (#22981)
Follow up to https://github.com/apache/kafka/pull/22748
Extends the in-place warm-up promotion to every role change a member can
make for a task it already holds, so that the client recycles the task
instead of closing and re-creating it (cf. KAFKA-9501):
- standby -> active and warm-up -> active (promotion): the member
keeps the standby or warm-up task until the active task is actually
granted, because that grant waits for the previous owner to release it.
- active -> standby (demotion): the standby task is granted in the same
step in which the active task is revoked. The active task still needs an
acknowledgement, so that the group observes its release before the task
is granted to its next owner.
- standby <-> warm-up (conversion): both roles run the same code
on the client, so the roles are swapped in a single step and no
revocation acknowledgement is needed at all.
Each of the three "unreleased task" predicates blocks a grant while the
member's *process* holds the task in another role, which is what keeps a
process from running the same task twice. These checks are now
member-precise: a task the member holds itself is a candidate for a role
change rather than a blocker, while a task held by a sibling member on
the same process still blocks.
Reviewers: Sean Quah <[email protected]>
---
.../group/streams/CurrentAssignmentBuilder.java | 311 ++++++++++++++----
.../streams/CurrentAssignmentBuilderTest.java | 352 +++++++++++++++++++++
.../integration/RestoreIntegrationTest.java | 9 +-
3 files changed, 609 insertions(+), 63 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 8544bed067b..0b256a82089 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
@@ -424,20 +424,7 @@ 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 ||
- // 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))
+ (subtopologyId, partitionId) ->
isUnreleasedActiveTask(memberAssignedTasks, subtopologyId, partitionId)
);
boolean hasUnreleasedStandbyTasks = computeAssignmentDifference(
@@ -446,13 +433,7 @@ public class CurrentAssignmentBuilder {
newStandbyAssignedTasks,
newStandbyTasksPendingRevocation,
newStandbyTasksPendingAssignment,
- (subtopologyId, partitionId) ->
- Objects.equals(currentActiveTaskProcessId.apply(subtopologyId,
partitionId),
- member.processId()) ||
- currentStandbyTaskProcessIds.apply(subtopologyId,
partitionId)
- .contains(member.processId()) ||
- currentWarmupTaskProcessIds.apply(subtopologyId,
partitionId)
- .contains(member.processId())
+ (subtopologyId, partitionId) ->
isUnreleasedStandbyTask(memberAssignedTasks, subtopologyId, partitionId)
);
boolean hasUnreleasedWarmupTasks = computeAssignmentDifference(
@@ -461,22 +442,45 @@ public class CurrentAssignmentBuilder {
newWarmupAssignedTasks,
newWarmupTasksPendingRevocation,
newWarmupTasksPendingAssignment,
- (subtopologyId, partitionId) ->
- Objects.equals(currentActiveTaskProcessId.apply(subtopologyId,
partitionId),
- member.processId()) ||
- currentStandbyTaskProcessIds.apply(subtopologyId,
partitionId)
- .contains(member.processId()) ||
- currentWarmupTaskProcessIds.apply(subtopologyId,
partitionId)
- .contains(member.processId())
+ (subtopologyId, partitionId) ->
isUnreleasedWarmupTask(memberAssignedTasks, subtopologyId, partitionId)
);
- TasksTupleWithEpochs newTasksPendingRevocation = applyWarmupPromotions(
- memberAssignedTasks,
- newActiveTasksPendingRevocation,
+ // A role change that does not depend on any other member takes effect
right away: the target role is moved
+ // from the pending assignment to the assigned tasks, so that the
member is told about both halves of the
+ // change in the same heartbeat and the client recycles the task
instead of closing and re-creating it.
+ // A demoted active task still stays in the revocation set, because
the group must observe its release
+ // before the task can be granted to its next owner.
+
+ // The three calls below modify the maps passed to them in place, so
they have to run before the new member is built.
+
+ // revoked active tasks are already reflected in
`newActiveTasksPendingRevocation`;
+ // we only need to take care of standby task part
+ demoteActiveTasksToStandby(memberAssignedTasks,
newStandbyTasksPendingAssignment, newStandbyAssignedTasks);
+ // convert standby tasks to warmup task
+ convertReplicaRole(
+ memberAssignedTasks.standbyTasks(),
newStandbyTasksPendingRevocation,
+ newWarmupTasksPendingAssignment,
+ newWarmupAssignedTasks
+ );
+ // convert warmup tasks to standby tasks
+ convertReplicaRole(
+ memberAssignedTasks.warmupTasks(),
newWarmupTasksPendingRevocation,
+ newStandbyTasksPendingAssignment,
+ newStandbyAssignedTasks
+ );
+
+ // A promotion to active, in contrast, has to wait for the previous
owner to release the active task, so the
+ // member keeps its standby or warm-up task until the active task is
actually granted.
+ TasksTupleWithEpochs newTasksPendingRevocation =
applyPromotionsToActive(
+ memberAssignedTasks,
+ newActiveTasksPendingRevocation,
+ newStandbyTasksPendingRevocation, // modified in-place by
applyPromotionsToActive
+ newWarmupTasksPendingRevocation, // modified in-place by
applyPromotionsToActive
newActiveTasksPendingAssignment,
- newWarmupAssignedTasks // modified in-place by
applyWarmupPromotions
+ newStandbyAssignedTasks, // modified in-place by
applyPromotionsToActive
+ newWarmupAssignedTasks // modified in-place by
applyPromotionsToActive
);
return buildNewMember(
@@ -497,32 +501,162 @@ 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.
+ * An active task can only be granted once its previous owner anywhere in
the group has released it -- a task has
+ * exactly one active owner. It can also not be granted to a _process_
that still holds the task as a standby or
+ * warm-up, because a single process must never run the same task twice.
The exception is an in-place promotion: if
+ * THIS MEMBER holds the standby or warm-up task, the active task
supersedes it in a single step, so that the client
+ * recycles the task.
+ */
+ private boolean isUnreleasedActiveTask(TasksTupleWithEpochs
memberAssignedTasks,
+ String subtopologyId,
+ Integer partitionId) {
+ return currentActiveTaskProcessId.apply(subtopologyId, partitionId) !=
null
+ || heldByAnotherMemberOnThisProcess(currentStandbyTaskProcessIds,
memberAssignedTasks.standbyTasks(), subtopologyId, partitionId)
+ || heldByAnotherMemberOnThisProcess(currentWarmupTaskProcessIds,
memberAssignedTasks.warmupTasks(), subtopologyId, partitionId);
+ }
+
+ /**
+ * A standby task can only be granted once the previous holder WITHIN THE
SAME PROCESS has released it -- unlike an
+ * active task, the same standby task also exists on other processes, and
those do not block. It can also not be
+ * granted to a process that holds the task as an active or warm-up task,
because a single process must never run
+ * the same task twice. The exceptions are in-place role changes of THIS
MEMBER's own task: demoting its active task
+ * to a standby task, or converting its warm-up task into a standby task.
+ */
+ private boolean isUnreleasedStandbyTask(TasksTupleWithEpochs
memberAssignedTasks,
+ String subtopologyId,
+ Integer partitionId) {
+ return runByAnotherMemberOnThisProcess(memberAssignedTasks,
subtopologyId, partitionId)
+ || currentStandbyTaskProcessIds.apply(subtopologyId,
partitionId).contains(member.processId())
+ || heldByAnotherMemberOnThisProcess(currentWarmupTaskProcessIds,
memberAssignedTasks.warmupTasks(), subtopologyId, partitionId);
+ }
+
+ /**
+ * A warm-up task can only be granted once the previous holder WITHIN THE
SAME PROCESS has released it -- as for
+ * standby tasks, holders on other processes do not block. It can also not
be granted to a process that holds the
+ * task as an active or standby task, because a single process must never
run the same task twice. The exception is
+ * an in-place conversion of THIS MEMBER's own standby task into a warm-up
task.
+ */
+ private boolean isUnreleasedWarmupTask(TasksTupleWithEpochs
memberAssignedTasks,
+ String subtopologyId,
+ Integer partitionId) {
+ return Objects.equals(currentActiveTaskProcessId.apply(subtopologyId,
partitionId), member.processId())
+ || heldByAnotherMemberOnThisProcess(currentStandbyTaskProcessIds,
memberAssignedTasks.standbyTasks(), subtopologyId, partitionId)
+ || currentWarmupTaskProcessIds.apply(subtopologyId,
partitionId).contains(member.processId());
+ }
+
+ /**
+ * Checks whether the task is run as an active task on this member's
process by a member other than this one. The
+ * member's own active task is a candidate for an in-place role change
rather than a blocker.
+ */
+ private boolean runByAnotherMemberOnThisProcess(TasksTupleWithEpochs
memberAssignedTasks,
+ String subtopologyId,
+ Integer partitionId) {
+ return Objects.equals(currentActiveTaskProcessId.apply(subtopologyId,
partitionId), member.processId())
+ &&
!memberAssignedTasks.activeTasksWithEpochs().getOrDefault(subtopologyId,
Map.of()).containsKey(partitionId);
+ }
+
+ /**
+ * Checks whether the task is held in the given replica role on this
member's process by a member other than this
+ * one. A task the member holds itself is a candidate for an in-place role
change rather than a blocker.
*
- * @return the member's pending revocation after the promoted warm-ups
have been removed from it.
+ * @param currentProcessIds The process IDs currently holding the task in
that role.
+ * @param memberTasks The tasks this member holds in that role.
*/
- 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);
+ private boolean heldByAnotherMemberOnThisProcess(BiFunction<String,
Integer, Set<String>> currentProcessIds,
+ Map<String, Set<Integer>>
memberTasks,
+ String subtopologyId,
+ Integer partitionId) {
+ return currentProcessIds.apply(subtopologyId,
partitionId).contains(member.processId())
+ && !memberTasks.getOrDefault(subtopologyId,
Set.of()).contains(partitionId);
+ }
+
+ /**
+ * Demotes the member's active tasks to standby tasks in place. The active
task remains pending revocation -- the
+ * group must observe its release before its next owner can run it -- but
the standby task is granted right away,
+ * so the client recycles the task's state store instead of closing it and
restoring a new standby task from the
+ * changelog (cf. KAFKA-9501).
+ *
+ * @param memberAssignedTasks The tasks this member
currently holds.
+ * @param newStandbyTasksPendingAssignment Modified in place: the demoted
task is removed, as the standby is granted
+ * in this step rather than left
pending.
+ * @param newStandbyAssignedTasks Modified in place: the demoted
task is added as an assigned standby.
+ */
+ private static void demoteActiveTasksToStandby(TasksTupleWithEpochs
memberAssignedTasks,
+ Map<String, Set<Integer>>
newStandbyTasksPendingAssignment,
+ Map<String, Set<Integer>>
newStandbyAssignedTasks) {
+ for (Map.Entry<String, Map<Integer, Integer>> activeTasks :
memberAssignedTasks.activeTasksWithEpochs().entrySet()) {
+ String subtopologyId = activeTasks.getKey();
+ for (Integer partitionId : activeTasks.getValue().keySet()) {
+ if (isPendingAssignment(newStandbyTasksPendingAssignment,
subtopologyId, partitionId)) {
+ grantNow(newStandbyTasksPendingAssignment,
newStandbyAssignedTasks, subtopologyId, partitionId);
}
}
}
+ }
+
+ /**
+ * Converts one of the member's replica roles into the other in place,
i.e. a standby task into a warm-up task or
+ * vice versa. Both roles run the same code on the client, so the
conversion needs no client-side action at all
+ * and hence no revocation ack either: the outgoing role is dropped in the
same step in which the incoming role is
+ * granted. This keeps the task on the member throughout, which is what
the process-level exclusivity checks of
+ * the other members rely on.
+ *
+ * @param memberOutgoingTasks The tasks this member holds
in the role it gives up.
+ * @param newOutgoingTasksPendingRevocation Modified in place: the
converted task is removed, since the conversion
+ * needs no client-side
revocation ack.
+ * @param newIncomingTasksPendingAssignment Modified in place: the
converted task is removed, as it is granted in
+ * this step rather than left
pending.
+ * @param newIncomingAssignedTasks Modified in place: the
converted task is added, granting the new role.
+ */
+ private static void convertReplicaRole(Map<String, Set<Integer>>
memberOutgoingTasks,
+ Map<String, Set<Integer>>
newOutgoingTasksPendingRevocation,
+ Map<String, Set<Integer>>
newIncomingTasksPendingAssignment,
+ Map<String, Set<Integer>>
newIncomingAssignedTasks) {
+ for (Map.Entry<String, Set<Integer>> outgoingTasks :
memberOutgoingTasks.entrySet()) {
+ String subtopologyId = outgoingTasks.getKey();
+ for (Integer partitionId : outgoingTasks.getValue()) {
+ if (isPendingAssignment(newIncomingTasksPendingAssignment,
subtopologyId, partitionId)) {
+ grantNow(newIncomingTasksPendingAssignment,
newIncomingAssignedTasks, subtopologyId, partitionId);
+ removeFromTaskSet(newOutgoingTasksPendingRevocation,
subtopologyId, partitionId);
+ }
+ }
+ }
+ }
+
+ /**
+ * Applies promotions to active to the freshly computed reconciliation
result. A task the member holds as a standby
+ * or 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 standby or warm-up is never
routed through the (ack-based) revocation path.
+ * <p>
+ * Must be called before the caller builds the member's new assigned tasks
and pending revocation, because four of
+ * the maps below are modified in place here rather than being returned.
+ *
+ * @param memberAssignedTasks The tasks this member currently
holds; the standby and warm-up sets are
+ * the promotion candidates.
+ * @param newActiveTasksPendingRevocation Not modified; passed through
into the returned pending revocation.
+ * @param newStandbyTasksPendingRevocation Modified in place: promoted
standby tasks are removed, since a promotion
+ * needs no client-side revocation
ack.
+ * @param newWarmupTasksPendingRevocation Modified in place: as above,
for warm-up tasks.
+ * @param newActiveTasksPendingAssignment Not modified; consulted to see
whether the active task is granted in
+ * this step, which decides
whether the standby or warm-up is kept.
+ * @param newStandbyAssignedTasks Modified in place: a standby
whose promotion is not granted in this step
+ * is added back, so the member
keeps it for now.
+ * @param newWarmupAssignedTasks Modified in place: as above,
for warm-up tasks.
+ * @return the member's pending revocation after the promoted standby and
warm-up tasks have been removed from it.
+ */
+ private TasksTupleWithEpochs applyPromotionsToActive(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>> newStandbyAssignedTasks,
+ Map<String,
Set<Integer>> newWarmupAssignedTasks) {
+ // If we promote a standby or warm-up to active, that task does not
need an explicit client-side revocation.
+ // Thus, it can be removed from the revocation set -- we don't expect
a client ack back. We need to remove it
+ // regardless of whether the promotion happens now, or is still
pending on the active task revocation.
+ dropPromotedTasksFromRevocation(memberAssignedTasks.standbyTasks(),
newStandbyTasksPendingRevocation);
+ dropPromotedTasksFromRevocation(memberAssignedTasks.warmupTasks(),
newWarmupTasksPendingRevocation);
TasksTupleWithEpochs newTasksPendingRevocation = new
TasksTupleWithEpochs(
newActiveTasksPendingRevocation,
@@ -532,23 +666,82 @@ public class CurrentAssignmentBuilder {
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
+ // keep the standby or warm-up task and don't revoke it if we are not
ready for the promotion to active
// - 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();
+
+ // keep standby tasks on this member and wait on active to be
released, for later in-place promotion
+ keepPromotedTasksUntilGranted(
+ memberAssignedTasks.standbyTasks(),
+ newStandbyAssignedTasks,
+ newActiveTasksPendingAssignment,
+ hasTasksToBeRevoked
+ );
+ // keep warm-up tasks on this member and wait on active to be
released, for later in-place promotion
+ keepPromotedTasksUntilGranted(
+ memberAssignedTasks.warmupTasks(),
+ newWarmupAssignedTasks,
+ newActiveTasksPendingAssignment,
+ hasTasksToBeRevoked
+ );
+
+ return newTasksPendingRevocation;
+ }
+
+ /**
+ * Takes the tasks that are being promoted to active out of the revocation
set of the role they are promoted from.
+ */
+ private void dropPromotedTasksFromRevocation(Map<String, Set<Integer>>
memberTasks,
+ Map<String, Set<Integer>>
newTasksPendingRevocation) {
+ for (Map.Entry<String, Set<Integer>> standbyOrWarmup :
memberTasks.entrySet()) {
+ String subtopologyId = standbyOrWarmup.getKey();
+ Set<Integer> targetActiveTasks =
targetAssignment.activeTasks().getOrDefault(subtopologyId, Set.of());
+ for (Integer partitionId : standbyOrWarmup.getValue()) {
+ if (targetActiveTasks.contains(partitionId)) {
+ removeFromTaskSet(newTasksPendingRevocation,
subtopologyId, partitionId);
+ }
+ }
+ }
+ }
+
+ /**
+ * Keeps the tasks that are being promoted to active assigned in the role
they are promoted from, for as long as
+ * the active task is not granted in this step.
+ */
+ private void keepPromotedTasksUntilGranted(Map<String, Set<Integer>>
memberTasks,
+ Map<String, Set<Integer>>
newAssignedTasks,
+ Map<String, Map<Integer,
Integer>> newActiveTasksPendingAssignment,
+ boolean hasTasksToBeRevoked) {
+ for (Map.Entry<String, Set<Integer>> standbyOrWarmup :
memberTasks.entrySet()) {
+ String subtopologyId = standbyOrWarmup.getKey();
Set<Integer> targetActiveTasks =
targetAssignment.activeTasks().getOrDefault(subtopologyId, Set.of());
Map<Integer, Integer> grantedActiveTasks =
newActiveTasksPendingAssignment.getOrDefault(subtopologyId, Map.of());
- for (Integer partitionId : warmup.getValue()) {
+ for (Integer partitionId : standbyOrWarmup.getValue()) {
boolean promotedThisStep = !hasTasksToBeRevoked &&
grantedActiveTasks.containsKey(partitionId);
if (targetActiveTasks.contains(partitionId) &&
!promotedThisStep) {
- newWarmupAssignedTasks.computeIfAbsent(subtopologyId, __
-> new HashSet<>()).add(partitionId);
+ newAssignedTasks.computeIfAbsent(subtopologyId, __ -> new
HashSet<>()).add(partitionId);
}
}
}
+ }
- return newTasksPendingRevocation;
+ private static boolean isPendingAssignment(Map<String, Set<Integer>>
tasksPendingAssignment,
+ String subtopologyId,
+ Integer partitionId) {
+ return tasksPendingAssignment.getOrDefault(subtopologyId,
Set.of()).contains(partitionId);
+ }
+
+ /**
+ * Grants a task in this step instead of leaving it pending, by moving it
from the pending assignment to the
+ * assigned tasks.
+ */
+ private static void grantNow(Map<String, Set<Integer>>
tasksPendingAssignment,
+ Map<String, Set<Integer>> assignedTasks,
+ String subtopologyId,
+ Integer partitionId) {
+ removeFromTaskSet(tasksPendingAssignment, subtopologyId, partitionId);
+ assignedTasks.computeIfAbsent(subtopologyId, __ -> new
HashSet<>()).add(partitionId);
}
/**
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 3473d90f4ac..eb0a84833e9 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
@@ -1173,4 +1173,356 @@ public class CurrentAssignmentBuilderTest {
updatedMember
);
}
+
+ @Test
+ public void testStandbyPromotedToActiveWhenActiveIsReleased() {
+ final int memberEpoch = 10;
+
+ StreamsGroupMember member = new StreamsGroupMember.Builder(MEMBER_NAME)
+ .setState(MemberState.STABLE)
+ .setProcessId(PROCESS_ID)
+ .setMemberEpoch(memberEpoch)
+ .setPreviousMemberEpoch(memberEpoch)
+ .setAssignedTasks(mkTasksTupleWithCommonEpoch(TaskRole.STANDBY,
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
+ // standby is promoted in place: the active is granted and the standby
is dropped in one step, which lets the
+ // client recycle the task instead of closing it and restoring it
again as an active task.
+ StreamsGroupMember updatedMember = new CurrentAssignmentBuilder(member)
+ .withTargetAssignment(memberEpoch + 1,
mkTasksTuple(TaskRole.ACTIVE,
+ mkTasks(SUBTOPOLOGY_ID1, 0)))
+ .withCurrentActiveTaskProcessId((subtopologyId, partitionId) ->
null)
+ .withCurrentStandbyTaskProcessIds((subtopologyId, partitionId) ->
Set.of(PROCESS_ID))
+ .withCurrentWarmupTaskProcessIds((subtopologyId, partitionId) ->
Set.of())
+ .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
testStandbyPromotionHoldsStandbyWhileActiveOwnedByAnotherMember() {
+ 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.STANDBY,
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 standby (so it stays caught up) and waits in
UNRELEASED_TASKS -- it is not asked to
+ // revoke the standby.
+ StreamsGroupMember updatedMember = new CurrentAssignmentBuilder(member)
+ .withTargetAssignment(memberEpoch + 1,
mkTasksTuple(TaskRole.ACTIVE,
+ mkTasks(SUBTOPOLOGY_ID1, 0)))
+ .withCurrentActiveTaskProcessId((subtopologyId, partitionId) ->
otherProcessId)
+ .withCurrentStandbyTaskProcessIds((subtopologyId, partitionId) ->
Set.of(PROCESS_ID))
+ .withCurrentWarmupTaskProcessIds((subtopologyId, partitionId) ->
Set.of())
+ .build();
+
+ assertEquals(
+ new StreamsGroupMember.Builder(MEMBER_NAME)
+ .setState(MemberState.UNRELEASED_TASKS)
+ .setProcessId(PROCESS_ID)
+ .setMemberEpoch(memberEpoch + 1)
+ .setPreviousMemberEpoch(memberEpoch)
+
.setAssignedTasks(mkTasksTupleWithCommonEpoch(TaskRole.STANDBY, memberEpoch,
+ mkTasks(SUBTOPOLOGY_ID1, 0)))
+ .setTasksPendingRevocation(TasksTupleWithEpochs.EMPTY)
+ .build(),
+ updatedMember
+ );
+ }
+
+ @Test
+ public void
testActiveNotGrantedWhileAnotherMemberOnSameProcessHoldsStandby() {
+ final int memberEpoch = 10;
+
+ // This member does not hold the standby itself; a *different* member
on the same process does (reflected by
+ // currentStandbyTaskProcessIds returning this member's process). The
active must not be granted -- the process
+ // would otherwise run the task as both active (this member) and
standby (the sibling).
+ 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(PROCESS_ID))
+ .withCurrentWarmupTaskProcessIds((subtopologyId, partitionId) ->
Set.of())
+ .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 testActiveDemotedToStandbyInPlace() {
+ final int memberEpoch = 10;
+
+ StreamsGroupMember member = 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();
+
+ // The target wants this member to keep the task as a standby. The
standby is granted in the same step in
+ // which the active task is revoked, so the client recycles the task.
The active task still needs an
+ // acknowledgement, because the group must observe its release before
its next owner can run it.
+ StreamsGroupMember updatedMember = new CurrentAssignmentBuilder(member)
+ .withTargetAssignment(memberEpoch + 1,
mkTasksTuple(TaskRole.STANDBY,
+ mkTasks(SUBTOPOLOGY_ID1, 0)))
+ .withCurrentActiveTaskProcessId((subtopologyId, partitionId) ->
PROCESS_ID)
+ .withCurrentStandbyTaskProcessIds((subtopologyId, partitionId) ->
Set.of())
+ .withCurrentWarmupTaskProcessIds((subtopologyId, partitionId) ->
Set.of())
+ .build();
+
+ assertEquals(
+ new StreamsGroupMember.Builder(MEMBER_NAME)
+ .setState(MemberState.UNREVOKED_TASKS)
+ .setProcessId(PROCESS_ID)
+ .setMemberEpoch(memberEpoch)
+ .setPreviousMemberEpoch(memberEpoch)
+
.setAssignedTasks(mkTasksTupleWithCommonEpoch(TaskRole.STANDBY, memberEpoch,
+ mkTasks(SUBTOPOLOGY_ID1, 0)))
+
.setTasksPendingRevocation(mkTasksTupleWithCommonEpoch(TaskRole.ACTIVE,
memberEpoch,
+ mkTasks(SUBTOPOLOGY_ID1, 0)))
+ .build(),
+ updatedMember
+ );
+ }
+
+ @Test
+ public void testActiveDemotionCompletesWhenRevocationIsAcknowledged() {
+ final int memberEpoch = 10;
+
+ // The member was told to demote its active task to a standby task in
the previous step: it already holds the
+ // standby, and the active task is pending revocation.
+ StreamsGroupMember member = new StreamsGroupMember.Builder(MEMBER_NAME)
+ .setState(MemberState.UNREVOKED_TASKS)
+ .setProcessId(PROCESS_ID)
+ .setMemberEpoch(memberEpoch)
+ .setPreviousMemberEpoch(memberEpoch)
+ .setAssignedTasks(mkTasksTupleWithCommonEpoch(TaskRole.STANDBY,
memberEpoch,
+ mkTasks(SUBTOPOLOGY_ID1, 0)))
+
.setTasksPendingRevocation(mkTasksTupleWithCommonEpoch(TaskRole.ACTIVE,
memberEpoch,
+ mkTasks(SUBTOPOLOGY_ID1, 0)))
+ .build();
+
+ // The member reports the task as a standby only, which acknowledges
the release of the active task.
+ StreamsGroupMember updatedMember = new CurrentAssignmentBuilder(member)
+ .withTargetAssignment(memberEpoch + 1,
mkTasksTuple(TaskRole.STANDBY,
+ mkTasks(SUBTOPOLOGY_ID1, 0)))
+ .withCurrentActiveTaskProcessId((subtopologyId, partitionId) ->
PROCESS_ID)
+ .withCurrentStandbyTaskProcessIds((subtopologyId, partitionId) ->
Set.of(PROCESS_ID))
+ .withCurrentWarmupTaskProcessIds((subtopologyId, partitionId) ->
Set.of())
+ .withOwnedAssignment(mkTasksTuple(TaskRole.STANDBY,
mkTasks(SUBTOPOLOGY_ID1, 0)))
+ .build();
+
+ assertEquals(
+ new StreamsGroupMember.Builder(MEMBER_NAME)
+ .setState(MemberState.STABLE)
+ .setProcessId(PROCESS_ID)
+ .setMemberEpoch(memberEpoch + 1)
+ .setPreviousMemberEpoch(memberEpoch)
+
.setAssignedTasks(mkTasksTupleWithCommonEpoch(TaskRole.STANDBY, memberEpoch,
+ mkTasks(SUBTOPOLOGY_ID1, 0)))
+ .setTasksPendingRevocation(TasksTupleWithEpochs.EMPTY)
+ .build(),
+ updatedMember
+ );
+ }
+
+ @Test
+ public void
testActiveNotDemotedInPlaceWhileAnotherMemberOnSameProcessHoldsStandby() {
+ final int memberEpoch = 10;
+
+ // A *different* member on the same process already holds the task as
a standby, so this member cannot take it
+ // on as a standby as well. The demotion falls back to a plain
revocation, and the standby is granted only
+ // once the sibling has given it up.
+ StreamsGroupMember member = 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();
+
+ StreamsGroupMember updatedMember = new CurrentAssignmentBuilder(member)
+ .withTargetAssignment(memberEpoch + 1,
mkTasksTuple(TaskRole.STANDBY,
+ mkTasks(SUBTOPOLOGY_ID1, 0)))
+ .withCurrentActiveTaskProcessId((subtopologyId, partitionId) ->
PROCESS_ID)
+ .withCurrentStandbyTaskProcessIds((subtopologyId, partitionId) ->
Set.of(PROCESS_ID))
+ .withCurrentWarmupTaskProcessIds((subtopologyId, partitionId) ->
Set.of())
+ .build();
+
+ assertEquals(
+ new StreamsGroupMember.Builder(MEMBER_NAME)
+ .setState(MemberState.UNREVOKED_TASKS)
+ .setProcessId(PROCESS_ID)
+ .setMemberEpoch(memberEpoch)
+ .setPreviousMemberEpoch(memberEpoch)
+ .setAssignedTasks(TasksTupleWithEpochs.EMPTY)
+
.setTasksPendingRevocation(mkTasksTupleWithCommonEpoch(TaskRole.ACTIVE,
memberEpoch,
+ mkTasks(SUBTOPOLOGY_ID1, 0)))
+ .build(),
+ updatedMember
+ );
+ }
+
+ @Test
+ public void testStandbyConvertedToWarmupInPlace() {
+ final int memberEpoch = 10;
+
+ StreamsGroupMember member = new StreamsGroupMember.Builder(MEMBER_NAME)
+ .setState(MemberState.STABLE)
+ .setProcessId(PROCESS_ID)
+ .setMemberEpoch(memberEpoch)
+ .setPreviousMemberEpoch(memberEpoch)
+ .setAssignedTasks(mkTasksTupleWithCommonEpoch(TaskRole.STANDBY,
memberEpoch,
+ mkTasks(SUBTOPOLOGY_ID1, 0)))
+ .setTasksPendingRevocation(TasksTupleWithEpochs.EMPTY)
+ .build();
+
+ // Standby and warm-up tasks run the same code on the client, so
relabelling one into the other needs no
+ // client-side action and hence no revocation acknowledgement: the
standby is dropped in the same step in
+ // which the warm-up is granted, and the task keeps running throughout.
+ StreamsGroupMember updatedMember = new CurrentAssignmentBuilder(member)
+ .withTargetAssignment(memberEpoch + 1,
mkTasksTuple(TaskRole.WARMUP,
+ mkTasks(SUBTOPOLOGY_ID1, 0)))
+ .withCurrentActiveTaskProcessId((subtopologyId, partitionId) ->
null)
+ .withCurrentStandbyTaskProcessIds((subtopologyId, partitionId) ->
Set.of(PROCESS_ID))
+ .withCurrentWarmupTaskProcessIds((subtopologyId, partitionId) ->
Set.of())
+ .build();
+
+ assertEquals(
+ new StreamsGroupMember.Builder(MEMBER_NAME)
+ .setState(MemberState.STABLE)
+ .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 testWarmupConvertedToStandbyInPlace() {
+ 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 reverse of the conversion above, which is what happens when a
planned migration is called off and the
+ // warm-up becomes a regular standby.
+ StreamsGroupMember updatedMember = new CurrentAssignmentBuilder(member)
+ .withTargetAssignment(memberEpoch + 1,
mkTasksTuple(TaskRole.STANDBY,
+ 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.STANDBY, memberEpoch,
+ mkTasks(SUBTOPOLOGY_ID1, 0)))
+ .setTasksPendingRevocation(TasksTupleWithEpochs.EMPTY)
+ .build(),
+ updatedMember
+ );
+ }
+
+ @Test
+ public void
testStandbyRevokedNormallyWhenConversionIsBlockedByAnotherMemberOnSameProcess()
{
+ final int memberEpoch = 10;
+
+ // A *different* member on the same process already holds the task as
a warm-up, so this member cannot convert
+ // its standby into a warm-up. The standby must then be revoked the
regular way (with an acknowledgement),
+ // because the process may not hold the task twice.
+ StreamsGroupMember member = new StreamsGroupMember.Builder(MEMBER_NAME)
+ .setState(MemberState.STABLE)
+ .setProcessId(PROCESS_ID)
+ .setMemberEpoch(memberEpoch)
+ .setPreviousMemberEpoch(memberEpoch)
+ .setAssignedTasks(mkTasksTupleWithCommonEpoch(TaskRole.STANDBY,
memberEpoch,
+ mkTasks(SUBTOPOLOGY_ID1, 0)))
+ .setTasksPendingRevocation(TasksTupleWithEpochs.EMPTY)
+ .build();
+
+ StreamsGroupMember updatedMember = new CurrentAssignmentBuilder(member)
+ .withTargetAssignment(memberEpoch + 1,
mkTasksTuple(TaskRole.WARMUP,
+ mkTasks(SUBTOPOLOGY_ID1, 0)))
+ .withCurrentActiveTaskProcessId((subtopologyId, partitionId) ->
null)
+ .withCurrentStandbyTaskProcessIds((subtopologyId, partitionId) ->
Set.of(PROCESS_ID))
+ .withCurrentWarmupTaskProcessIds((subtopologyId, partitionId) ->
Set.of(PROCESS_ID))
+ .build();
+
+ assertEquals(
+ new StreamsGroupMember.Builder(MEMBER_NAME)
+ .setState(MemberState.UNREVOKED_TASKS)
+ .setProcessId(PROCESS_ID)
+ .setMemberEpoch(memberEpoch)
+ .setPreviousMemberEpoch(memberEpoch)
+ .setAssignedTasks(TasksTupleWithEpochs.EMPTY)
+
.setTasksPendingRevocation(mkTasksTupleWithCommonEpoch(TaskRole.STANDBY,
memberEpoch,
+ mkTasks(SUBTOPOLOGY_ID1, 0)))
+ .build(),
+ updatedMember
+ );
+ }
}
diff --git
a/streams/integration-tests/src/test/java/org/apache/kafka/streams/integration/RestoreIntegrationTest.java
b/streams/integration-tests/src/test/java/org/apache/kafka/streams/integration/RestoreIntegrationTest.java
index 29d3cf367a3..8e7068de1d7 100644
---
a/streams/integration-tests/src/test/java/org/apache/kafka/streams/integration/RestoreIntegrationTest.java
+++
b/streams/integration-tests/src/test/java/org/apache/kafka/streams/integration/RestoreIntegrationTest.java
@@ -621,10 +621,11 @@ public class RestoreIntegrationTest {
assertThat(restoreListener.totalNumRestored(),
CoreMatchers.equalTo(initialNunRestoredCount));
- // After stopping instance 2 and letting instance 1 take over its
tasks, we should have closed the stores on instance 2.
- // Under the new group protocol, an extra store close can occur
during rebalance; account for that here.
- final int expectedAfterStreams2Close = initialStoreCloseCount +
(useNewProtocol ? 3 : 2);
- assertThat(CloseCountingInMemoryStore.numStoresClosed(),
equalTo(expectedAfterStreams2Close));
+ // After stopping instance 2 and letting instance 1 take over its
tasks, we should have closed just two stores
+ // total: the active and standby tasks on instance 2. The new
protocol used to close one store more, because
+ // the standby that instance 1 already held was closed and
re-created instead of being promoted in place;
+ // now that the reconciler changes the role in place, both
protocols close the same two stores.
+ assertThat(CloseCountingInMemoryStore.numStoresClosed(),
equalTo(initialStoreCloseCount + 2));
} finally {
streams1.close(Duration.ofSeconds(60));
}
