@Test public void testStop() throws Exception { final ExecutionVertexID executionVertexId = new ExecutionVertexID(new JobVertexID(), 0); final DefaultExecutionSlotAllocator executionSlotAllocator = createExecutionSlotAllocator(); slotProvider.disableSlotAllocation(); final List<ExecutionVertexSchedulingRequirements> schedulingRequirements = createSchedulingRequirements(executionVertexId); executionSlotAllocator.allocateSlotsFor(schedulingRequirements); executionSlotAllocator.stop().get(); assertThat(slotProvider.getCancelledSlotRequestIds(), hasSize(1)); assertThat(slotProvider.getCancelledSlotRequestIds(), contains(slotProvider.getReceivedSlotRequestIds().toArray())); assertEquals(0, executionSlotAllocator.getNumberOfPendingSlotAssignments()); }
@Override public CompletableFuture<Void> stop() { List<ExecutionVertexID> executionVertexIds = new ArrayList<>(pendingSlotAssignments.keySet()); executionVertexIds.forEach(this::cancel); return CompletableFuture.completedFuture(null); }
DefaultExecutionSlotAllocator implements ExecutionSlotAllocator { @Override public CompletableFuture<Void> stop() { List<ExecutionVertexID> executionVertexIds = new ArrayList<>(pendingSlotAssignments.keySet()); executionVertexIds.forEach(this::cancel); return CompletableFuture.completedFuture(null); } }
DefaultExecutionSlotAllocator implements ExecutionSlotAllocator { @Override public CompletableFuture<Void> stop() { List<ExecutionVertexID> executionVertexIds = new ArrayList<>(pendingSlotAssignments.keySet()); executionVertexIds.forEach(this::cancel); return CompletableFuture.completedFuture(null); } DefaultExecutionSlotAllocator(
			SlotProvider slotProvider,
			InputsLocationsRetriever inputsLocationsRetriever,
			Time allocationTimeout); }
DefaultExecutionSlotAllocator implements ExecutionSlotAllocator { @Override public CompletableFuture<Void> stop() { List<ExecutionVertexID> executionVertexIds = new ArrayList<>(pendingSlotAssignments.keySet()); executionVertexIds.forEach(this::cancel); return CompletableFuture.completedFuture(null); } DefaultExecutionSlotAllocator(
			SlotProvider slotProvider,
			InputsLocationsRetriever inputsLocationsRetriever,
			Time allocationTimeout); @Override Collection<SlotExecutionVertexAssignment> allocateSlotsFor(
			Collection<ExecutionVertexSchedulingRequirements> executionVertexSchedulingRequirements); @Override void cancel(ExecutionVertexID executionVertexId); @Override CompletableFuture<Void> stop(); }
DefaultExecutionSlotAllocator implements ExecutionSlotAllocator { @Override public CompletableFuture<Void> stop() { List<ExecutionVertexID> executionVertexIds = new ArrayList<>(pendingSlotAssignments.keySet()); executionVertexIds.forEach(this::cancel); return CompletableFuture.completedFuture(null); } DefaultExecutionSlotAllocator(
			SlotProvider slotProvider,
			InputsLocationsRetriever inputsLocationsRetriever,
			Time allocationTimeout); @Override Collection<SlotExecutionVertexAssignment> allocateSlotsFor(
			Collection<ExecutionVertexSchedulingRequirements> executionVertexSchedulingRequirements); @Override void cancel(ExecutionVertexID executionVertexId); @Override CompletableFuture<Void> stop();  }