@Test public void testSlotContextFutureCompletion() throws Exception { final TestingAllocatedSlotActions allocatedSlotActions = new TestingAllocatedSlotActions(); final SlotSharingManager slotSharingManager = new SlotSharingManager( SLOT_SHARING_GROUP_ID, allocatedSlotActions, SLOT_OWNER); final SlotContext slotContext = new SimpleSlotContext( new AllocationID(), new LocalTaskManagerLocation(), 0, new SimpleAckingTaskManagerGateway()); CompletableFuture<SlotContext> slotContextFuture = new CompletableFuture<>(); SlotSharingManager.MultiTaskSlot rootSlot = slotSharingManager.createRootSlot( new SlotRequestId(), slotContextFuture, new SlotRequestId()); Locality locality1 = Locality.LOCAL; SlotSharingManager.SingleTaskSlot singleTaskSlot1 = rootSlot.allocateSingleTaskSlot( new SlotRequestId(), ResourceProfile.UNKNOWN, new AbstractID(), locality1); Locality locality2 = Locality.HOST_LOCAL; SlotSharingManager.SingleTaskSlot singleTaskSlot2 = rootSlot.allocateSingleTaskSlot( new SlotRequestId(), ResourceProfile.UNKNOWN, new AbstractID(), locality2); CompletableFuture<LogicalSlot> logicalSlotFuture1 = singleTaskSlot1.getLogicalSlotFuture(); CompletableFuture<LogicalSlot> logicalSlotFuture2 = singleTaskSlot2.getLogicalSlotFuture(); assertFalse(logicalSlotFuture1.isDone()); assertFalse(logicalSlotFuture2.isDone()); slotContextFuture.complete(slotContext); assertTrue(logicalSlotFuture1.isDone()); assertTrue(logicalSlotFuture2.isDone()); final LogicalSlot logicalSlot1 = logicalSlotFuture1.get(); final LogicalSlot logicalSlot2 = logicalSlotFuture2.get(); assertEquals(logicalSlot1.getAllocationId(), slotContext.getAllocationId()); assertEquals(logicalSlot2.getAllocationId(), slotContext.getAllocationId()); assertEquals(locality1, logicalSlot1.getLocality()); assertEquals(locality2, logicalSlot2.getLocality()); Locality locality3 = Locality.NON_LOCAL; SlotSharingManager.SingleTaskSlot singleTaskSlot3 = rootSlot.allocateSingleTaskSlot( new SlotRequestId(), ResourceProfile.UNKNOWN, new AbstractID(), locality3); CompletableFuture<LogicalSlot> logicalSlotFuture3 = singleTaskSlot3.getLogicalSlotFuture(); assertTrue(logicalSlotFuture3.isDone()); LogicalSlot logicalSlot3 = logicalSlotFuture3.get(); assertEquals(locality3, logicalSlot3.getLocality()); assertEquals(slotContext.getAllocationId(), logicalSlot3.getAllocationId()); }
@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();  }