/*
 * Licensed to the Apache Software Foundation (ASF) under one
 * or more contributor license agreements.  See the NOTICE file
 * distributed with this work for additional information
 * regarding copyright ownership.  The ASF licenses this file
 * to you under the Apache License, Version 2.0 (the
 * "License"); you may not use this file except in compliance
 * with the License.  You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

package org.apache.flink.runtime.scheduler.strategy;

import org.apache.flink.runtime.jobgraph.JobVertexID;
import org.apache.flink.runtime.scheduler.ExecutionVertexDeploymentOption;
import org.apache.flink.util.TestLogger;

import org.junit.Before;
import org.junit.Test;

import java.util.Arrays;
import java.util.Collection;
import java.util.HashSet;
import java.util.Set;
import java.util.stream.Collectors;

import static org.hamcrest.Matchers.containsInAnyOrder;
import static org.hamcrest.Matchers.hasItem;
import static org.hamcrest.Matchers.hasSize;
import static org.junit.Assert.assertThat;

/**
 * Tests for {@link EagerSchedulingStrategy}.
 */
public class EagerSchedulingStrategyTest extends TestLogger {

	private TestingSchedulerOperations testingSchedulerOperations;

	private TestingSchedulingTopology testingSchedulingTopology;

	private EagerSchedulingStrategy schedulingStrategy;

	@Before
	public void setUp() {
		testingSchedulerOperations = new TestingSchedulerOperations();
		testingSchedulingTopology = new TestingSchedulingTopology();
		schedulingStrategy = new EagerSchedulingStrategy(
				testingSchedulerOperations,
				testingSchedulingTopology);
	}

	/**
	 * Tests that when start scheduling eager scheduling strategy will start all vertices in scheduling topology.
	 */
	@Test
	public void testStartScheduling() {
		JobVertexID jobVertexID = new JobVertexID();
		testingSchedulingTopology.addSchedulingExecutionVertex(new TestingSchedulingExecutionVertex(jobVertexID, 0));
		testingSchedulingTopology.addSchedulingExecutionVertex(new TestingSchedulingExecutionVertex(jobVertexID, 1));
		testingSchedulingTopology.addSchedulingExecutionVertex(new TestingSchedulingExecutionVertex(jobVertexID, 2));
		testingSchedulingTopology.addSchedulingExecutionVertex(new TestingSchedulingExecutionVertex(jobVertexID, 3));
		testingSchedulingTopology.addSchedulingExecutionVertex(new TestingSchedulingExecutionVertex(jobVertexID, 4));

		schedulingStrategy.startScheduling();

		assertThat(testingSchedulerOperations.getScheduledVertices(), hasSize(1));

		Collection<ExecutionVertexDeploymentOption> scheduledVertices = testingSchedulerOperations.getScheduledVertices().get(0);
		assertThat(scheduledVertices, hasSize(5));
		Collection<ExecutionVertexID> vertices = getExecutionVertexIdsFromDeployOptions(scheduledVertices);
		for (SchedulingExecutionVertex schedulingExecutionVertex : testingSchedulingTopology.getVertices()) {
			assertThat(vertices, hasItem(schedulingExecutionVertex.getId()));
		}
	}

	/**
	 * Tests that eager scheduling strategy will restart all vertices needing restarted at same time.
	 */
	@Test
	public void testRestartTasks() {
		JobVertexID jobVertexID = new JobVertexID();
		testingSchedulingTopology.addSchedulingExecutionVertex(new TestingSchedulingExecutionVertex(jobVertexID, 0));
		testingSchedulingTopology.addSchedulingExecutionVertex(new TestingSchedulingExecutionVertex(jobVertexID, 1));
		testingSchedulingTopology.addSchedulingExecutionVertex(new TestingSchedulingExecutionVertex(jobVertexID, 2));
		testingSchedulingTopology.addSchedulingExecutionVertex(new TestingSchedulingExecutionVertex(jobVertexID, 3));
		testingSchedulingTopology.addSchedulingExecutionVertex(new TestingSchedulingExecutionVertex(jobVertexID, 4));

		Set<ExecutionVertexID> toBeRestartedVertices1 = new HashSet<>(Arrays.asList(
				new ExecutionVertexID(jobVertexID, 0),
				new ExecutionVertexID(jobVertexID, 4)));
		schedulingStrategy.restartTasks(toBeRestartedVertices1);

		Set<ExecutionVertexID> toBeRestartedVertices2 = new HashSet<>(Arrays.asList(
				new ExecutionVertexID(jobVertexID, 1),
				new ExecutionVertexID(jobVertexID, 2),
				new ExecutionVertexID(jobVertexID, 3)));
		schedulingStrategy.restartTasks(toBeRestartedVertices2);

		assertThat(testingSchedulerOperations.getScheduledVertices(), hasSize(2));

		Collection<ExecutionVertexDeploymentOption> scheduledVertices1 = testingSchedulerOperations.getScheduledVertices().get(0);
		assertThat(getExecutionVertexIdsFromDeployOptions(scheduledVertices1), containsInAnyOrder(toBeRestartedVertices1.toArray()));

		Collection<ExecutionVertexDeploymentOption> scheduledVertices2 = testingSchedulerOperations.getScheduledVertices().get(1);
		assertThat(getExecutionVertexIdsFromDeployOptions(scheduledVertices2), containsInAnyOrder(toBeRestartedVertices2.toArray()));
	}

	private static Collection<ExecutionVertexID> getExecutionVertexIdsFromDeployOptions(
			Collection<ExecutionVertexDeploymentOption> deploymentOptions) {
		return deploymentOptions.stream()
				.map(ExecutionVertexDeploymentOption::getExecutionVertexId)
				.collect(Collectors.toList());
	}
}

@@ -3002,53 +3002,8 @@
 0);%0A
-%09%09assertThat(scheduledVertices, hasSize(5));%0A
 %09%09Co
@@ -3030,23 +3030,33 @@
 rtexID%3E 
-vertice
+scheduledVertexID
 s = getE
@@ -3110,16 +3110,62 @@
 tices);%0A
+%09%09assertThat(scheduledVertexIDs, hasSize(5));%0A
 %09%09for (S
@@ -3274,23 +3274,33 @@
 ertThat(
-vertice
+scheduledVertexID
 s, hasIt
@@ -4130,37 +4130,33 @@
 rtexID%3E 
-toBeRestartedVertices
+verticesToRestart
 1 = new 
@@ -4302,37 +4302,33 @@
 rtTasks(
-toBeRestartedVertices
+verticesToRestart
 1);%0A%0A%09%09S
@@ -4349,37 +4349,33 @@
 rtexID%3E 
-toBeRestartedVertices
+verticesToRestart
 2 = new 
@@ -4564,37 +4564,33 @@
 rtTasks(
-toBeRestartedVertices
+verticesToRestart
 2);%0A%0A%09%09a
@@ -4881,37 +4881,33 @@
 nyOrder(
-toBeRestartedVertices
+verticesToRestart
 1.toArra
@@ -5135,29 +5135,25 @@
 der(
-toBeRestartedVertices
+verticesToRestart
 2.to

Rename variables in EagerSchedulingStrategyTest