KAFKA-6145: KIP-441: Enforce Standby Task Stickiness by vvcephei · Pull Request #8696 · apache/kafka

vvcephei · 2020-05-19T21:27:00Z

We should treat standbys similarly to active stateful tasks and
re-assign them to instances that are already caught-up on them
while we warm them up on the desired destination, instead of
immediately moving them to the destination.

Committer Checklist (excluded from commit message)

Verify design and implementation
Verify test coverage and CI build status
Verify documentation (including upgrade notes)

We should treat standbys similarly to active stateful tasks and re-assign them to instances that are already caught-up on them while we warm them up on the desired destination, instead of immediately moving them to the destination.

vvcephei

Hey @ableegoldman and @cadonna , what do you think about this?

vvcephei · 2020-05-19T21:28:08Z


-        final boolean probingRebalanceNeeded = assignTaskMovements(
-            tasksToCaughtUpClients(statefulTasks, clientStates, configs.acceptableRecoveryLag),
+        final AtomicInteger remainingWarmupReplicas = new AtomicInteger(configs.maxWarmupReplicas);


Moved the counter out here because we need to decrement it while assigning both active and standby warmups

vvcephei · 2020-05-19T21:29:45Z

+        final int neededStandbyTaskMovements = assignStandbyTaskMovements(
+            tasksToCaughtUpClients,
+            clientStates,
+            remainingWarmupReplicas,
+            warmups


The mechanism by which we enforce "stickiness" is by assigning movements after computing the ideal assignment, so if we want standbys as well as actives to be sticky, we need to assign movements for those as well.

vvcephei · 2020-05-19T21:32:38Z

        assignStatelessActiveTasks(clientStates, diff(TreeSet::new, allTaskIds, statefulTasks));

+        // We shouldn't plan a probing rebalance if we _needed_ task movements, but couldn't do any
+        // due to being configured for no warmups.


One might wonder whether we should even allow "max_warmups := 0". I think this is actually ok, as someone might want to completely disable this state shuffling mechanism and instead just be 100% sticky. Also factoring into my thinking is that it's pretty obvious what will happen if you configure "no warmups", so I don't think it's going to hurt someone who didn't actually want to completely disable the warmup mechanism.

To be clear, according to StreamsConfig, we do NOT allow max.warmup.replicas = 0. It must at least be 1. Or was your statement hypothetical, that it would be OK to allow it? Anyway, I am in favour of keeping the > 0 check here.

Yeah, I've just realized this, too. And upon second consideration, I don't think the warmup=0 really provides a good mechanism for what I was thinking of. I think we'd better leave it as "at least one". Thanks!

vvcephei · 2020-05-19T21:34:04Z

-    static boolean assignTaskMovements(final Map<TaskId, SortedSet<UUID>> tasksToCaughtUpClients,
-                                       final Map<UUID, ClientState> clientStates,
-                                       final int maxWarmupReplicas) {
+    static int assignActiveTaskMovements(final Map<TaskId, SortedSet<UUID>> tasksToCaughtUpClients,


I changed this to return an int just because it made stepping through the assignment in the debugger a bit easier to understand. It serves no algorithmic purpose.

vvcephei · 2020-05-19T21:35:28Z

        return movementsNeeded;
    }

+    static int assignStandbyTaskMovements(final Map<TaskId, SortedSet<UUID>> tasksToCaughtUpClients,


This algorithm is similar to the active one, but there are also important differences, so I didn't converge them.

vvcephei · 2020-05-19T21:36:47Z

+    static Matcher<ClientState> hasAssignedTasks(final int taskCount) {
+        return hasProperty("assignedTasks", ClientState::assignedTaskCount, taskCount);
+    }


Similar to the other matchers, it just gives us mildly nicer test output.

vvcephei · 2020-05-19T21:37:56Z

    );

+    @Test
+    public void shouldBeStickyForActiveAndStandbyTasksEvenIfNoWarmups() {


First test for stickiness: we should be 100% sticky and also not schedule a probing rebalance when we are configured for no warmups.

vvcephei · 2020-05-19T21:38:43Z

+    }
+
+    @Test
+    public void shouldBeStickyForActiveAndStandbyTasksWhileWarmingUp() {


Main test case for stickiness: we should be sticky for standbys, and also schedule warmups.

vvcephei · 2020-05-19T21:39:08Z

                }
                if (!harness.clientStates.isEmpty()) {
-                    testForConvergence(harness, configs, numStatefulTasks * 2);
+                    testForConvergence(harness, configs, 2 * (numStatefulTasks + numStatefulTasks * numStandbyReplicas));


Now that we're warming up standbys also, we need to relax the convergence limit.

vvcephei · 2020-05-19T21:39:43Z

+                new TreeMap<>(),
+                new AtomicInteger(maxWarmupReplicas)
+            ),
+            is(0)


Just accommodating the new method signature, no semantic changes.

cadonna

Hey @vvcephei, Thank you for the PR.

Here my feeddback.

cadonna · 2020-05-22T11:37:00Z

+            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 (!taskIsCaughtUpOnClientOrNoCaughtUpClientsExist(task, destination, tasksToCaughtUpClients)) {


prop: Could you add a method taskIsNotCaughtUpOnClientAndCaughtUpClientsExist()? Applying De Morgan's law every time I read this code gives me headache.

Haha, sure :)

cadonna · 2020-05-22T11:54:00Z

        assignStatelessActiveTasks(clientStates, diff(TreeSet::new, allTaskIds, statefulTasks));

+        // We shouldn't plan a probing rebalance if we _needed_ task movements, but couldn't do any
+        // due to being configured for no warmups.


To be clear, according to StreamsConfig, we do NOT allow max.warmup.replicas = 0. It must at least be 1. Or was your statement hypothetical, that it would be OK to allow it? Anyway, I am in favour of keeping the > 0 check here.

vvcephei · 2020-05-22T16:39:43Z

Hey @cadonna , thanks for the feedback! I think I'll go ahead and merge this because I've already fixed the "max warmup" thing in a separate bugfix PR, which builds on this one.

I'll apply your DeMorgan's law suggestion also in the follow-up.

* 'trunk' of github.com:apache/kafka: KAFKA-9888: Copy connector configs before passing to REST extensions (apache#8511) KAFKA-9931: Implement KIP-605 to expand support for Connect worker internal topic configurations (apache#8654) KAFKA-6145: Add unit tests for assignments of only stateless tasks (apache#8713) MINOR: Fix join group request timeout lower bound (apache#8702) MINOR: Improve security documentation for Kafka Streams apache#8710 KAFKA-6145: KIP-441: Enforce Standby Task Stickiness (apache#8696) KAFKA-10003: Mark KStream.through() as deprecated and update Scala API (apache#8679)

KAFKA-6145: KIP-441: Enforce Standby Task Stickiness

d85c12b

We should treat standbys similarly to active stateful tasks and re-assign them to instances that are already caught-up on them while we warm them up on the desired destination, instead of immediately moving them to the destination.

vvcephei added the streams label May 19, 2020

vvcephei commented May 19, 2020

View reviewed changes

cadonna reviewed May 22, 2020

View reviewed changes

cadonna approved these changes May 22, 2020

View reviewed changes

vvcephei merged commit 9fdd877 into apache:trunk May 22, 2020

vvcephei deleted the kafka-6145-standby-stickiness branch May 22, 2020 16:41

vvcephei mentioned this pull request May 22, 2020

KAFKA-6145: KIP-441: Fix assignor config passthough #8716

Merged

3 tasks

mjsax added the kip Requires or implements a KIP label Jun 12, 2020

Conversation

vvcephei commented May 19, 2020

Committer Checklist (excluded from commit message)

Uh oh!

vvcephei left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

cadonna left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

vvcephei commented May 22, 2020

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

3 participants