tillrohrmann commented on a change in pull request #10682:
[FLINK-15247][Runtime] Wait for all slots to be free before task executor
services shutdown upon stopping
URL: https://github.com/apache/flink/pull/10682#discussion_r369052941
##########
File path:
flink-runtime/src/test/java/org/apache/flink/runtime/taskexecutor/slot/TaskSlotTableTest.java
##########
@@ -238,8 +208,112 @@ public void testGenerateSlotReport() throws
SlotNotFoundException {
is(new SlotStatus(new SlotID(resourceId, 1),
TaskSlotUtils.DEFAULT_RESOURCE_PROFILE, null, null)),
is(new SlotStatus(new SlotID(resourceId, 2),
TaskSlotUtils.DEFAULT_RESOURCE_PROFILE, null, null)),
is(new
SlotStatus(SlotID.generateDynamicSlotID(resourceId),
TaskSlotUtils.DEFAULT_RESOURCE_PROFILE, jobId, allocationId3))));
- } finally {
- taskSlotTable.stop();
}
}
+
+ @Test
+ public void testAllocateSlot() throws Exception {
+ final JobID jobId = new JobID();
+ final AllocationID allocationId = new AllocationID();
+ try (final TaskSlotTable taskSlotTable =
+ createTaskSlotTableWithAllocatedSlot(jobId,
allocationId, new TestingSlotActionsBuilder().build())) {
+ Iterator<TaskSlot> allocatedSlots =
taskSlotTable.getAllocatedSlots(jobId);
+ TaskSlot nextSlot = allocatedSlots.next();
+ assertThat(nextSlot.getIndex(), is(0));
+ assertThat(nextSlot.getAllocationId(),
is(allocationId));
+ assertThat(nextSlot.getJobId(), is(jobId));
+ assertThat(allocatedSlots.hasNext(), is(false));
+ }
+ }
+
+ @Test
+ public void testAddTask() throws Exception {
+ final JobID jobId = new JobID();
+ final ExecutionAttemptID executionAttemptId = new
ExecutionAttemptID();
+ final AllocationID allocationId = new AllocationID();
+ Task task = TaskSlotUtils.createTask(
+ jobId,
+ executionAttemptId,
+ allocationId,
+ EmptyInvokable.class);
+ try (final TaskSlotTable taskSlotTable =
+ createTaskSlotTableWithStartedTask(task, new
TestingSlotActionsBuilder().build())) {
+ Iterator<Task> tasks = taskSlotTable.getTasks(jobId);
+ Task nextTask = tasks.next();
+ assertThat(nextTask.getExecutionId(),
is(executionAttemptId));
+ assertThat(nextTask.getAllocationId(),
is(allocationId));
+ assertThat(tasks.hasNext(), is(false));
+ }
+ }
+
+ @Test(timeout = 10000)
+ public void testRemoveTaskCallsFreeSlotAction() throws Exception {
+ final JobID jobId = new JobID();
+ final ExecutionAttemptID executionAttemptId = new
ExecutionAttemptID();
+ final AllocationID allocationId = new AllocationID();
+ CompletableFuture<AllocationID> freeSlotFuture = new
CompletableFuture<>();
+ SlotActions slotActions = new
TestingSlotActions(freeSlotFuture::complete, (aid, uid) -> {});
+ Task task = TaskSlotUtils.createTask(
+ jobId,
+ executionAttemptId,
+ allocationId,
+ EmptyInvokable.class);
+ try (final TaskSlotTable taskSlotTable =
createTaskSlotTableWithStartedTask(task, slotActions)) {
+ // we have to initiate closing of the slot externally
+ // to enable that the last remaining finished task does
the final slot freeing
+ taskSlotTable.freeSlot(allocationId);
+ taskSlotTable.removeTask(executionAttemptId);
+ assertThat(freeSlotFuture.get(), is(allocationId));
+ }
+ }
+
+ @Test(timeout = 10000)
+ public void testFreeSlotInterruptsSubmittedTask() throws Exception {
+ final JobID jobId = new JobID();
+ final ExecutionAttemptID executionAttemptId = new
ExecutionAttemptID();
+ final AllocationID allocationId = new AllocationID();
+ TestInterruptableInvokable.reset();
+ Task task = TaskSlotUtils.createTask(
+ jobId,
+ executionAttemptId,
+ allocationId,
+ TestInterruptableInvokable.class);
+ final TaskSlotTable taskSlotTable =
createTaskSlotTableWithStartedTask(task, new
TestingSlotActionsBuilder().build());
+ TestInterruptableInvokable.waitForStarted();
+ assertThat(taskSlotTable.freeSlot(allocationId), is(-1));
+ TestInterruptableInvokable.waitForInterrupted();
+ CompletableFuture<Void> closingFuture =
taskSlotTable.closeAsync();
+ assertThat(closingFuture.isDone(), is(false));
+ TestInterruptableInvokable.finish();
+ closingFuture.get();
Review comment:
nit: technically we are testing two things here. Don't have to change this
now but keep this in mind for the future.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
With regards,
Apache Git Services
