class TaskInputSplitProvider implements InputSplitProvider {

	@Override
	public InputSplit getNextInputSplit() {
		try {
			final Future<Object> response = jobManager.ask(
					new JobManagerMessages.RequestNextInputSplit(jobId, vertexId, executionID),
					timeout);

			final Object result = Await.result(response, timeout);

			if(!(result instanceof JobManagerMessages.NextInputSplit)){
				throw new RuntimeException("RequestNextInputSplit requires a response of type " +
						"NextInputSplit. Instead response is of type " + result.getClass() + ".");
			} else {
				final JobManagerMessages.NextInputSplit nextInputSplit =
						(JobManagerMessages.NextInputSplit) result;

				byte[] serializedData = nextInputSplit.splitData();

				if(serializedData == null) {
					return null;
				} else {
					Object deserialized = InstantiationUtil.deserializeObject(serializedData,
							usercodeClassLoader);
					return (InputSplit) deserialized;
				}
			}
		} catch (Exception e) {
			throw new RuntimeException("Requesting the next InputSplit failed.", e);
		}
	}

	public  TaskInputSplitProvider(
			ActorGateway jobManager,
			JobID jobId,
			JobVertexID vertexId,
			ExecutionAttemptID executionID,
			ClassLoader userCodeClassLoader,
			FiniteDuration timeout);

}