@Test public void testCreateAllocatedSlotReport() throws Exception { try (SlotPoolImpl slotPool = createSlotPoolImpl()) { final ArrayBlockingQueue<AllocationID> allocationIds = new ArrayBlockingQueue<>(1); resourceManagerGateway.setRequestSlotConsumer( slotRequest -> allocationIds.offer(slotRequest.getAllocationId())); setupSlotPool(slotPool, resourceManagerGateway, mainThreadExecutor); Scheduler scheduler = setupScheduler(slotPool, mainThreadExecutor); final SlotRequestId slotRequestId = new SlotRequestId(); final CompletableFuture<LogicalSlot> slotRequestFuture = allocateSlot(scheduler, slotRequestId); final List<AllocatedSlotInfo> allocatedSlotInfos = new ArrayList<>(2); final List<SlotOffer> slotOffers = new ArrayList<>(2); final AllocationID allocatedId = allocationIds.take(); slotOffers.add(new SlotOffer(allocatedId, 0, ResourceProfile.UNKNOWN)); allocatedSlotInfos.add(new AllocatedSlotInfo(0, allocatedId)); final AllocationID availableId = new AllocationID(); slotOffers.add(new SlotOffer(availableId, 1, ResourceProfile.UNKNOWN)); allocatedSlotInfos.add(new AllocatedSlotInfo(1, availableId)); slotPool.registerTaskManager(taskManagerLocation.getResourceID()); slotPool.offerSlots(taskManagerLocation, taskManagerGateway, slotOffers); slotRequestFuture.get(); final AllocatedSlotReport slotReport = slotPool.createAllocatedSlotReport(taskManagerLocation.getResourceID()); assertThat(jobId, is(slotReport.getJobId())); assertThat(slotReport.getAllocatedSlotInfos(), containsInAnyOrder(isEachEqual(allocatedSlotInfos))); } }
@Override public AllocatedSlotReport createAllocatedSlotReport(ResourceID taskManagerId) { final Set<AllocatedSlot> availableSlotsForTaskManager = availableSlots.getSlotsForTaskManager(taskManagerId); final Set<AllocatedSlot> allocatedSlotsForTaskManager = allocatedSlots.getSlotsForTaskManager(taskManagerId); List<AllocatedSlotInfo> allocatedSlotInfos = new ArrayList<>( availableSlotsForTaskManager.size() + allocatedSlotsForTaskManager.size()); for (AllocatedSlot allocatedSlot : Iterables.concat(availableSlotsForTaskManager, allocatedSlotsForTaskManager)) { allocatedSlotInfos.add( new AllocatedSlotInfo(allocatedSlot.getPhysicalSlotNumber(), allocatedSlot.getAllocationId())); } return new AllocatedSlotReport(jobId, allocatedSlotInfos); }
SlotPoolImpl implements SlotPool { @Override public AllocatedSlotReport createAllocatedSlotReport(ResourceID taskManagerId) { final Set<AllocatedSlot> availableSlotsForTaskManager = availableSlots.getSlotsForTaskManager(taskManagerId); final Set<AllocatedSlot> allocatedSlotsForTaskManager = allocatedSlots.getSlotsForTaskManager(taskManagerId); List<AllocatedSlotInfo> allocatedSlotInfos = new ArrayList<>( availableSlotsForTaskManager.size() + allocatedSlotsForTaskManager.size()); for (AllocatedSlot allocatedSlot : Iterables.concat(availableSlotsForTaskManager, allocatedSlotsForTaskManager)) { allocatedSlotInfos.add( new AllocatedSlotInfo(allocatedSlot.getPhysicalSlotNumber(), allocatedSlot.getAllocationId())); } return new AllocatedSlotReport(jobId, allocatedSlotInfos); } }
SlotPoolImpl implements SlotPool { @Override public AllocatedSlotReport createAllocatedSlotReport(ResourceID taskManagerId) { final Set<AllocatedSlot> availableSlotsForTaskManager = availableSlots.getSlotsForTaskManager(taskManagerId); final Set<AllocatedSlot> allocatedSlotsForTaskManager = allocatedSlots.getSlotsForTaskManager(taskManagerId); List<AllocatedSlotInfo> allocatedSlotInfos = new ArrayList<>( availableSlotsForTaskManager.size() + allocatedSlotsForTaskManager.size()); for (AllocatedSlot allocatedSlot : Iterables.concat(availableSlotsForTaskManager, allocatedSlotsForTaskManager)) { allocatedSlotInfos.add( new AllocatedSlotInfo(allocatedSlot.getPhysicalSlotNumber(), allocatedSlot.getAllocationId())); } return new AllocatedSlotReport(jobId, allocatedSlotInfos); } SlotPoolImpl(
			JobID jobId,
			Clock clock,
			Time rpcTimeout,
			Time idleSlotTimeout,
			Time batchSlotTimeout); }
