@Test public void testAllocateSlotsParameters() { final ExecutionVertexID executionVertexId = new ExecutionVertexID(new JobVertexID(), 0); final AllocationID allocationId = new AllocationID(); final SlotSharingGroupId sharingGroupId = new SlotSharingGroupId(); final ResourceProfile resourceProfile = new ResourceProfile(0.5, 250); final CoLocationConstraint coLocationConstraint = new CoLocationGroup().getLocationConstraint(0); final Collection<TaskManagerLocation> taskManagerLocations = Collections.singleton(new LocalTaskManagerLocation()); final DefaultExecutionSlotAllocator executionSlotAllocator = createExecutionSlotAllocator(); final List<ExecutionVertexSchedulingRequirements> schedulingRequirements = Arrays.asList( new ExecutionVertexSchedulingRequirements.Builder() .withExecutionVertexId(executionVertexId) .withPreviousAllocationId(allocationId) .withSlotSharingGroupId(sharingGroupId) .withPreferredLocations(taskManagerLocations) .withResourceProfile(resourceProfile) .withCoLocationConstraint(coLocationConstraint) .build() ); executionSlotAllocator.allocateSlotsFor(schedulingRequirements); assertThat(slotProvider.getSlotAllocationRequests(), hasSize(1)); ScheduledUnit expectedTask = slotProvider.getSlotAllocationRequests().get(0).f1; SlotProfile expectedSlotProfile = slotProvider.getSlotAllocationRequests().get(0).f2; assertEquals(sharingGroupId, expectedTask.getSlotSharingGroupId()); assertEquals(coLocationConstraint, expectedTask.getCoLocationConstraint()); assertThat(expectedSlotProfile.getPreferredAllocations(), contains(allocationId)); assertThat(expectedSlotProfile.getPreviousExecutionGraphAllocations(), contains(allocationId)); assertEquals(resourceProfile, expectedSlotProfile.getResourceProfile()); assertThat(expectedSlotProfile.getPreferredLocations(), contains(taskManagerLocations.toArray())); }
@Override public Collection<SlotExecutionVertexAssignment> allocateSlotsFor( Collection<ExecutionVertexSchedulingRequirements> executionVertexSchedulingRequirements) { List<SlotExecutionVertexAssignment> slotExecutionVertexAssignments = new ArrayList<>(executionVertexSchedulingRequirements.size()); Set<AllocationID> allPreviousAllocationIds = computeAllPriorAllocationIds(executionVertexSchedulingRequirements); for (ExecutionVertexSchedulingRequirements schedulingRequirements : executionVertexSchedulingRequirements) { final ExecutionVertexID executionVertexId = schedulingRequirements.getExecutionVertexId(); final SlotRequestId slotRequestId = new SlotRequestId(); final SlotSharingGroupId slotSharingGroupId = schedulingRequirements.getSlotSharingGroupId(); LOG.debug("Allocate slot with id {} for execution {}", slotRequestId, executionVertexId); CompletableFuture<LogicalSlot> slotFuture = calculatePreferredLocations( executionVertexId, schedulingRequirements.getPreferredLocations(), inputsLocationsRetriever).thenCompose( (Collection<TaskManagerLocation> preferredLocations) -> slotProvider.allocateSlot( slotRequestId, new ScheduledUnit( executionVertexId.getJobVertexId(), slotSharingGroupId, schedulingRequirements.getCoLocationConstraint()), new SlotProfile( schedulingRequirements.getResourceProfile(), preferredLocations, Arrays.asList(schedulingRequirements.getPreviousAllocationId()), allPreviousAllocationIds), true, allocationTimeout)); SlotExecutionVertexAssignment slotExecutionVertexAssignment = new SlotExecutionVertexAssignment(executionVertexId, slotFuture); pendingSlotAssignments.put(executionVertexId, slotExecutionVertexAssignment); slotFuture.whenComplete( (ignored, throwable) -> { pendingSlotAssignments.remove(executionVertexId); if (throwable != null) { slotProvider.cancelSlotRequest(slotRequestId, slotSharingGroupId, throwable); } }); slotExecutionVertexAssignments.add(slotExecutionVertexAssignment); } return slotExecutionVertexAssignments; }
DefaultExecutionSlotAllocator implements ExecutionSlotAllocator { @Override public Collection<SlotExecutionVertexAssignment> allocateSlotsFor( Collection<ExecutionVertexSchedulingRequirements> executionVertexSchedulingRequirements) { List<SlotExecutionVertexAssignment> slotExecutionVertexAssignments = new ArrayList<>(executionVertexSchedulingRequirements.size()); Set<AllocationID> allPreviousAllocationIds = computeAllPriorAllocationIds(executionVertexSchedulingRequirements); for (ExecutionVertexSchedulingRequirements schedulingRequirements : executionVertexSchedulingRequirements) { final ExecutionVertexID executionVertexId = schedulingRequirements.getExecutionVertexId(); final SlotRequestId slotRequestId = new SlotRequestId(); final SlotSharingGroupId slotSharingGroupId = schedulingRequirements.getSlotSharingGroupId(); LOG.debug("Allocate slot with id {} for execution {}", slotRequestId, executionVertexId); CompletableFuture<LogicalSlot> slotFuture = calculatePreferredLocations( executionVertexId, schedulingRequirements.getPreferredLocations(), inputsLocationsRetriever).thenCompose( (Collection<TaskManagerLocation> preferredLocations) -> slotProvider.allocateSlot( slotRequestId, new ScheduledUnit( executionVertexId.getJobVertexId(), slotSharingGroupId, schedulingRequirements.getCoLocationConstraint()), new SlotProfile( schedulingRequirements.getResourceProfile(), preferredLocations, Arrays.asList(schedulingRequirements.getPreviousAllocationId()), allPreviousAllocationIds), true, allocationTimeout)); SlotExecutionVertexAssignment slotExecutionVertexAssignment = new SlotExecutionVertexAssignment(executionVertexId, slotFuture); pendingSlotAssignments.put(executionVertexId, slotExecutionVertexAssignment); slotFuture.whenComplete( (ignored, throwable) -> { pendingSlotAssignments.remove(executionVertexId); if (throwable != null) { slotProvider.cancelSlotRequest(slotRequestId, slotSharingGroupId, throwable); } }); slotExecutionVertexAssignments.add(slotExecutionVertexAssignment); } return slotExecutionVertexAssignments; } }
DefaultExecutionSlotAllocator implements ExecutionSlotAllocator { @Override public Collection<SlotExecutionVertexAssignment> allocateSlotsFor( Collection<ExecutionVertexSchedulingRequirements> executionVertexSchedulingRequirements) { List<SlotExecutionVertexAssignment> slotExecutionVertexAssignments = new ArrayList<>(executionVertexSchedulingRequirements.size()); Set<AllocationID> allPreviousAllocationIds = computeAllPriorAllocationIds(executionVertexSchedulingRequirements); for (ExecutionVertexSchedulingRequirements schedulingRequirements : executionVertexSchedulingRequirements) { final ExecutionVertexID executionVertexId = schedulingRequirements.getExecutionVertexId(); final SlotRequestId slotRequestId = new SlotRequestId(); final SlotSharingGroupId slotSharingGroupId = schedulingRequirements.getSlotSharingGroupId(); LOG.debug("Allocate slot with id {} for execution {}", slotRequestId, executionVertexId); CompletableFuture<LogicalSlot> slotFuture = calculatePreferredLocations( executionVertexId, schedulingRequirements.getPreferredLocations(), inputsLocationsRetriever).thenCompose( (Collection<TaskManagerLocation> preferredLocations) -> slotProvider.allocateSlot( slotRequestId, new ScheduledUnit( executionVertexId.getJobVertexId(), slotSharingGroupId, schedulingRequirements.getCoLocationConstraint()), new SlotProfile( schedulingRequirements.getResourceProfile(), preferredLocations, Arrays.asList(schedulingRequirements.getPreviousAllocationId()), allPreviousAllocationIds), true, allocationTimeout)); SlotExecutionVertexAssignment slotExecutionVertexAssignment = new SlotExecutionVertexAssignment(executionVertexId, slotFuture); pendingSlotAssignments.put(executionVertexId, slotExecutionVertexAssignment); slotFuture.whenComplete( (ignored, throwable) -> { pendingSlotAssignments.remove(executionVertexId); if (throwable != null) { slotProvider.cancelSlotRequest(slotRequestId, slotSharingGroupId, throwable); } }); slotExecutionVertexAssignments.add(slotExecutionVertexAssignment); } return slotExecutionVertexAssignments; } DefaultExecutionSlotAllocator(
			SlotProvider slotProvider,
			InputsLocationsRetriever inputsLocationsRetriever,
			Time allocationTimeout); }
