class SlotPool extends RpcEndpoint implements SlotPoolGateway {

	@Override
	public CompletableFuture<Boolean> offerSlot(final AllocatedSlot slot) {
		validateRunsInMainThread();

		// check if this TaskManager is valid
		final ResourceID resourceID = slot.getTaskManagerId();
		final AllocationID allocationID = slot.getSlotAllocationId();

		if (!registeredTaskManagers.contains(resourceID)) {
			LOG.debug("Received outdated slot offering [{}] from unregistered TaskManager: {}",
					slot.getSlotAllocationId(), slot);
			return CompletableFuture.completedFuture(false);
		}

		// check whether we have already using this slot
		if (allocatedSlots.contains(allocationID) || availableSlots.contains(allocationID)) {
			LOG.debug("Received repeated offer for slot [{}]. Ignoring.", allocationID);

			// return true here so that the sender will get a positive acknowledgement to the retry
			// and mark the offering as a success
			return CompletableFuture.completedFuture(true);
		}

		// check whether we have request waiting for this slot
		PendingRequest pendingRequest = pendingRequests.removeKeyB(allocationID);
		if (pendingRequest != null) {
			// we were waiting for this!
			SimpleSlot resultSlot = createSimpleSlot(slot, Locality.UNKNOWN);
			pendingRequest.getFuture().complete(resultSlot);
			allocatedSlots.add(pendingRequest.getSlotRequestId(), resultSlot);
		}
		else {
			// we were actually not waiting for this:
			//   - could be that this request had been fulfilled
			//   - we are receiving the slots from TaskManagers after becoming leaders
			availableSlots.add(slot, clock.relativeTimeMillis());
		}

		// we accepted the request in any case. slot will be released after it idled for
		// too long and timed out
		return CompletableFuture.completedFuture(true);
	}

	public  SlotPool(RpcService rpcService, JobID jobId);
	public  SlotPool(
			RpcService rpcService,
			JobID jobId,
			Clock clock,
			Time slotRequestTimeout,
			Time resourceManagerAllocationTimeout,
			Time resourceManagerRequestTimeout);

}