class TaskInputSplitProvider implements InputSplitProvider {

	@Override
	public InputSplit getNextInputSplit(ClassLoader userCodeClassLoader) throws InputSplitProviderException {
		Preconditions.checkNotNull(userCodeClassLoader);

		final Future<Object> response = jobManager.ask(
			new JobManagerMessages.RequestNextInputSplit(jobID, vertexID, executionID),
			timeout);

		final Object result;

		try {
			result = Await.result(response, timeout);
		} catch (Exception e) {
			throw new InputSplitProviderException("Did not receive next input split from JobManager.", e);
		}

		if(result instanceof JobManagerMessages.NextInputSplit){
			final JobManagerMessages.NextInputSplit nextInputSplit =
				(JobManagerMessages.NextInputSplit) result;

			byte[] serializedData = nextInputSplit.splitData();

			if(serializedData == null) {
				return null;
			} else {
				final Object deserialized;

				try {
					deserialized = InstantiationUtil.deserializeObject(serializedData,
						userCodeClassLoader);
				} catch (Exception e) {
					throw new InputSplitProviderException("Could not deserialize the serialized input split.", e);
				}

				return (InputSplit) deserialized;
			}
		} else {
			throw new InputSplitProviderException("RequestNextInputSplit requires a response of type " +
				"NextInputSplit. Instead response is of type " + result.getClass() + '.');
		}

	}

	public  TaskInputSplitProvider(
		ActorGateway jobManager,
		JobID jobID,
		JobVertexID vertexID,
		ExecutionAttemptID executionID,
		FiniteDuration timeout);

}