@Test public void testHashEnoughResourceOfMultiTaskSlot() { ResourceProfile rp1 = new ResourceProfile(1.0, 100); ResourceProfile rp2 = new ResourceProfile(2.0, 200); ResourceProfile allocatedSlotRp = new ResourceProfile(2.0, 200); final TestingAllocatedSlotActions allocatedSlotActions = new TestingAllocatedSlotActions(); SlotSharingManager slotSharingManager = new SlotSharingManager( SLOT_SHARING_GROUP_ID, allocatedSlotActions, SLOT_OWNER); CompletableFuture<SlotContext> slotContextFuture = new CompletableFuture<>(); SlotSharingManager.MultiTaskSlot unresolvedRootSlot = slotSharingManager.createRootSlot( new SlotRequestId(), slotContextFuture, new SlotRequestId()); SlotSharingManager.MultiTaskSlot multiTaskSlot = unresolvedRootSlot.allocateMultiTaskSlot(new SlotRequestId(), new SlotSharingGroupId()); SlotSharingManager.SingleTaskSlot firstChild = multiTaskSlot.allocateSingleTaskSlot( new SlotRequestId(), rp1, new SlotSharingGroupId(), Locality.LOCAL); assertThat(multiTaskSlot.mayHaveEnoughResourcesToFulfill(rp1), is(true)); assertThat(multiTaskSlot.mayHaveEnoughResourcesToFulfill(rp2), is(true)); assertThat(multiTaskSlot.mayHaveEnoughResourcesToFulfill(ResourceProfile.UNKNOWN), is(true)); slotContextFuture.complete(new AllocatedSlot( new AllocationID(), new TaskManagerLocation(new ResourceID("tm-X"), InetAddress.getLoopbackAddress(), 46), 0, allocatedSlotRp, mock(TaskManagerGateway.class))); assertThat(multiTaskSlot.mayHaveEnoughResourcesToFulfill(rp1), is(true)); assertThat(multiTaskSlot.mayHaveEnoughResourcesToFulfill(rp2), is(false)); assertThat(multiTaskSlot.mayHaveEnoughResourcesToFulfill(ResourceProfile.UNKNOWN), is(true)); }
@Nonnull MultiTaskSlot createRootSlot( SlotRequestId slotRequestId, CompletableFuture<? extends SlotContext> slotContextFuture, SlotRequestId allocatedSlotRequestId) { final MultiTaskSlot rootMultiTaskSlot = new MultiTaskSlot( slotRequestId, slotContextFuture, allocatedSlotRequestId); LOG.debug("Create multi task slot [{}] in slot [{}].", slotRequestId, allocatedSlotRequestId); allTaskSlots.put(slotRequestId, rootMultiTaskSlot); unresolvedRootSlots.put(slotRequestId, rootMultiTaskSlot); slotContextFuture.whenComplete( (SlotContext slotContext, Throwable throwable) -> { if (slotContext != null) { final MultiTaskSlot resolvedRootNode = unresolvedRootSlots.remove(slotRequestId); if (resolvedRootNode != null) { final AllocationID allocationId = slotContext.getAllocationId(); LOG.trace("Fulfill multi task slot [{}] with slot [{}].", slotRequestId, allocationId); final Map<AllocationID, MultiTaskSlot> innerMap = resolvedRootSlots.computeIfAbsent( slotContext.getTaskManagerLocation(), taskManagerLocation -> new HashMap<>(4)); MultiTaskSlot previousValue = innerMap.put(allocationId, resolvedRootNode); Preconditions.checkState(previousValue == null); } } else { rootMultiTaskSlot.release(throwable); } }); return rootMultiTaskSlot; }
SlotSharingManager { @Nonnull MultiTaskSlot createRootSlot( SlotRequestId slotRequestId, CompletableFuture<? extends SlotContext> slotContextFuture, SlotRequestId allocatedSlotRequestId) { final MultiTaskSlot rootMultiTaskSlot = new MultiTaskSlot( slotRequestId, slotContextFuture, allocatedSlotRequestId); LOG.debug("Create multi task slot [{}] in slot [{}].", slotRequestId, allocatedSlotRequestId); allTaskSlots.put(slotRequestId, rootMultiTaskSlot); unresolvedRootSlots.put(slotRequestId, rootMultiTaskSlot); slotContextFuture.whenComplete( (SlotContext slotContext, Throwable throwable) -> { if (slotContext != null) { final MultiTaskSlot resolvedRootNode = unresolvedRootSlots.remove(slotRequestId); if (resolvedRootNode != null) { final AllocationID allocationId = slotContext.getAllocationId(); LOG.trace("Fulfill multi task slot [{}] with slot [{}].", slotRequestId, allocationId); final Map<AllocationID, MultiTaskSlot> innerMap = resolvedRootSlots.computeIfAbsent( slotContext.getTaskManagerLocation(), taskManagerLocation -> new HashMap<>(4)); MultiTaskSlot previousValue = innerMap.put(allocationId, resolvedRootNode); Preconditions.checkState(previousValue == null); } } else { rootMultiTaskSlot.release(throwable); } }); return rootMultiTaskSlot; } }
SlotSharingManager { @Nonnull MultiTaskSlot createRootSlot( SlotRequestId slotRequestId, CompletableFuture<? extends SlotContext> slotContextFuture, SlotRequestId allocatedSlotRequestId) { final MultiTaskSlot rootMultiTaskSlot = new MultiTaskSlot( slotRequestId, slotContextFuture, allocatedSlotRequestId); LOG.debug("Create multi task slot [{}] in slot [{}].", slotRequestId, allocatedSlotRequestId); allTaskSlots.put(slotRequestId, rootMultiTaskSlot); unresolvedRootSlots.put(slotRequestId, rootMultiTaskSlot); slotContextFuture.whenComplete( (SlotContext slotContext, Throwable throwable) -> { if (slotContext != null) { final MultiTaskSlot resolvedRootNode = unresolvedRootSlots.remove(slotRequestId); if (resolvedRootNode != null) { final AllocationID allocationId = slotContext.getAllocationId(); LOG.trace("Fulfill multi task slot [{}] with slot [{}].", slotRequestId, allocationId); final Map<AllocationID, MultiTaskSlot> innerMap = resolvedRootSlots.computeIfAbsent( slotContext.getTaskManagerLocation(), taskManagerLocation -> new HashMap<>(4)); MultiTaskSlot previousValue = innerMap.put(allocationId, resolvedRootNode); Preconditions.checkState(previousValue == null); } } else { rootMultiTaskSlot.release(throwable); } }); return rootMultiTaskSlot; } SlotSharingManager(
			SlotSharingGroupId slotSharingGroupId,
			AllocatedSlotActions allocatedSlotActions,
			SlotOwner slotOwner); }
