Skip to content

KAFKA-13600: Kafka Streams - Fall back to most caught up client if no caught up clients exist #11760

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
cadonna merged 10 commits into from
Mar 28, 2022
Merged

KAFKA-13600: Kafka Streams - Fall back to most caught up client if no caught up clients exist #11760

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
10 commits
Select commit Hold shift + click to select a range
382ae5a
[KAFKA-13600] Unit test to capture not assigning to caught up task if…
tim-patterson Feb 14, 2022
6cec06c
[KAFKA-13600] Fall back to most caught up client if a caught up clien…
tim-patterson Feb 15, 2022
5284d29
[KAFKA-13600] Fix copy paste mistake
tim-patterson Feb 16, 2022
7b1e879
[KAFKA-13600] Make all cases for active assignment more explicit
tim-patterson Feb 18, 2022
26149ab
[KAFKA-13600] Use SortedSet for clients in tasksToClientByLag
tim-patterson Feb 28, 2022
67d5ba0
[KAFKA-13600] Merge remote-tracking branch 'origin/trunk' into KAFKA-…
tim-patterson Mar 11, 2022
ec7e24c
[KAFKA-13600] Fix merge issues
tim-patterson Mar 11, 2022
2dfb79d
[KAFKA-13600] break up large method
tim-patterson Mar 14, 2022
067897e
Update streams/src/main/java/org/apache/kafka/streams/processor/inter…
tim-patterson Mar 25, 2022
d3798f5
[KAFKA-13600] Add taskmovement test for moving active to most caught up
tim-patterson Mar 25, 2022
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
17 changes: 17 additions & 0 deletions ...org/apache/kafka/streams/processor/internals/assignment/HighAvailabilityTaskAssignor.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -22,6 +22,7 @@
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.Comparator;
import java.util.HashMap;
import java.util.Iterator;
import java.util.Map;
Expand Down Expand Up @@ -68,6 +69,8 @@ public boolean assign(final Map<UUID, ClientState> clients,
configs.acceptableRecoveryLag
);

final Map<TaskId, SortedSet<UUID>> tasksToClientByLag = tasksToClientByLag(statefulTasks, clientStates);

