@Test public void testRequestNextInputSplit() throws Exception { final List<TestingInputSplit> expectedInputSplits = Arrays.asList( new TestingInputSplit(1), new TestingInputSplit(42), new TestingInputSplit(1337)); InputSplitSource<TestingInputSplit> inputSplitSource = new TestingInputSplitSource(expectedInputSplits); JobVertex source = new JobVertex("vertex1"); source.setParallelism(1); source.setInputSplitSource(inputSplitSource); source.setInvokableClass(AbstractInvokable.class); final JobGraph testJobGraph = new JobGraph(source); testJobGraph.setAllowQueuedScheduling(true); configuration.setLong(ConfigConstants.RESTART_STRATEGY_FIXED_DELAY_ATTEMPTS, 1); configuration.setString(ConfigConstants.RESTART_STRATEGY_FIXED_DELAY_DELAY, "0 s"); final JobManagerSharedServices jobManagerSharedServices = new TestingJobManagerSharedServicesBuilder() .setRestartStrategyFactory(RestartStrategyFactory.createRestartStrategyFactory(configuration)) .build(); final JobMaster jobMaster = createJobMaster( configuration, testJobGraph, haServices, jobManagerSharedServices); CompletableFuture<Acknowledge> startFuture = jobMaster.start(jobMasterId); try { startFuture.get(testingTimeout.toMilliseconds(), TimeUnit.MILLISECONDS); final JobMasterGateway jobMasterGateway = jobMaster.getSelfGateway(JobMasterGateway.class); ExecutionGraph eg = jobMaster.getExecutionGraph(); ExecutionVertex ev = eg.getAllExecutionVertices().iterator().next(); final SupplierWithException<SerializedInputSplit, Exception> inputSplitSupplier = () -> jobMasterGateway.requestNextInputSplit( source.getID(), ev.getCurrentExecutionAttempt().getAttemptId()).get(); List<InputSplit> actualInputSplits = getInputSplits( expectedInputSplits.size(), inputSplitSupplier); final Matcher<Iterable<? extends InputSplit>> expectedInputSplitsMatcher = containsInAnyOrder(expectedInputSplits.toArray(EMPTY_TESTING_INPUT_SPLITS)); assertThat(actualInputSplits, expectedInputSplitsMatcher); final long maxWaitMillis = 2000L; ExecutionGraphTestUtils.waitUntilExecutionVertexState(ev, ExecutionState.SCHEDULED, maxWaitMillis); CompletableFuture.runAsync(() -> eg.failGlobal(new Exception("Testing exception")), eg.getJobMasterMainThreadExecutor()).get(); ExecutionGraphTestUtils.waitUntilExecutionVertexState(ev, ExecutionState.SCHEDULED, maxWaitMillis); actualInputSplits = getInputSplits( expectedInputSplits.size(), inputSplitSupplier); assertThat(actualInputSplits, expectedInputSplitsMatcher); } finally { RpcUtils.terminateRpcEndpoint(jobMaster, testingTimeout); } }
@Override public CompletableFuture<SerializedInputSplit> requestNextInputSplit( final JobVertexID vertexID, final ExecutionAttemptID executionAttempt) { final Execution execution = executionGraph.getRegisteredExecutions().get(executionAttempt); if (execution == null) { if (log.isDebugEnabled()) { log.debug("Can not find Execution for attempt {}.", executionAttempt); } return FutureUtils.completedExceptionally(new Exception("Can not find Execution for attempt " + executionAttempt)); } final ExecutionJobVertex vertex = executionGraph.getJobVertex(vertexID); if (vertex == null) { log.error("Cannot find execution vertex for vertex ID {}.", vertexID); return FutureUtils.completedExceptionally(new Exception("Cannot find execution vertex for vertex ID " + vertexID)); } final InputSplitAssigner splitAssigner = vertex.getSplitAssigner(); if (splitAssigner == null) { log.error("No InputSplitAssigner for vertex ID {}.", vertexID); return FutureUtils.completedExceptionally(new Exception("No InputSplitAssigner for vertex ID " + vertexID)); } final LogicalSlot slot = execution.getAssignedResource(); final int taskId = execution.getVertex().getParallelSubtaskIndex(); final String host = slot != null ? slot.getTaskManagerLocation().getHostname() : null; final InputSplit nextInputSplit = splitAssigner.getNextInputSplit(host, taskId); if (log.isDebugEnabled()) { log.debug("Send next input split {}.", nextInputSplit); } try { final byte[] serializedInputSplit = InstantiationUtil.serializeObject(nextInputSplit); return CompletableFuture.completedFuture(new SerializedInputSplit(serializedInputSplit)); } catch (Exception ex) { log.error("Could not serialize the next input split of class {}.", nextInputSplit.getClass(), ex); IOException reason = new IOException("Could not serialize the next input split of class " + nextInputSplit.getClass() + ".", ex); vertex.fail(reason); return FutureUtils.completedExceptionally(reason); } }
JobMaster extends FencedRpcEndpoint<JobMasterId> implements JobMasterGateway, JobMasterService { @Override public CompletableFuture<SerializedInputSplit> requestNextInputSplit( final JobVertexID vertexID, final ExecutionAttemptID executionAttempt) { final Execution execution = executionGraph.getRegisteredExecutions().get(executionAttempt); if (execution == null) { if (log.isDebugEnabled()) { log.debug("Can not find Execution for attempt {}.", executionAttempt); } return FutureUtils.completedExceptionally(new Exception("Can not find Execution for attempt " + executionAttempt)); } final ExecutionJobVertex vertex = executionGraph.getJobVertex(vertexID); if (vertex == null) { log.error("Cannot find execution vertex for vertex ID {}.", vertexID); return FutureUtils.completedExceptionally(new Exception("Cannot find execution vertex for vertex ID " + vertexID)); } final InputSplitAssigner splitAssigner = vertex.getSplitAssigner(); if (splitAssigner == null) { log.error("No InputSplitAssigner for vertex ID {}.", vertexID); return FutureUtils.completedExceptionally(new Exception("No InputSplitAssigner for vertex ID " + vertexID)); } final LogicalSlot slot = execution.getAssignedResource(); final int taskId = execution.getVertex().getParallelSubtaskIndex(); final String host = slot != null ? slot.getTaskManagerLocation().getHostname() : null; final InputSplit nextInputSplit = splitAssigner.getNextInputSplit(host, taskId); if (log.isDebugEnabled()) { log.debug("Send next input split {}.", nextInputSplit); } try { final byte[] serializedInputSplit = InstantiationUtil.serializeObject(nextInputSplit); return CompletableFuture.completedFuture(new SerializedInputSplit(serializedInputSplit)); } catch (Exception ex) { log.error("Could not serialize the next input split of class {}.", nextInputSplit.getClass(), ex); IOException reason = new IOException("Could not serialize the next input split of class " + nextInputSplit.getClass() + ".", ex); vertex.fail(reason); return FutureUtils.completedExceptionally(reason); } } }
JobMaster extends FencedRpcEndpoint<JobMasterId> implements JobMasterGateway, JobMasterService { @Override public CompletableFuture<SerializedInputSplit> requestNextInputSplit( final JobVertexID vertexID, final ExecutionAttemptID executionAttempt) { final Execution execution = executionGraph.getRegisteredExecutions().get(executionAttempt); if (execution == null) { if (log.isDebugEnabled()) { log.debug("Can not find Execution for attempt {}.", executionAttempt); } return FutureUtils.completedExceptionally(new Exception("Can not find Execution for attempt " + executionAttempt)); } final ExecutionJobVertex vertex = executionGraph.getJobVertex(vertexID); if (vertex == null) { log.error("Cannot find execution vertex for vertex ID {}.", vertexID); return FutureUtils.completedExceptionally(new Exception("Cannot find execution vertex for vertex ID " + vertexID)); } final InputSplitAssigner splitAssigner = vertex.getSplitAssigner(); if (splitAssigner == null) { log.error("No InputSplitAssigner for vertex ID {}.", vertexID); return FutureUtils.completedExceptionally(new Exception("No InputSplitAssigner for vertex ID " + vertexID)); } final LogicalSlot slot = execution.getAssignedResource(); final int taskId = execution.getVertex().getParallelSubtaskIndex(); final String host = slot != null ? slot.getTaskManagerLocation().getHostname() : null; final InputSplit nextInputSplit = splitAssigner.getNextInputSplit(host, taskId); if (log.isDebugEnabled()) { log.debug("Send next input split {}.", nextInputSplit); } try { final byte[] serializedInputSplit = InstantiationUtil.serializeObject(nextInputSplit); return CompletableFuture.completedFuture(new SerializedInputSplit(serializedInputSplit)); } catch (Exception ex) { log.error("Could not serialize the next input split of class {}.", nextInputSplit.getClass(), ex); IOException reason = new IOException("Could not serialize the next input split of class " + nextInputSplit.getClass() + ".", ex); vertex.fail(reason); return FutureUtils.completedExceptionally(reason); } } JobMaster(
			RpcService rpcService,
			JobMasterConfiguration jobMasterConfiguration,
			ResourceID resourceId,
			JobGraph jobGraph,
			HighAvailabilityServices highAvailabilityService,
			SlotPoolFactory slotPoolFactory,
			SchedulerFactory schedulerFactory,
			JobManagerSharedServices jobManagerSharedServices,
			HeartbeatServices heartbeatServices,
			JobManagerJobMetricGroupFactory jobMetricGroupFactory,
			OnCompletionActions jobCompletionActions,
			FatalErrorHandler fatalErrorHandler,
			ClassLoader userCodeLoader); }