SlotSharingManager { @Nonnull MultiTaskSlot createRootSlot( SlotRequestId slotRequestId, CompletableFuture<? extends SlotContext> slotContextFuture, SlotRequestId allocatedSlotRequestId) { final MultiTaskSlot rootMultiTaskSlot = new MultiTaskSlot( slotRequestId, slotContextFuture, allocatedSlotRequestId); LOG.debug("Create multi task slot [{}] in slot [{}].", slotRequestId, allocatedSlotRequestId); allTaskSlots.put(slotRequestId, rootMultiTaskSlot); unresolvedRootSlots.put(slotRequestId, rootMultiTaskSlot); slotContextFuture.whenComplete( (SlotContext slotContext, Throwable throwable) -> { if (slotContext != null) { final MultiTaskSlot resolvedRootNode = unresolvedRootSlots.remove(slotRequestId); if (resolvedRootNode != null) { final AllocationID allocationId = slotContext.getAllocationId(); LOG.trace("Fulfill multi task slot [{}] with slot [{}].", slotRequestId, allocationId); final Map<AllocationID, MultiTaskSlot> innerMap = resolvedRootSlots.computeIfAbsent( slotContext.getTaskManagerLocation(), taskManagerLocation -> new HashMap<>(4)); MultiTaskSlot previousValue = innerMap.put(allocationId, resolvedRootNode); Preconditions.checkState(previousValue == null); } } else { rootMultiTaskSlot.release(throwable); } }); return rootMultiTaskSlot; } SlotSharingManager(
			SlotSharingGroupId slotSharingGroupId,
			AllocatedSlotActions allocatedSlotActions,
			SlotOwner slotOwner); boolean isEmpty(); boolean contains(SlotRequestId slotRequestId); @Nonnull Collection<SlotSelectionStrategy.SlotInfoAndResources> listResolvedRootSlotInfo(@Nullable AbstractID groupId); @Nullable MultiTaskSlot getResolvedRootSlot(@Nonnull SlotInfo slotInfo); @Override String toString(); @VisibleForTesting Collection<MultiTaskSlot> getResolvedRootSlots(); }
SlotSharingManager { @Nonnull MultiTaskSlot createRootSlot( SlotRequestId slotRequestId, CompletableFuture<? extends SlotContext> slotContextFuture, SlotRequestId allocatedSlotRequestId) { final MultiTaskSlot rootMultiTaskSlot = new MultiTaskSlot( slotRequestId, slotContextFuture, allocatedSlotRequestId); LOG.debug("Create multi task slot [{}] in slot [{}].", slotRequestId, allocatedSlotRequestId); allTaskSlots.put(slotRequestId, rootMultiTaskSlot); unresolvedRootSlots.put(slotRequestId, rootMultiTaskSlot); slotContextFuture.whenComplete( (SlotContext slotContext, Throwable throwable) -> { if (slotContext != null) { final MultiTaskSlot resolvedRootNode = unresolvedRootSlots.remove(slotRequestId); if (resolvedRootNode != null) { final AllocationID allocationId = slotContext.getAllocationId(); LOG.trace("Fulfill multi task slot [{}] with slot [{}].", slotRequestId, allocationId); final Map<AllocationID, MultiTaskSlot> innerMap = resolvedRootSlots.computeIfAbsent( slotContext.getTaskManagerLocation(), taskManagerLocation -> new HashMap<>(4)); MultiTaskSlot previousValue = innerMap.put(allocationId, resolvedRootNode); Preconditions.checkState(previousValue == null); } } else { rootMultiTaskSlot.release(throwable); } }); return rootMultiTaskSlot; } SlotSharingManager(
			SlotSharingGroupId slotSharingGroupId,
			AllocatedSlotActions allocatedSlotActions,
			SlotOwner slotOwner); boolean isEmpty(); boolean contains(SlotRequestId slotRequestId); @Nonnull Collection<SlotSelectionStrategy.SlotInfoAndResources> listResolvedRootSlotInfo(@Nullable AbstractID groupId); @Nullable MultiTaskSlot getResolvedRootSlot(@Nonnull SlotInfo slotInfo); @Override String toString(); @VisibleForTesting Collection<MultiTaskSlot> getResolvedRootSlots();  }