### Question:
SlotPoolImpl implements SlotPool { @Override public Optional<ResourceID> failAllocation(final AllocationID allocationID, final Exception cause) { componentMainThreadExecutor.assertRunningInMainThread(); final PendingRequest pendingRequest = pendingRequests.removeKeyB(allocationID); if (pendingRequest != null) { failPendingRequest(pendingRequest, cause); return Optional.empty(); } else { return tryFailingAllocatedSlot(allocationID, cause); } } @VisibleForTesting  SlotPoolImpl(JobID jobId);  SlotPoolImpl(
			JobID jobId,
			Clock clock,
			Time rpcTimeout,
			Time idleSlotTimeout); 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); @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);  }### Answer:
@Test public void testFailingAllocationFailsPendingSlotRequests() throws Exception { try (SlotPoolImpl slotPool = new SlotPoolImpl(jobId)) { final CompletableFuture<AllocationID> allocationIdFuture = new CompletableFuture<>(); resourceManagerGateway.setRequestSlotConsumer(slotRequest -> allocationIdFuture.complete(slotRequest.getAllocationId())); setupSlotPool(slotPool, resourceManagerGateway, mainThreadExecutor); Scheduler scheduler = setupScheduler(slotPool, mainThreadExecutor); final CompletableFuture<LogicalSlot> slotFuture = allocateSlot(scheduler, new SlotRequestId()); final AllocationID allocationId = allocationIdFuture.get(); assertThat(slotFuture.isDone(), is(false)); final FlinkException cause = new FlinkException("Fail pending slot request failure."); final Optional<ResourceID> responseFuture = slotPool.failAllocation(allocationId, cause); assertThat(responseFuture.isPresent(), is(false)); try { slotFuture.get(); fail("Expected a slot allocation failure."); } catch (ExecutionException ee) { assertThat(ExceptionUtils.stripExecutionException(ee), equalTo(cause)); } } }