### Question:
Scheduler implements InstanceListener, SlotAvailabilityListener, SlotProvider { @Override public CompletableFuture<LogicalSlot> allocateSlot( SlotRequestId slotRequestId, ScheduledUnit task, boolean allowQueued, SlotProfile slotProfile, Time allocationTimeout) { try { final Object ret = scheduleTask(task, allowQueued, slotProfile.getPreferredLocations()); if (ret instanceof SimpleSlot) { return CompletableFuture.completedFuture((SimpleSlot) ret); } else if (ret instanceof CompletableFuture) { @SuppressWarnings("unchecked") CompletableFuture<LogicalSlot> typed = (CompletableFuture<LogicalSlot>) ret; return FutureUtils.orTimeout(typed, allocationTimeout.toMilliseconds(), TimeUnit.MILLISECONDS); } else { throw new RuntimeException(); } } catch (NoResourceAvailableException e) { return FutureUtils.completedExceptionally(e); } } Scheduler(Executor executor); void shutdown(); @Override CompletableFuture<LogicalSlot> allocateSlot(
			SlotRequestId slotRequestId,
			ScheduledUnit task,
			boolean allowQueued,
			SlotProfile slotProfile,
			Time allocationTimeout); @Override CompletableFuture<Acknowledge> cancelSlotRequest(SlotRequestId slotRequestId, @Nullable SlotSharingGroupId slotSharingGroupId, Throwable cause); @Override void newSlotAvailable(final Instance instance); @Override void newInstanceAvailable(Instance instance); @Override void instanceDied(Instance instance); int getNumberOfAvailableSlots(); int getTotalNumberOfSlots(); int getNumberOfAvailableInstances(); int getNumberOfInstancesWithAvailableSlots(); Map<String, List<Instance>> getInstancesByHost(); int getNumberOfUnconstrainedAssignments(); int getNumberOfLocalizedAssignments(); int getNumberOfNonLocalizedAssignments(); @VisibleForTesting @Nullable Instance getInstance(ResourceID resourceId);  }### Answer:
@Test public void testSlotAllocationTimeout() throws Exception { final Scheduler scheduler = new Scheduler(TestingUtils.defaultExecutor()); final ExecutionGraph executionGraph = ExecutionGraphTestUtils.createSimpleTestGraph(); final Map<ExecutionAttemptID, Execution> registeredExecutions = executionGraph.getRegisteredExecutions(); assertThat(registeredExecutions.values(), Matchers.not(Matchers.empty())); final Execution execution = registeredExecutions.values().iterator().next(); final CompletableFuture<LogicalSlot> slotFuture = scheduler.allocateSlot( new ScheduledUnit( execution), true, SlotProfile.noRequirements(), Time.milliseconds(1L)); try { slotFuture.get(); } catch (ExecutionException ee) { assertThat(ExceptionUtils.stripExecutionException(ee), Matchers.instanceOf(TimeoutException.class)); } }