@Test public void testComputeAllPriorAllocationIds() { List<AllocationID> expectAllocationIds = Arrays.asList(new AllocationID(), new AllocationID()); List<ExecutionVertexSchedulingRequirements> testSchedulingRequirements = Arrays.asList( new ExecutionVertexSchedulingRequirements.Builder(). withExecutionVertexId(new ExecutionVertexID(new JobVertexID(), 0)). withPreviousAllocationId(expectAllocationIds.get(0)). build(), new ExecutionVertexSchedulingRequirements.Builder(). withExecutionVertexId(new ExecutionVertexID(new JobVertexID(), 1)). withPreviousAllocationId(expectAllocationIds.get(0)). build(), new ExecutionVertexSchedulingRequirements.Builder(). withExecutionVertexId(new ExecutionVertexID(new JobVertexID(), 2)). withPreviousAllocationId(expectAllocationIds.get(1)). build(), new ExecutionVertexSchedulingRequirements.Builder(). withExecutionVertexId(new ExecutionVertexID(new JobVertexID(), 3)). build() ); Set<AllocationID> allPriorAllocationIds = DefaultExecutionSlotAllocator.computeAllPriorAllocationIds(testSchedulingRequirements); assertThat(allPriorAllocationIds, containsInAnyOrder(expectAllocationIds.toArray())); }
@VisibleForTesting static Set<AllocationID> computeAllPriorAllocationIds( Collection<ExecutionVertexSchedulingRequirements> executionVertexSchedulingRequirements) { return executionVertexSchedulingRequirements .stream() .map(ExecutionVertexSchedulingRequirements::getPreviousAllocationId) .filter(Objects::nonNull) .collect(Collectors.toSet()); }
DefaultExecutionSlotAllocator implements ExecutionSlotAllocator { @VisibleForTesting static Set<AllocationID> computeAllPriorAllocationIds( Collection<ExecutionVertexSchedulingRequirements> executionVertexSchedulingRequirements) { return executionVertexSchedulingRequirements .stream() .map(ExecutionVertexSchedulingRequirements::getPreviousAllocationId) .filter(Objects::nonNull) .collect(Collectors.toSet()); } }
DefaultExecutionSlotAllocator implements ExecutionSlotAllocator { @VisibleForTesting static Set<AllocationID> computeAllPriorAllocationIds( Collection<ExecutionVertexSchedulingRequirements> executionVertexSchedulingRequirements) { return executionVertexSchedulingRequirements .stream() .map(ExecutionVertexSchedulingRequirements::getPreviousAllocationId) .filter(Objects::nonNull) .collect(Collectors.toSet()); } DefaultExecutionSlotAllocator(
			SlotProvider slotProvider,
			InputsLocationsRetriever inputsLocationsRetriever,
			Time allocationTimeout); }
DefaultExecutionSlotAllocator implements ExecutionSlotAllocator { @VisibleForTesting static Set<AllocationID> computeAllPriorAllocationIds( Collection<ExecutionVertexSchedulingRequirements> executionVertexSchedulingRequirements) { return executionVertexSchedulingRequirements .stream() .map(ExecutionVertexSchedulingRequirements::getPreviousAllocationId) .filter(Objects::nonNull) .collect(Collectors.toSet()); } 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 { @VisibleForTesting static Set<AllocationID> computeAllPriorAllocationIds( Collection<ExecutionVertexSchedulingRequirements> executionVertexSchedulingRequirements) { return executionVertexSchedulingRequirements .stream() .map(ExecutionVertexSchedulingRequirements::getPreviousAllocationId) .filter(Objects::nonNull) .collect(Collectors.toSet()); } DefaultExecutionSlotAllocator(
			SlotProvider slotProvider,
			InputsLocationsRetriever inputsLocationsRetriever,
			Time allocationTimeout); @Override Collection<SlotExecutionVertexAssignment> allocateSlotsFor(
			Collection<ExecutionVertexSchedulingRequirements> executionVertexSchedulingRequirements); @Override void cancel(ExecutionVertexID executionVertexId); @Override CompletableFuture<Void> stop();  }