SlotPoolImpl implements SlotPool { @Override public AllocatedSlotReport createAllocatedSlotReport(ResourceID taskManagerId) { final Set<AllocatedSlot> availableSlotsForTaskManager = availableSlots.getSlotsForTaskManager(taskManagerId); final Set<AllocatedSlot> allocatedSlotsForTaskManager = allocatedSlots.getSlotsForTaskManager(taskManagerId); List<AllocatedSlotInfo> allocatedSlotInfos = new ArrayList<>( availableSlotsForTaskManager.size() + allocatedSlotsForTaskManager.size()); for (AllocatedSlot allocatedSlot : Iterables.concat(availableSlotsForTaskManager, allocatedSlotsForTaskManager)) { allocatedSlotInfos.add( new AllocatedSlotInfo(allocatedSlot.getPhysicalSlotNumber(), allocatedSlot.getAllocationId())); } return new AllocatedSlotReport(jobId, allocatedSlotInfos); } SlotPoolImpl(
			JobID jobId,
			Clock clock,
			Time rpcTimeout,
			Time idleSlotTimeout,
			Time batchSlotTimeout); void start(
		@Nonnull JobMasterId jobMasterId,
		@Nonnull String newJobManagerAddress,
		@Nonnull ComponentMainThreadExecutor componentMainThreadExecutor); @Override void suspend(); @Override void close(); @Override void connectToResourceManager(@Nonnull ResourceManagerGateway resourceManagerGateway); @Override void disconnectResourceManager(); @Override void releaseSlot(@Nonnull SlotRequestId slotRequestId, @Nullable Throwable cause); @Override Optional<PhysicalSlot> allocateAvailableSlot(
		@Nonnull SlotRequestId slotRequestId,
		@Nonnull AllocationID allocationID); @Nonnull @Override CompletableFuture<PhysicalSlot> requestNewAllocatedSlot(
			@Nonnull SlotRequestId slotRequestId,
			@Nonnull ResourceProfile resourceProfile,
			Time timeout); @Nonnull @Override CompletableFuture<PhysicalSlot> requestNewAllocatedBatchSlot(
		@Nonnull SlotRequestId slotRequestId,
		@Nonnull ResourceProfile resourceProfile); @Override @Nonnull Collection<SlotInfo> getAvailableSlotsInformation(); @Override Collection<SlotOffer> offerSlots(
			TaskManagerLocation taskManagerLocation,
			TaskManagerGateway taskManagerGateway,
			Collection<SlotOffer> offers); @Override Optional<ResourceID> failAllocation(final AllocationID allocationID, final Exception cause); @Override boolean registerTaskManager(final ResourceID resourceID); @Override boolean releaseTaskManager(final ResourceID resourceId, final Exception cause); @Override AllocatedSlotReport createAllocatedSlotReport(ResourceID taskManagerId); }
SlotPoolImpl implements SlotPool { @Override public AllocatedSlotReport createAllocatedSlotReport(ResourceID taskManagerId) { final Set<AllocatedSlot> availableSlotsForTaskManager = availableSlots.getSlotsForTaskManager(taskManagerId); final Set<AllocatedSlot> allocatedSlotsForTaskManager = allocatedSlots.getSlotsForTaskManager(taskManagerId); List<AllocatedSlotInfo> allocatedSlotInfos = new ArrayList<>( availableSlotsForTaskManager.size() + allocatedSlotsForTaskManager.size()); for (AllocatedSlot allocatedSlot : Iterables.concat(availableSlotsForTaskManager, allocatedSlotsForTaskManager)) { allocatedSlotInfos.add( new AllocatedSlotInfo(allocatedSlot.getPhysicalSlotNumber(), allocatedSlot.getAllocationId())); } return new AllocatedSlotReport(jobId, allocatedSlotInfos); } SlotPoolImpl(
			JobID jobId,
			Clock clock,
			Time rpcTimeout,
			Time idleSlotTimeout,
			Time batchSlotTimeout); void start(
		@Nonnull JobMasterId jobMasterId,
		@Nonnull String newJobManagerAddress,
		@Nonnull ComponentMainThreadExecutor componentMainThreadExecutor); @Override void suspend(); @Override void close(); @Override void connectToResourceManager(@Nonnull ResourceManagerGateway resourceManagerGateway); @Override void disconnectResourceManager(); @Override void releaseSlot(@Nonnull SlotRequestId slotRequestId, @Nullable Throwable cause); @Override Optional<PhysicalSlot> allocateAvailableSlot(
		@Nonnull SlotRequestId slotRequestId,
		@Nonnull AllocationID allocationID); @Nonnull @Override CompletableFuture<PhysicalSlot> requestNewAllocatedSlot(
			@Nonnull SlotRequestId slotRequestId,
			@Nonnull ResourceProfile resourceProfile,
			Time timeout); @Nonnull @Override CompletableFuture<PhysicalSlot> requestNewAllocatedBatchSlot(
		@Nonnull SlotRequestId slotRequestId,
		@Nonnull ResourceProfile resourceProfile); @Override @Nonnull Collection<SlotInfo> getAvailableSlotsInformation(); @Override Collection<SlotOffer> offerSlots(
			TaskManagerLocation taskManagerLocation,
			TaskManagerGateway taskManagerGateway,
			Collection<SlotOffer> offers); @Override Optional<ResourceID> failAllocation(final AllocationID allocationID, final Exception cause); @Override boolean registerTaskManager(final ResourceID resourceID); @Override boolean releaseTaskManager(final ResourceID resourceId, final Exception cause); @Override AllocatedSlotReport createAllocatedSlotReport(ResourceID taskManagerId);  }