test_class.file: flink-runtime/src/test/java/org/apache/flink/runtime/taskmanager/TaskInputSplitProviderTest.java
test_class.identifier: TaskInputSplitProviderTest
test_case.body: @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
  		);
  
  		// The jobManager will return a
  		InputSplit nextInputSplit = provider.getNextInputSplit();
  
  		assertTrue(nextInputSplit == null);
  	}
test_case.class_method_signature: TaskInputSplitProviderTest.testRequestNextInputSplitWithInvalidExecutionID()
test_case.full_signature: @Test public void testRequestNextInputSplitWithInvalidExecutionID()
test_case.identifier: testRequestNextInputSplitWithInvalidExecutionID
test_case.invocations: getClassLoader getClass getNextInputSplit assertTrue
test_case.modifiers: @Test public
test_case.parameters: ()
test_case.return: void
test_case.signature: void testRequestNextInputSplitWithInvalidExecutionID()
focal_class.declarator: jobManager
focal_class.modifier: private final
focal_class.original_string: private final ActorGateway jobManager;
focal_class.type: ActorGateway
focal_class.var_name: jobManager
focal_class.declarator: jobId
focal_class.modifier: private final
focal_class.original_string: private final JobID jobId;
focal_class.type: JobID
focal_class.var_name: jobId
focal_class.declarator: vertexId
focal_class.modifier: private final
focal_class.original_string: private final JobVertexID vertexId;
focal_class.type: JobVertexID
focal_class.var_name: vertexId
focal_class.declarator: executionID
focal_class.modifier: private final
focal_class.original_string: private final ExecutionAttemptID executionID;
focal_class.type: ExecutionAttemptID
focal_class.var_name: executionID
focal_class.declarator: usercodeClassLoader
focal_class.modifier: private final
focal_class.original_string: private final ClassLoader usercodeClassLoader;
focal_class.type: ClassLoader
focal_class.var_name: usercodeClassLoader
focal_class.declarator: timeout
focal_class.modifier: private final
focal_class.original_string: private final FiniteDuration timeout;
focal_class.type: FiniteDuration
focal_class.var_name: timeout
focal_class.file: flink-runtime/src/main/java/org/apache/flink/runtime/taskmanager/TaskInputSplitProvider.java
focal_class.identifier: TaskInputSplitProvider
focal_class.interfaces: implements InputSplitProvider
focal_class.class_method_signature: TaskInputSplitProvider.TaskInputSplitProvider(
  			ActorGateway jobManager,
  			JobID jobId,
  			JobVertexID vertexId,
  			ExecutionAttemptID executionID,
  			ClassLoader userCodeClassLoader,
  			FiniteDuration timeout)
focal_class.full_signature: public  TaskInputSplitProvider(
  			ActorGateway jobManager,
  			JobID jobId,
  			JobVertexID vertexId,
  			ExecutionAttemptID executionID,
  			ClassLoader userCodeClassLoader,
  			FiniteDuration timeout)
focal_class.identifier: TaskInputSplitProvider
focal_class.modifiers: public
focal_class.parameters: (
  			ActorGateway jobManager,
  			JobID jobId,
  			JobVertexID vertexId,
  			ExecutionAttemptID executionID,
  			ClassLoader userCodeClassLoader,
  			FiniteDuration timeout)
focal_class.signature:  TaskInputSplitProvider(
  			ActorGateway jobManager,
  			JobID jobId,
  			JobVertexID vertexId,
  			ExecutionAttemptID executionID,
  			ClassLoader userCodeClassLoader,
  			FiniteDuration timeout)
focal_class.class_method_signature: TaskInputSplitProvider.getNextInputSplit()
focal_class.full_signature: @Override public InputSplit getNextInputSplit()
focal_class.identifier: getNextInputSplit
focal_class.modifiers: @Override public
focal_class.parameters: ()
focal_class.return: InputSplit
focal_class.signature: InputSplit getNextInputSplit()
focal_method.body: @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);
  		}
  	}
focal_method.class_method_signature: TaskInputSplitProvider.getNextInputSplit()
focal_method.full_signature: @Override public InputSplit getNextInputSplit()
focal_method.identifier: getNextInputSplit
focal_method.invocations: ask result getClass splitData deserializeObject
focal_method.modifiers: @Override public
focal_method.parameters: ()
focal_method.return: InputSplit
focal_method.signature: InputSplit getNextInputSplit()