### Question:
SlotSharingManager { @Nullable TaskSlot getTaskSlot(SlotRequestId slotRequestId) { return allTaskSlots.get(slotRequestId); } 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();  }### Answer:
@Test public void testSlotAllocatedWithEnoughResource() { SlotSharingResourceTestContext context = createResourceTestContext(new ResourceProfile(16.0, 1600)); for (SlotSharingManager.SingleTaskSlot singleTaskSlot : context.singleTaskSlotsInOrder) { assertThat(singleTaskSlot.getLogicalSlotFuture().isDone(), is(true)); assertThat(singleTaskSlot.getLogicalSlotFuture().isCompletedExceptionally(), is(false)); } assertThat(context.slotSharingManager.getTaskSlot(context.coLocationTaskSlot.getSlotRequestId()), notNullValue()); }
@Test public void testSlotOverAllocatedAndSingleSlotReleased() { SlotSharingResourceTestContext context = createResourceTestContext(new ResourceProfile(7.0, 700)); for (int i = 0; i < context.singleTaskSlotsInOrder.size(); ++i) { SlotSharingManager.SingleTaskSlot singleTaskSlot = context.singleTaskSlotsInOrder.get(i); assertThat(singleTaskSlot.getLogicalSlotFuture().isDone(), is(true)); if (i != 3) { assertThat(singleTaskSlot.getLogicalSlotFuture().isCompletedExceptionally(), is(false)); } else { assertThat(singleTaskSlot.getLogicalSlotFuture().isCompletedExceptionally(), is(true)); singleTaskSlot.getLogicalSlotFuture().whenComplete((LogicalSlot ignored, Throwable throwable) -> { assertThat(throwable instanceof SharedSlotOversubscribedException, is(true)); assertThat(((SharedSlotOversubscribedException) throwable).canRetry(), is(true)); }); } } assertThat(context.slotSharingManager.getTaskSlot(context.coLocationTaskSlot.getSlotRequestId()), notNullValue()); }
@Test public void testSlotOverAllocatedAndMultiTaskSlotReleased() { SlotSharingResourceTestContext context = createResourceTestContext(new ResourceProfile(3.0, 300)); for (int i = 0; i < context.singleTaskSlotsInOrder.size(); ++i) { SlotSharingManager.SingleTaskSlot singleTaskSlot = context.singleTaskSlotsInOrder.get(i); assertThat(singleTaskSlot.getLogicalSlotFuture().isDone(), is(true)); if (i == 2) { assertThat(singleTaskSlot.getLogicalSlotFuture().isCompletedExceptionally(), is(false)); } else { assertThat(singleTaskSlot.getLogicalSlotFuture().isCompletedExceptionally(), is(true)); singleTaskSlot.getLogicalSlotFuture().whenComplete((LogicalSlot ignored, Throwable throwable) -> { assertThat(throwable instanceof SharedSlotOversubscribedException, is(true)); assertThat(((SharedSlotOversubscribedException) throwable).canRetry(), is(true)); }); } } assertThat(context.slotSharingManager.getTaskSlot(context.coLocationTaskSlot.getSlotRequestId()), nullValue()); }
@Test public void testSlotAllocatedWithEnoughResource() { SlotSharingResourceTestContext context = createResourceTestContext(ResourceProfile.fromResources(16.0, 1600)); for (SlotSharingManager.SingleTaskSlot singleTaskSlot : context.singleTaskSlotsInOrder) { assertThat(singleTaskSlot.getLogicalSlotFuture().isDone(), is(true)); assertThat(singleTaskSlot.getLogicalSlotFuture().isCompletedExceptionally(), is(false)); } assertThat(context.slotSharingManager.getTaskSlot(context.coLocationTaskSlot.getSlotRequestId()), notNullValue()); }