// We temporarily need to know which standby tasks were intended as warmups
// for active tasks, so that we don't move them (again) when we plan standby
// task movements. We can then immediately treat warmups exactly the same as
Expand All @@ -77,13 +80,15 @@ public boolean assign(final Map<UUID, ClientState> clients,

final int neededActiveTaskMovements = assignActiveTaskMovements(
tasksToCaughtUpClients,
tasksToClientByLag,
clientStates,
warmups,
remainingWarmupReplicas
);

final int neededStandbyTaskMovements = assignStandbyTaskMovements(
tasksToCaughtUpClients,
tasksToClientByLag,
clientStates,
remainingWarmupReplicas,
warmups
Expand Down Expand Up @@ -247,6 +252,18 @@ private static Map<TaskId, SortedSet<UUID>> tasksToCaughtUpClients(final Set<Tas
return taskToCaughtUpClients;
}

private static Map<TaskId, SortedSet<UUID>> tasksToClientByLag(final Set<TaskId> statefulTasks,
final Map<UUID, ClientState> clientStates) {
final Map<TaskId, SortedSet<UUID>> tasksToClientByLag = new HashMap<>();
for (final TaskId task : statefulTasks) {
final SortedSet<UUID> clientLag = new TreeSet<>(Comparator.<UUID>comparingLong(a ->
clientStates.get(a).lagFor(task)).thenComparing(a -> a));
clientLag.addAll(clientStates.keySet());
tasksToClientByLag.put(task, clientLag);
}
return tasksToClientByLag;
}

private static boolean unbounded(final long acceptableRecoveryLag) {
return acceptableRecoveryLag == Long.MAX_VALUE;
}
Expand Down
178 changes: 130 additions & 48 deletions ...s/src/main/java/org/apache/kafka/streams/processor/internals/assignment/TaskMovement.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -29,6 +29,7 @@
import java.util.UUID;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.function.BiFunction;
import java.util.function.Function;

import static java.util.Arrays.asList;
import static java.util.Objects.requireNonNull;
Expand All @@ -42,10 +43,6 @@ private TaskMovement(final TaskId task, final UUID destination, final SortedSet<
this.task = task;
this.destination = destination;
this.caughtUpClients = caughtUpClients;

if (caughtUpClients == null || caughtUpClients.isEmpty()) {
throw new IllegalStateException("Should not attempt to move a task if no caught up clients exist");
}
}

private TaskId task() {
Expand All @@ -56,25 +53,34 @@ private int numCaughtUpClients() {
return caughtUpClients.size();
}

private static boolean taskIsNotCaughtUpOnClientAndOtherCaughtUpClientsExist(final TaskId task,
final UUID client,
final Map<TaskId, SortedSet<UUID>> tasksToCaughtUpClients) {
return !taskIsCaughtUpOnClientOrNoCaughtUpClientsExist(task, client, tasksToCaughtUpClients);
private static boolean taskIsNotCaughtUpOnClientAndOtherMoreCaughtUpClientsExist(final TaskId task,
final UUID client,
final Map<UUID, ClientState> clientStates,
final Map<TaskId, SortedSet<UUID>> tasksToCaughtUpClients,
final Map<TaskId, SortedSet<UUID>> tasksToClientByLag) {
final SortedSet<UUID> taskClients = requireNonNull(tasksToClientByLag.get(task), "uninitialized set");
if (taskIsCaughtUpOnClient(task, client, tasksToCaughtUpClients)) {
return false;
}
final long mostCaughtUpLag = clientStates.get(taskClients.first()).lagFor(task);
final long clientLag = clientStates.get(client).lagFor(task);
return mostCaughtUpLag < clientLag;
}

private static boolean taskIsCaughtUpOnClientOrNoCaughtUpClientsExist(final TaskId task,
final UUID client,
final Map<TaskId, SortedSet<UUID>> tasksToCaughtUpClients) {
private static boolean taskIsCaughtUpOnClient(final TaskId task,
final UUID client,
final Map<TaskId, SortedSet<UUID>> tasksToCaughtUpClients) {
final Set<UUID> caughtUpClients = requireNonNull(tasksToCaughtUpClients.get(task), "uninitialized set");
return caughtUpClients.isEmpty() || caughtUpClients.contains(client);
return caughtUpClients.contains(client);
}

static int assignActiveTaskMovements(final Map<TaskId, SortedSet<UUID>> tasksToCaughtUpClients,
final Map<TaskId, SortedSet<UUID>> tasksToClientByLag,
final Map<UUID, ClientState> clientStates,
final Map<UUID, Set<TaskId>> warmups,
final AtomicInteger remainingWarmupReplicas) {
final BiFunction<UUID, TaskId, Boolean> caughtUpPredicate =
(client, task) -> taskIsCaughtUpOnClientOrNoCaughtUpClientsExist(task, client, tasksToCaughtUpClients);
(client, task) -> taskIsCaughtUpOnClient(task, client, tasksToCaughtUpClients);

final ConstrainedPrioritySet caughtUpClientsByTaskLoad = new ConstrainedPrioritySet(
caughtUpPredicate,
Expand All @@ -89,10 +95,10 @@ static int assignActiveTaskMovements(final Map<TaskId, SortedSet<UUID>> tasksToC
final UUID client = clientStateEntry.getKey();
final ClientState state = clientStateEntry.getValue();
for (final TaskId task : state.activeTasks()) {
// if the desired client is not caught up, and there is another client that _is_ caught up, then
// we schedule a movement, so we can move the active task to the caught-up client. We'll try to
// if the desired client is not caught up, and there is another client that _is_ more caught up, then
// we schedule a movement, so we can move the active task to a more caught-up client. We'll try to
// assign a warm-up to the desired client so that we can move it later on.
if (taskIsNotCaughtUpOnClientAndOtherCaughtUpClientsExist(task, client, tasksToCaughtUpClients)) {
if (taskIsNotCaughtUpOnClientAndOtherMoreCaughtUpClientsExist(task, client, clientStates, tasksToCaughtUpClients, tasksToClientByLag)) {
taskMovements.add(new TaskMovement(task, client, tasksToCaughtUpClients.get(task)));
}
}
Expand All @@ -102,45 +108,27 @@ static int assignActiveTaskMovements(final Map<TaskId, SortedSet<UUID>> tasksToC
final int movementsNeeded = taskMovements.size();

for (final TaskMovement movement : taskMovements) {
final UUID standbySourceClient = caughtUpClientsByTaskLoad.poll(
movement.task,
c -> clientStates.get(c).hasStandbyTask(movement.task)
);
if (standbySourceClient == null) {
// there's not a caught-up standby available to take over the task, so we'll schedule a warmup instead
final UUID sourceClient = requireNonNull(
caughtUpClientsByTaskLoad.poll(movement.task),
"Tried to move task to caught-up client but none exist"
);

moveActiveAndTryToWarmUp(
remainingWarmupReplicas,
movement.task,
clientStates.get(sourceClient),
clientStates.get(movement.destination),
warmups.computeIfAbsent(movement.destination, x -> new TreeSet<>())
);
caughtUpClientsByTaskLoad.offerAll(asList(sourceClient, movement.destination));
} else {
// we found a candidate to trade standby/active state with our destination, so we don't need a warmup
swapStandbyAndActive(
movement.task,
clientStates.get(standbySourceClient),
clientStates.get(movement.destination)
);
caughtUpClientsByTaskLoad.offerAll(asList(standbySourceClient, movement.destination));
// Attempt to find a caught up standby, otherwise find any caught up client, failing that use the most
// caught up client.
final boolean moved = tryToSwapStandbyAndActiveOnCaughtUpClient(clientStates, caughtUpClientsByTaskLoad, movement) ||
tryToMoveActiveToCaughtUpClientAndTryToWarmUp(clientStates, warmups, remainingWarmupReplicas, caughtUpClientsByTaskLoad, movement) ||
tryToMoveActiveToMostCaughtUpClient(tasksToClientByLag, clientStates, warmups, remainingWarmupReplicas, caughtUpClientsByTaskLoad, movement);

@cadonna cadonna Mar 23, 2022

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Could you add tests to TaskMovementTest for case tryToMoveActiveToMostCaughtUpClient() that you added?


if (!moved) {
throw new IllegalStateException("Tried to move task to more caught-up client as scheduled before but none exist");
}
}

return movementsNeeded;
}

static int assignStandbyTaskMovements(final Map<TaskId, SortedSet<UUID>> tasksToCaughtUpClients,
final Map<TaskId, SortedSet<UUID>> tasksToClientByLag,
final Map<UUID, ClientState> clientStates,
final AtomicInteger remainingWarmupReplicas,
final Map<UUID, Set<TaskId>> warmups) {
final BiFunction<UUID, TaskId, Boolean> caughtUpPredicate =
(client, task) -> taskIsCaughtUpOnClientOrNoCaughtUpClientsExist(task, client, tasksToCaughtUpClients);
(client, task) -> taskIsCaughtUpOnClient(task, client, tasksToCaughtUpClients);

final ConstrainedPrioritySet caughtUpClientsByTaskLoad = new ConstrainedPrioritySet(
caughtUpPredicate,
Expand All @@ -157,8 +145,8 @@ static int assignStandbyTaskMovements(final Map<TaskId, SortedSet<UUID>> tasksTo
for (final TaskId task : state.standbyTasks()) {
if (warmups.getOrDefault(destination, Collections.emptySet()).contains(task)) {
// this is a warmup, so we won't move it.
} else if (taskIsNotCaughtUpOnClientAndOtherCaughtUpClientsExist(task, destination, tasksToCaughtUpClients)) {
// if the desired client is not caught up, and there is another client that _is_ caught up, then
} else if (taskIsNotCaughtUpOnClientAndOtherMoreCaughtUpClientsExist(task, destination, clientStates, tasksToCaughtUpClients, tasksToClientByLag)) {
// if the desired client is not caught up, and there is another client that _is_ more caught up, then
// we schedule a movement, so we can move the active task to the caught-up client. We'll try to
// assign a warm-up to the desired client so that we can move it later on.
taskMovements.add(new TaskMovement(task, destination, tasksToCaughtUpClients.get(task)));
Expand All @@ -170,11 +158,17 @@ static int assignStandbyTaskMovements(final Map<TaskId, SortedSet<UUID>> tasksTo
int movementsNeeded = 0;

for (final TaskMovement movement : taskMovements) {
final UUID sourceClient = caughtUpClientsByTaskLoad.poll(
final Function<UUID, Boolean> eligibleClientPredicate =
clientId -> !clientStates.get(clientId).hasAssignedTask(movement.task);
UUID sourceClient = caughtUpClientsByTaskLoad.poll(
movement.task,
clientId -> !clientStates.get(clientId).hasAssignedTask(movement.task)
eligibleClientPredicate
);

if (sourceClient == null) {
sourceClient = mostCaughtUpEligibleClient(tasksToClientByLag, eligibleClientPredicate, movement.task, movement.destination);
}

if (sourceClient == null) {
// then there's no caught-up client that doesn't already have a copy of this task, so there's
// nowhere to move it.
Expand All @@ -193,6 +187,74 @@ static int assignStandbyTaskMovements(final Map<TaskId, SortedSet<UUID>> tasksTo
return movementsNeeded;
}

private static boolean tryToSwapStandbyAndActiveOnCaughtUpClient(final Map<UUID, ClientState> clientStates,
final ConstrainedPrioritySet caughtUpClientsByTaskLoad,
final TaskMovement movement) {
final UUID caughtUpStandbySourceClient = caughtUpClientsByTaskLoad.poll(
movement.task,
c -> clientStates.get(c).hasStandbyTask(movement.task)
);
if (caughtUpStandbySourceClient != null) {
swapStandbyAndActive(
movement.task,
clientStates.get(caughtUpStandbySourceClient),
clientStates.get(movement.destination)
);
caughtUpClientsByTaskLoad.offerAll(asList(caughtUpStandbySourceClient, movement.destination));
return true;
}
return false;
}

private static boolean tryToMoveActiveToCaughtUpClientAndTryToWarmUp(final Map<UUID, ClientState> clientStates,
final Map<UUID, Set<TaskId>> warmups,
final AtomicInteger remainingWarmupReplicas,
final ConstrainedPrioritySet caughtUpClientsByTaskLoad,
final TaskMovement movement) {
final UUID caughtUpSourceClient = caughtUpClientsByTaskLoad.poll(movement.task);
if (caughtUpSourceClient != null) {
moveActiveAndTryToWarmUp(
remainingWarmupReplicas,
movement.task,
clientStates.get(caughtUpSourceClient),
clientStates.get(movement.destination),
warmups.computeIfAbsent(movement.destination, x -> new TreeSet<>())
);
caughtUpClientsByTaskLoad.offerAll(asList(caughtUpSourceClient, movement.destination));
return true;
}
return false;
}

private static boolean tryToMoveActiveToMostCaughtUpClient(final Map<TaskId, SortedSet<UUID>> tasksToClientByLag,
final Map<UUID, ClientState> clientStates,
final Map<UUID, Set<TaskId>> warmups,
final AtomicInteger remainingWarmupReplicas,
final ConstrainedPrioritySet caughtUpClientsByTaskLoad,
final TaskMovement movement) {
final UUID mostCaughtUpSourceClient = mostCaughtUpEligibleClient(tasksToClientByLag, movement.task, movement.destination);
if (mostCaughtUpSourceClient != null) {
if (clientStates.get(mostCaughtUpSourceClient).hasStandbyTask(movement.task)) {
swapStandbyAndActive(
movement.task,
clientStates.get(mostCaughtUpSourceClient),
clientStates.get(movement.destination)
);
} else {
moveActiveAndTryToWarmUp(
remainingWarmupReplicas,
movement.task,
clientStates.get(mostCaughtUpSourceClient),
clientStates.get(movement.destination),
warmups.computeIfAbsent(movement.destination, x -> new TreeSet<>())
);
}
caughtUpClientsByTaskLoad.offerAll(asList(mostCaughtUpSourceClient, movement.destination));
return true;
}
return false;
}

private static void moveActiveAndTryToWarmUp(final AtomicInteger remainingWarmupReplicas,
final TaskId task,
final ClientState sourceClientState,
Expand Down Expand Up @@ -235,4 +297,24 @@ private static void swapStandbyAndActive(final TaskId task,
destinationClientState.assignStandby(task);
}

private static UUID mostCaughtUpEligibleClient(final Map<TaskId, SortedSet<UUID>> tasksToClientByLag,
final TaskId task,
final UUID destinationClient) {
return mostCaughtUpEligibleClient(tasksToClientByLag, client -> true, task, destinationClient);
}

private static UUID mostCaughtUpEligibleClient(final Map<TaskId, SortedSet<UUID>> tasksToClientByLag,
final Function<UUID, Boolean> constraint,
final TaskId task,
final UUID destinationClient) {
for (final UUID client : tasksToClientByLag.get(task)) {
if (destinationClient.equals(client)) {
break;
} else if (constraint.apply(client)) {
return client;
}
}
return null;
}

}
Loading