JobMaster extends FencedRpcEndpoint<JobMasterId> implements JobMasterGateway, JobMasterService { @Override public CompletableFuture<SerializedInputSplit> requestNextInputSplit( final JobVertexID vertexID, final ExecutionAttemptID executionAttempt) { final Execution execution = executionGraph.getRegisteredExecutions().get(executionAttempt); if (execution == null) { if (log.isDebugEnabled()) { log.debug("Can not find Execution for attempt {}.", executionAttempt); } return FutureUtils.completedExceptionally(new Exception("Can not find Execution for attempt " + executionAttempt)); } final ExecutionJobVertex vertex = executionGraph.getJobVertex(vertexID); if (vertex == null) { log.error("Cannot find execution vertex for vertex ID {}.", vertexID); return FutureUtils.completedExceptionally(new Exception("Cannot find execution vertex for vertex ID " + vertexID)); } final InputSplitAssigner splitAssigner = vertex.getSplitAssigner(); if (splitAssigner == null) { log.error("No InputSplitAssigner for vertex ID {}.", vertexID); return FutureUtils.completedExceptionally(new Exception("No InputSplitAssigner for vertex ID " + vertexID)); } final LogicalSlot slot = execution.getAssignedResource(); final int taskId = execution.getVertex().getParallelSubtaskIndex(); final String host = slot != null ? slot.getTaskManagerLocation().getHostname() : null; final InputSplit nextInputSplit = splitAssigner.getNextInputSplit(host, taskId); if (log.isDebugEnabled()) { log.debug("Send next input split {}.", nextInputSplit); } try { final byte[] serializedInputSplit = InstantiationUtil.serializeObject(nextInputSplit); return CompletableFuture.completedFuture(new SerializedInputSplit(serializedInputSplit)); } catch (Exception ex) { log.error("Could not serialize the next input split of class {}.", nextInputSplit.getClass(), ex); IOException reason = new IOException("Could not serialize the next input split of class " + nextInputSplit.getClass() + ".", ex); vertex.fail(reason); return FutureUtils.completedExceptionally(reason); } } JobMaster(
			RpcService rpcService,
			JobMasterConfiguration jobMasterConfiguration,
			ResourceID resourceId,
			JobGraph jobGraph,
			HighAvailabilityServices highAvailabilityService,
			SlotPoolFactory slotPoolFactory,
			SchedulerFactory schedulerFactory,
			JobManagerSharedServices jobManagerSharedServices,
			HeartbeatServices heartbeatServices,
			JobManagerJobMetricGroupFactory jobMetricGroupFactory,
			OnCompletionActions jobCompletionActions,
			FatalErrorHandler fatalErrorHandler,
			ClassLoader userCodeLoader); CompletableFuture<Acknowledge> start(final JobMasterId newJobMasterId); CompletableFuture<Acknowledge> suspend(final Exception cause); @Override CompletableFuture<Void> onStop(); @Override CompletableFuture<Acknowledge> cancel(Time timeout); @Override CompletableFuture<Acknowledge> stop(Time timeout); @Override CompletableFuture<Acknowledge> rescaleJob(
			int newParallelism,
			RescalingBehaviour rescalingBehaviour,
			Time timeout); @Override CompletableFuture<Acknowledge> rescaleOperators(
			Collection<JobVertexID> operators,
			int newParallelism,
			RescalingBehaviour rescalingBehaviour,
			Time timeout); @Override CompletableFuture<Acknowledge> updateTaskExecutionState(
			final TaskExecutionState taskExecutionState); @Override CompletableFuture<SerializedInputSplit> requestNextInputSplit(
			final JobVertexID vertexID,
			final ExecutionAttemptID executionAttempt); @Override CompletableFuture<ExecutionState> requestPartitionState(
			final IntermediateDataSetID intermediateResultId,
			final ResultPartitionID resultPartitionId); @Override CompletableFuture<Acknowledge> scheduleOrUpdateConsumers(
			final ResultPartitionID partitionID,
			final Time timeout); @Override CompletableFuture<Acknowledge> disconnectTaskManager(final ResourceID resourceID, final Exception cause); @Override void acknowledgeCheckpoint(
			final JobID jobID,
			final ExecutionAttemptID executionAttemptID,
			final long checkpointId,
			final CheckpointMetrics checkpointMetrics,
			final TaskStateSnapshot checkpointState); @Override void declineCheckpoint(DeclineCheckpoint decline); @Override CompletableFuture<KvStateLocation> requestKvStateLocation(final JobID jobId, final String registrationName); @Override CompletableFuture<Acknowledge> notifyKvStateRegistered(
			final JobID jobId,
			final JobVertexID jobVertexId,
			final KeyGroupRange keyGroupRange,
			final String registrationName,
			final KvStateID kvStateId,
			final InetSocketAddress kvStateServerAddress); @Override CompletableFuture<Acknowledge> notifyKvStateUnregistered(
			JobID jobId,
			JobVertexID jobVertexId,
			KeyGroupRange keyGroupRange,
			String registrationName); @Override CompletableFuture<Collection<SlotOffer>> offerSlots(
			final ResourceID taskManagerId,
			final Collection<SlotOffer> slots,
			final Time timeout); @Override void failSlot(
			final ResourceID taskManagerId,
			final AllocationID allocationId,
			final Exception cause); @Override CompletableFuture<RegistrationResponse> registerTaskManager(
			final String taskManagerRpcAddress,
			final TaskManagerLocation taskManagerLocation,
			final Time timeout); @Override void disconnectResourceManager(
			final ResourceManagerId resourceManagerId,
			final Exception cause); @Override void heartbeatFromTaskManager(final ResourceID resourceID, AccumulatorReport accumulatorReport); @Override void heartbeatFromResourceManager(final ResourceID resourceID); @Override CompletableFuture<JobDetails> requestJobDetails(Time timeout); @Override CompletableFuture<JobStatus> requestJobStatus(Time timeout); @Override CompletableFuture<ArchivedExecutionGraph> requestJob(Time timeout); @Override CompletableFuture<String> triggerSavepoint(
			@Nullable final String targetDirectory,
			final boolean cancelJob,
			final Time timeout); @Override CompletableFuture<OperatorBackPressureStatsResponse> requestOperatorBackPressureStats(final JobVertexID jobVertexId); @Override void notifyAllocationFailure(AllocationID allocationID, Exception cause); @Override CompletableFuture<Object> updateGlobalAggregate(String aggregateName, Object aggregand, byte[] serializedAggregateFunction); @Override JobMasterGateway getGateway(); }
