lucasbru commented on code in PR #23675:
URL: https://github.com/apache/kafka/pull/23675#discussion_r4228882482
##########
streams/integration-tests/src/test/java/org/apache/kafka/streams/integration/StreamsGroupWarmupTaskLifecycleIntegrationTest.java:
##########
@@ -307,6 +322,112 @@ public void
shouldMoveTasksColdWhenWarmupTasksAreDisabled(final TestInfo testInf
"b should have taken over " + movedTask + " cold, restoring all of
it from the changelog");
}
+ @Test
+ public void shouldKeepAWarmupTaskThroughACoordinatorFailover(final
TestInfo testInfo) throws Exception {
+ setUp(testInfo, 2, appIdWithCoordinatorOffTheControllers(testInfo));
+ final Instance a = startOwnerOfAllTasks();
+
+ final Instance b = startInstance("b", RESTORE_BUDGET);
+ final int task = awaitStagedMigration(a, b);
+
+ // The coordinator holds the intermediate assignment and the reported
offsets in memory only, so the new
+ // coordinator derives the intermediate assignment again from the
members' persisted current assignments, and
+ // knows no lag until the members report their offsets to it. b has to
keep its warm-up task through that.
+ final int coordinator = describeGroup().coordinator().id();
+ assertFalse(CLUSTER.controllerNodeIds().contains(coordinator), "the
group coordinator is on a controller node");
+ CLUSTER.restartBroker(coordinator);
+ waitForGroup("the group to fail over with the warm-up task " + task +
" still on b", group -> isStable(group, 2)
+ && group.coordinator().id() != coordinator
+ && warmupTasks(group, b).equals(Set.of(task))
+ && activeTasks(group, a).contains(task));
+
+ b.gate.open();
+ waitForGroup("the warm-up task " + task + " to be promoted to active
on b", group -> isStable(group, 2)
+ && activeTasks(group, b).equals(Set.of(task))
+ && !activeTasks(group, a).contains(task)
+ && totalWarmupTasks(group) == 0);
+ awaitNoActiveRestore(b, task);
+ assertEquals(1, b.standbyUpdateStarts(task), "b should have kept its
warm-up task open across the failover");
+ }
+
+ @Test
+ public void
shouldContinueRestoringFromAWarmupTaskWhenTheActiveOwnerLeaves(final TestInfo
testInfo) throws Exception {
+ setUp(testInfo, 2);
+ final Instance a = startOwnerOfAllTasks();
+
+ final Instance b = startInstance("b", RESTORE_BUDGET);
+ final int task = awaitStagedMigration(a, b);
+ final int otherTask = task == 0 ? 1 : 0;
+ final long restoredByWarmup = b.gate.delivered.get();
+
+ // With its owner gone, the task runs nowhere, and no caught-up copy
of it exists, so the refiner grants the task
+ // to b right away, and b turns its warm-up task into the active task
in place.
+ instances.remove(a);
+ a.streams.close(Duration.ofSeconds(30));
+ waitForGroup("b to take over both tasks", group -> isStable(group, 1)
+ && activeTasks(group, b).equals(Set.of(0, 1))
+ && totalWarmupTasks(group) == 0);
+
+ // Only open the gate once b restores the task as an active task.
Until b has turned the warm-up task into the
+ // active task, an open gate would let the warm-up task itself catch
up.
+ TestUtils.waitForCondition(
+ () -> b.restoreStartingOffsetByTask.containsKey(task),
+ WAIT_MS,
+ () -> "b never started restoring " + task + " as an active task"
+ );
+ b.gate.open();
+ final Map<Integer, Long> changelogSizes = changelogSizes();
+ // The active task continues from where the warm-up task stopped,
rather than restoring from the start.
+ assertEquals(restoredByWarmup, b.restoreStartingOffsetByTask.get(task),
+ "b should have continued restoring " + task + " from where its
warm-up task stopped");
+ awaitRestoreEnd(b, task, 1);
+ assertEquals(changelogSizes.get(task) - restoredByWarmup,
b.totalRestoredByTask.get(task),
+ "b should have restored only what its warm-up task " + task + "
had not");
+ // The other task had no copy anywhere, so b restores all of it.
+ awaitRestoreEnd(b, otherTask, 1);
+ assertEquals(changelogSizes.get(otherTask),
b.totalRestoredByTask.get(otherTask),
+ "b should have restored all of " + otherTask);
+ }
+
+ @Test
+ public void
shouldNeverRunATaskTwiceWithMembersOfSeveralStreamThreads(final TestInfo
testInfo) throws Exception {
+ setUp(testInfo, 4);
+ final Instance a = startOwnerOfAllTasks();
+
+ final Instance b = startInstance("b", RESTORE_BUDGET, 2);
+ // Each of b's two stream threads is a separate group member, and each
is to take over one of a's tasks.
+ waitForGroup("each of b's threads to warm up one task", group ->
isStable(group, 3)
+ && members(group, b).stream().allMatch(member ->
partitions(member.assignment().warmupTasks()).size() == 1)
+ && warmupTasks(group, b).size() == 2
+ && activeTasks(group, a).size() == 4);
+
+ b.gate.open();
+ waitForGroup("b's threads to take over their tasks", group ->
isStable(group, 3)
+ && members(group, b).stream().allMatch(member ->
partitions(member.assignment().activeTasks()).size() == 1)
+ && activeTasks(group, a).size() == 2
+ && totalWarmupTasks(group) == 0);
+ final Set<Integer> tasksOfB = activeTasks(describeGroup(), b);
+ for (final int task : tasksOfB) {
+ awaitRestoreEnd(b, task, 1);
+ assertEquals(0L, b.totalRestoredByTask.get(task), "b should have
taken over the warmed-up task " + task);
+ }
+
+ // The remaining thread takes over the removed thread's task. Its
process ran the task, but the state lived
+ // in the removed thread's in-memory store, so the remaining thread
restores it from scratch. Handing that
+ // state over between the threads of a process is KAFKA-21090.
+ final String removedThread =
b.streams.removeStreamThread().orElseThrow();
+ final Set<Integer> tasksOfRemovedThread = new
HashSet<>(b.tasksOpenedByThread.getOrDefault(removedThread, Set.of()));
+ tasksOfRemovedThread.retainAll(tasksOfB);
Review Comment:
Right, if the first thread joins first it can get both warm-ups and then
hand one to its sibling, so the history names both. Changed it to snapshot
storeHolders right before removing the thread and take the removed thread's
task from that, and dropped the history map.
--
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]
