class SlotPoolImpl implements SlotPool {

	@Nonnull
	@Override
	public CompletableFuture<PhysicalSlot> requestNewAllocatedSlot(
			@Nonnull SlotRequestId slotRequestId,
			@Nonnull ResourceProfile resourceProfile,
			@Nullable Time timeout) {

		componentMainThreadExecutor.assertRunningInMainThread();

		final PendingRequest pendingRequest = PendingRequest.createStreamingRequest(slotRequestId, resourceProfile);

		if (timeout != null) {
			// register request timeout
			FutureUtils
				.orTimeout(
					pendingRequest.getAllocatedSlotFuture(),
					timeout.toMilliseconds(),
					TimeUnit.MILLISECONDS,
					componentMainThreadExecutor)
				.whenComplete(
					(AllocatedSlot ignored, Throwable throwable) -> {
						if (throwable instanceof TimeoutException) {
							timeoutPendingSlotRequest(slotRequestId);
						}
					});
		}

		return requestNewAllocatedSlotInternal(pendingRequest)
			.thenApply((Function.identity()));
	}

	public  SlotPoolImpl(
			JobID jobId,
			Clock clock,
			Time rpcTimeout,
			Time idleSlotTimeout,
			Time batchSlotTimeout);

}