JobMaster extends FencedRpcEndpoint<JobMasterId> implements JobMasterGateway, JobMasterService { @Override public CompletableFuture<SerializedInputSplit> requestNextInputSplit( final JobVertexID vertexID, final ExecutionAttemptID executionAttempt) { final Execution execution = executionGraph.getRegisteredExecutions().get(executionAttempt); if (execution == null) { if (log.isDebugEnabled()) { log.debug("Can not find Execution for attempt {}.", executionAttempt); } return FutureUtils.completedExceptionally(new Exception("Can not find Execution for attempt " + executionAttempt)); } final ExecutionJobVertex vertex = executionGraph.getJobVertex(vertexID); if (vertex == null) { log.error("Cannot find execution vertex for vertex ID {}.", vertexID); return FutureUtils.completedExceptionally(new Exception("Cannot find execution vertex for vertex ID " + vertexID)); } final InputSplitAssigner splitAssigner = vertex.getSplitAssigner(); if (splitAssigner == null) { log.error("No InputSplitAssigner for vertex ID {}.", vertexID); return FutureUtils.completedExceptionally(new Exception("No InputSplitAssigner for vertex ID " + vertexID)); } final LogicalSlot slot = execution.getAssignedResource(); final int taskId = execution.getVertex().getParallelSubtaskIndex(); final String host = slot != null ? slot.getTaskManagerLocation().getHostname() : null; final InputSplit nextInputSplit = splitAssigner.getNextInputSplit(host, taskId); if (log.isDebugEnabled()) { log.debug("Send next input split {}.", nextInputSplit); } try { final byte[] serializedInputSplit = InstantiationUtil.serializeObject(nextInputSplit); return CompletableFuture.completedFuture(new SerializedInputSplit(serializedInputSplit)); } catch (Exception ex) { log.error("Could not serialize the next input split of class {}.", nextInputSplit.getClass(), ex); IOException reason = new IOException("Could not serialize the next input split of class " + nextInputSplit.getClass() + ".", ex); vertex.fail(reason); return FutureUtils.completedExceptionally(reason); } } JobMaster(
			RpcService rpcService,
			JobMasterConfiguration jobMasterConfiguration,
			ResourceID resourceId,
			JobGraph jobGraph,
			HighAvailabilityServices highAvailabilityService,
			SlotPoolFactory slotPoolFactory,
			SchedulerFactory schedulerFactory,
			JobManagerSharedServices jobManagerSharedServices,
			HeartbeatServices heartbeatServices,
			JobManagerJobMetricGroupFactory jobMetricGroupFactory,
			OnCompletionActions jobCompletionActions,
			FatalErrorHandler fatalErrorHandler,
			ClassLoader userCodeLoader); CompletableFuture<Acknowledge> start(final JobMasterId newJobMasterId); CompletableFuture<Acknowledge> suspend(final Exception cause); @Override CompletableFuture<Void> onStop(); @Override CompletableFuture<Acknowledge> cancel(Time timeout); @Override CompletableFuture<Acknowledge> stop(Time timeout); @Override CompletableFuture<Acknowledge> rescaleJob(
			int newParallelism,
			RescalingBehaviour rescalingBehaviour,
			Time timeout); @Override CompletableFuture<Acknowledge> rescaleOperators(
			Collection<JobVertexID> operators,
			int newParallelism,
			RescalingBehaviour rescalingBehaviour,
			Time timeout); @Override CompletableFuture<Acknowledge> updateTaskExecutionState(
			final TaskExecutionState taskExecutionState); @Override CompletableFuture<SerializedInputSplit> requestNextInputSplit(
			final JobVertexID vertexID,
			final ExecutionAttemptID executionAttempt); @Override CompletableFuture<ExecutionState> requestPartitionState(
			final IntermediateDataSetID intermediateResultId,
			final ResultPartitionID resultPartitionId); @Override CompletableFuture<Acknowledge> scheduleOrUpdateConsumers(
			final ResultPartitionID partitionID,
			final Time timeout); @Override CompletableFuture<Acknowledge> disconnectTaskManager(final ResourceID resourceID, final Exception cause); @Override void acknowledgeCheckpoint(
			final JobID jobID,
			final ExecutionAttemptID executionAttemptID,
			final long checkpointId,
			final CheckpointMetrics checkpointMetrics,
			final TaskStateSnapshot checkpointState); @Override void declineCheckpoint(DeclineCheckpoint decline); @Override CompletableFuture<KvStateLocation> requestKvStateLocation(final JobID jobId, final String registrationName); @Override CompletableFuture<Acknowledge> notifyKvStateRegistered(
			final JobID jobId,
			final JobVertexID jobVertexId,
			final KeyGroupRange keyGroupRange,
			final String registrationName,
			final KvStateID kvStateId,
			final InetSocketAddress kvStateServerAddress); @Override CompletableFuture<Acknowledge> notifyKvStateUnregistered(
			JobID jobId,
			JobVertexID jobVertexId,
			KeyGroupRange keyGroupRange,
			String registrationName); @Override CompletableFuture<Collection<SlotOffer>> offerSlots(
			final ResourceID taskManagerId,
			final Collection<SlotOffer> slots,
			final Time timeout); @Override void failSlot(
			final ResourceID taskManagerId,
			final AllocationID allocationId,
			final Exception cause); @Override CompletableFuture<RegistrationResponse> registerTaskManager(
			final String taskManagerRpcAddress,
			final TaskManagerLocation taskManagerLocation,
			final Time timeout); @Override void disconnectResourceManager(
			final ResourceManagerId resourceManagerId,
			final Exception cause); @Override void heartbeatFromTaskManager(final ResourceID resourceID, AccumulatorReport accumulatorReport); @Override void heartbeatFromResourceManager(final ResourceID resourceID); @Override CompletableFuture<JobDetails> requestJobDetails(Time timeout); @Override CompletableFuture<JobStatus> requestJobStatus(Time timeout); @Override CompletableFuture<ArchivedExecutionGraph> requestJob(Time timeout); @Override CompletableFuture<String> triggerSavepoint(
			@Nullable final String targetDirectory,
			final boolean cancelJob,
			final Time timeout); @Override CompletableFuture<OperatorBackPressureStatsResponse> requestOperatorBackPressureStats(final JobVertexID jobVertexId); @Override void notifyAllocationFailure(AllocationID allocationID, Exception cause); @Override CompletableFuture<Object> updateGlobalAggregate(String aggregateName, Object aggregand, byte[] serializedAggregateFunction); @Override JobMasterGateway getGateway(); static final String JOB_MANAGER_NAME; static final String ARCHIVE_NAME; }