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); } } TaskInputSplitProvider(
			ActorGateway jobManager,
			JobID jobId,
			JobVertexID vertexId,
			ExecutionAttemptID executionID,
			ClassLoader userCodeClassLoader,
			FiniteDuration timeout); @Override InputSplit getNextInputSplit();  }
@Test public void testRequestNextInputSplitWithInvalidExecutionID() { final JobID jobID = new JobID(); final JobVertexID vertexID = new JobVertexID(); final ExecutionAttemptID executionID = new ExecutionAttemptID(); final FiniteDuration timeout = new FiniteDuration(10, TimeUnit.SECONDS); final ActorGateway gateway = new NullInputSplitGateway(); final TaskInputSplitProvider provider = new TaskInputSplitProvider( gateway, jobID, vertexID, executionID, getClass().getClassLoader(), timeout ); InputSplit nextInputSplit = provider.getNextInputSplit(); assertTrue(nextInputSplit == null); }