@Test public void testStartScheduling() { final TestingSchedulingTopology testingSchedulingTopology = new TestingSchedulingTopology(); final List<TestingSchedulingExecutionVertex> producers = testingSchedulingTopology.addExecutionVertices().finish(); final List<TestingSchedulingExecutionVertex> consumers = testingSchedulingTopology.addExecutionVertices().finish(); testingSchedulingTopology.connectAllToAll(producers, consumers).finish(); startScheduling(testingSchedulingTopology); assertThat(testingSchedulerOperation, hasScheduledVertices(producers)); }
@Override public void startScheduling() { final DeploymentOption updateOption = new DeploymentOption(true); final DeploymentOption nonUpdateOption = new DeploymentOption(false); for (SchedulingExecutionVertex schedulingVertex : schedulingTopology.getVertices()) { DeploymentOption option = nonUpdateOption; for (SchedulingResultPartition srp : schedulingVertex.getProducedResultPartitions()) { if (srp.getPartitionType().isPipelined()) { option = updateOption; } inputConstraintChecker.addSchedulingResultPartition(srp); } deploymentOptions.put(schedulingVertex.getId(), option); } allocateSlotsAndDeployExecutionVertexIds(getAllVerticesFromTopology()); }
LazyFromSourcesSchedulingStrategy implements SchedulingStrategy { @Override public void startScheduling() { final DeploymentOption updateOption = new DeploymentOption(true); final DeploymentOption nonUpdateOption = new DeploymentOption(false); for (SchedulingExecutionVertex schedulingVertex : schedulingTopology.getVertices()) { DeploymentOption option = nonUpdateOption; for (SchedulingResultPartition srp : schedulingVertex.getProducedResultPartitions()) { if (srp.getPartitionType().isPipelined()) { option = updateOption; } inputConstraintChecker.addSchedulingResultPartition(srp); } deploymentOptions.put(schedulingVertex.getId(), option); } allocateSlotsAndDeployExecutionVertexIds(getAllVerticesFromTopology()); } }
LazyFromSourcesSchedulingStrategy implements SchedulingStrategy { @Override public void startScheduling() { final DeploymentOption updateOption = new DeploymentOption(true); final DeploymentOption nonUpdateOption = new DeploymentOption(false); for (SchedulingExecutionVertex schedulingVertex : schedulingTopology.getVertices()) { DeploymentOption option = nonUpdateOption; for (SchedulingResultPartition srp : schedulingVertex.getProducedResultPartitions()) { if (srp.getPartitionType().isPipelined()) { option = updateOption; } inputConstraintChecker.addSchedulingResultPartition(srp); } deploymentOptions.put(schedulingVertex.getId(), option); } allocateSlotsAndDeployExecutionVertexIds(getAllVerticesFromTopology()); } LazyFromSourcesSchedulingStrategy(
			SchedulerOperations schedulerOperations,
			SchedulingTopology schedulingTopology); }
LazyFromSourcesSchedulingStrategy implements SchedulingStrategy { @Override public void startScheduling() { final DeploymentOption updateOption = new DeploymentOption(true); final DeploymentOption nonUpdateOption = new DeploymentOption(false); for (SchedulingExecutionVertex schedulingVertex : schedulingTopology.getVertices()) { DeploymentOption option = nonUpdateOption; for (SchedulingResultPartition srp : schedulingVertex.getProducedResultPartitions()) { if (srp.getPartitionType().isPipelined()) { option = updateOption; } inputConstraintChecker.addSchedulingResultPartition(srp); } deploymentOptions.put(schedulingVertex.getId(), option); } allocateSlotsAndDeployExecutionVertexIds(getAllVerticesFromTopology()); } LazyFromSourcesSchedulingStrategy(
			SchedulerOperations schedulerOperations,
			SchedulingTopology schedulingTopology); @Override void startScheduling(); @Override void restartTasks(Set<ExecutionVertexID> verticesToRestart); @Override void onExecutionStateChange(ExecutionVertexID executionVertexId, ExecutionState executionState); @Override void onPartitionConsumable(ExecutionVertexID executionVertexId, ResultPartitionID resultPartitionId); }
LazyFromSourcesSchedulingStrategy implements SchedulingStrategy { @Override public void startScheduling() { final DeploymentOption updateOption = new DeploymentOption(true); final DeploymentOption nonUpdateOption = new DeploymentOption(false); for (SchedulingExecutionVertex schedulingVertex : schedulingTopology.getVertices()) { DeploymentOption option = nonUpdateOption; for (SchedulingResultPartition srp : schedulingVertex.getProducedResultPartitions()) { if (srp.getPartitionType().isPipelined()) { option = updateOption; } inputConstraintChecker.addSchedulingResultPartition(srp); } deploymentOptions.put(schedulingVertex.getId(), option); } allocateSlotsAndDeployExecutionVertexIds(getAllVerticesFromTopology()); } LazyFromSourcesSchedulingStrategy(
			SchedulerOperations schedulerOperations,
			SchedulingTopology schedulingTopology); @Override void startScheduling(); @Override void restartTasks(Set<ExecutionVertexID> verticesToRestart); @Override void onExecutionStateChange(ExecutionVertexID executionVertexId, ExecutionState executionState); @Override void onPartitionConsumable(ExecutionVertexID executionVertexId, ResultPartitionID resultPartitionId);  }