DefaultExecutionSlotAllocator implements ExecutionSlotAllocator { @Override public Collection<SlotExecutionVertexAssignment> allocateSlotsFor( Collection<ExecutionVertexSchedulingRequirements> executionVertexSchedulingRequirements) { List<SlotExecutionVertexAssignment> slotExecutionVertexAssignments = new ArrayList<>(executionVertexSchedulingRequirements.size()); Set<AllocationID> allPreviousAllocationIds = computeAllPriorAllocationIds(executionVertexSchedulingRequirements); for (ExecutionVertexSchedulingRequirements schedulingRequirements : executionVertexSchedulingRequirements) { final ExecutionVertexID executionVertexId = schedulingRequirements.getExecutionVertexId(); final SlotRequestId slotRequestId = new SlotRequestId(); final SlotSharingGroupId slotSharingGroupId = schedulingRequirements.getSlotSharingGroupId(); LOG.debug("Allocate slot with id {} for execution {}", slotRequestId, executionVertexId); CompletableFuture<LogicalSlot> slotFuture = calculatePreferredLocations( executionVertexId, schedulingRequirements.getPreferredLocations(), inputsLocationsRetriever).thenCompose( (Collection<TaskManagerLocation> preferredLocations) -> slotProvider.allocateSlot( slotRequestId, new ScheduledUnit( executionVertexId.getJobVertexId(), slotSharingGroupId, schedulingRequirements.getCoLocationConstraint()), new SlotProfile( schedulingRequirements.getResourceProfile(), preferredLocations, Arrays.asList(schedulingRequirements.getPreviousAllocationId()), allPreviousAllocationIds), true, allocationTimeout)); SlotExecutionVertexAssignment slotExecutionVertexAssignment = new SlotExecutionVertexAssignment(executionVertexId, slotFuture); pendingSlotAssignments.put(executionVertexId, slotExecutionVertexAssignment); slotFuture.whenComplete( (ignored, throwable) -> { pendingSlotAssignments.remove(executionVertexId); if (throwable != null) { slotProvider.cancelSlotRequest(slotRequestId, slotSharingGroupId, throwable); } }); slotExecutionVertexAssignments.add(slotExecutionVertexAssignment); } return slotExecutionVertexAssignments; } 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 Collection<SlotExecutionVertexAssignment> allocateSlotsFor( Collection<ExecutionVertexSchedulingRequirements> executionVertexSchedulingRequirements) { List<SlotExecutionVertexAssignment> slotExecutionVertexAssignments = new ArrayList<>(executionVertexSchedulingRequirements.size()); Set<AllocationID> allPreviousAllocationIds = computeAllPriorAllocationIds(executionVertexSchedulingRequirements); for (ExecutionVertexSchedulingRequirements schedulingRequirements : executionVertexSchedulingRequirements) { final ExecutionVertexID executionVertexId = schedulingRequirements.getExecutionVertexId(); final SlotRequestId slotRequestId = new SlotRequestId(); final SlotSharingGroupId slotSharingGroupId = schedulingRequirements.getSlotSharingGroupId(); LOG.debug("Allocate slot with id {} for execution {}", slotRequestId, executionVertexId); CompletableFuture<LogicalSlot> slotFuture = calculatePreferredLocations( executionVertexId, schedulingRequirements.getPreferredLocations(), inputsLocationsRetriever).thenCompose( (Collection<TaskManagerLocation> preferredLocations) -> slotProvider.allocateSlot( slotRequestId, new ScheduledUnit( executionVertexId.getJobVertexId(), slotSharingGroupId, schedulingRequirements.getCoLocationConstraint()), new SlotProfile( schedulingRequirements.getResourceProfile(), preferredLocations, Arrays.asList(schedulingRequirements.getPreviousAllocationId()), allPreviousAllocationIds), true, allocationTimeout)); SlotExecutionVertexAssignment slotExecutionVertexAssignment = new SlotExecutionVertexAssignment(executionVertexId, slotFuture); pendingSlotAssignments.put(executionVertexId, slotExecutionVertexAssignment); slotFuture.whenComplete( (ignored, throwable) -> { pendingSlotAssignments.remove(executionVertexId); if (throwable != null) { slotProvider.cancelSlotRequest(slotRequestId, slotSharingGroupId, throwable); } }); slotExecutionVertexAssignments.add(slotExecutionVertexAssignment); } return slotExecutionVertexAssignments; } DefaultExecutionSlotAllocator(
			SlotProvider slotProvider,
			InputsLocationsRetriever inputsLocationsRetriever,
			Time allocationTimeout); @Override Collection<SlotExecutionVertexAssignment> allocateSlotsFor(
			Collection<ExecutionVertexSchedulingRequirements> executionVertexSchedulingRequirements); @Override void cancel(ExecutionVertexID executionVertexId); @Override CompletableFuture<Void> stop();  }