### Question:
SlotSharingManager { MultiTaskSlot createRootSlot( SlotRequestId slotRequestId, CompletableFuture<? extends SlotContext> slotContextFuture, SlotRequestId allocatedSlotRequestId) { final MultiTaskSlot rootMultiTaskSlot = new MultiTaskSlot( slotRequestId, slotContextFuture, allocatedSlotRequestId); allTaskSlots.put(slotRequestId, rootMultiTaskSlot); synchronized (lock) { unresolvedRootSlots.put(slotRequestId, rootMultiTaskSlot); } slotContextFuture.whenComplete( (SlotContext slotContext, Throwable throwable) -> { if (slotContext != null) { synchronized (lock) { final MultiTaskSlot resolvedRootNode = unresolvedRootSlots.remove(slotRequestId); if (resolvedRootNode != null) { final Set<MultiTaskSlot> innerCollection = resolvedRootSlots.computeIfAbsent( slotContext.getTaskManagerLocation(), taskManagerLocation -> new HashSet<>(4)); innerCollection.add(resolvedRootNode); } } } else { rootMultiTaskSlot.release(throwable); } }); return rootMultiTaskSlot; } SlotSharingManager(
			SlotSharingGroupId slotSharingGroupId,
			AllocatedSlotActions allocatedSlotActions,
			SlotOwner slotOwner); boolean isEmpty(); boolean contains(SlotRequestId slotRequestId); @Override String toString(); @VisibleForTesting Collection<MultiTaskSlot> getResolvedRootSlots();  }### Answer:
@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(), new AbstractID(), locality1); Locality locality2 = Locality.HOST_LOCAL; SlotSharingManager.SingleTaskSlot singleTaskSlot2 = rootSlot.allocateSingleTaskSlot( new SlotRequestId(), 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(), new AbstractID(), locality3); CompletableFuture<LogicalSlot> logicalSlotFuture3 = singleTaskSlot3.getLogicalSlotFuture(); assertTrue(logicalSlotFuture3.isDone()); LogicalSlot logicalSlot3 = logicalSlotFuture3.get(); assertEquals(locality3, logicalSlot3.getLocality()); assertEquals(slotContext.getAllocationId(), logicalSlot3.getAllocationId()); }