/**
 * 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
 * <p>
 * http://www.apache.org/licenses/LICENSE-2.0
 * <p>
 * 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.storm.scheduler.resource;

import java.time.Duration;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collection;
import java.util.Collections;
import java.util.HashMap;
import java.util.HashSet;
import java.util.LinkedList;
import java.util.List;
import java.util.Map;
import java.util.Map.Entry;
import java.util.Set;
import java.util.TreeMap;
import java.util.concurrent.atomic.AtomicLong;

import org.apache.storm.Config;
import org.apache.storm.DaemonConfig;
import org.apache.storm.generated.StormTopology;
import org.apache.storm.generated.WorkerResources;
import org.apache.storm.metric.StormMetricsRegistry;
import org.apache.storm.scheduler.Cluster;
import org.apache.storm.scheduler.ExecutorDetails;
import org.apache.storm.scheduler.INimbus;
import org.apache.storm.scheduler.IScheduler;
import org.apache.storm.scheduler.SchedulerAssignment;
import org.apache.storm.scheduler.SchedulerAssignmentImpl;
import org.apache.storm.scheduler.SupervisorDetails;
import org.apache.storm.scheduler.Topologies;
import org.apache.storm.scheduler.TopologyDetails;
import org.apache.storm.scheduler.WorkerSlot;
import org.apache.storm.scheduler.resource.normalization.NormalizedResources;
import org.apache.storm.scheduler.resource.strategies.scheduling.BaseResourceAwareStrategy;
import org.apache.storm.scheduler.resource.strategies.scheduling.ConstraintSolverStrategy;
import org.apache.storm.scheduler.resource.strategies.scheduling.DefaultResourceAwareStrategy;
import org.apache.storm.scheduler.resource.strategies.scheduling.DefaultResourceAwareStrategyOld;
import org.apache.storm.scheduler.resource.strategies.scheduling.GenericResourceAwareStrategy;
import org.apache.storm.scheduler.resource.strategies.scheduling.GenericResourceAwareStrategyOld;
import org.apache.storm.scheduler.resource.strategies.scheduling.RoundRobinResourceAwareStrategy;
import org.apache.storm.scheduler.resource.normalization.ResourceMetrics;
import org.apache.storm.testing.PerformanceTest;
import org.apache.storm.testing.TestWordCounter;
import org.apache.storm.testing.TestWordSpout;
import org.apache.storm.topology.TopologyBuilder;
import org.apache.storm.utils.ConfigUtils;
import org.apache.storm.utils.DisallowedStrategyException;
import org.apache.storm.utils.ReflectionUtils;
import org.apache.storm.utils.Time;
import org.apache.storm.utils.Utils;
import org.apache.storm.validation.ConfigValidation;
import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.Test;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import static org.apache.storm.scheduler.resource.TestUtilsForResourceAwareScheduler.*;
import static org.junit.jupiter.api.Assertions.assertTrue;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertFalse;


public class TestResourceAwareScheduler {

    private static final Logger LOG = LoggerFactory.getLogger(TestResourceAwareScheduler.class);
    private static final Class[] strategyClasses = {
            DefaultResourceAwareStrategy.class,
            RoundRobinResourceAwareStrategy.class,
            GenericResourceAwareStrategy.class,
    };
    private final Config defaultTopologyConf;
    private int currentTime = 1450418597;
    private IScheduler scheduler = null;

    public TestResourceAwareScheduler() {
        defaultTopologyConf = createClusterConfig(DefaultResourceAwareStrategy.class, 10, 128, 0, null);
        defaultTopologyConf.put(Config.TOPOLOGY_WORKER_MAX_HEAP_SIZE_MB, 8192.0);
        defaultTopologyConf.put(Config.TOPOLOGY_PRIORITY, 0);
    }

    private Config createClusterConfig(Class strategyClass, double compPcore, double compOnHeap, double compOffHeap,
                                       Map<String, Map<String, Number>> pools) {
        Config config = TestUtilsForResourceAwareScheduler.createClusterConfig(compPcore, compOnHeap, compOffHeap, pools);
        config.put(Config.TOPOLOGY_SCHEDULER_STRATEGY, strategyClass.getName());
        return config;
    }

    @AfterEach
    public void cleanup() {
        if (scheduler != null) {
            scheduler.cleanup();
            scheduler = null;
        }
    }

    @Test
    public void testRASNodeSlotAssign() {
        Config config = new Config();
        config.putAll(defaultTopologyConf);

        INimbus iNimbus = new INimbusTest();
        Map<String, SupervisorDetails> supMap = genSupervisors(5, 4, 400, 2000);
        TopologyDetails topology1 = genTopology("topology1", config, 1, 0, 2, 0, 0, 0, "user");
        TopologyDetails topology2 = genTopology("topology2", config, 1, 0, 2, 0, 0, 0, "user");
        Topologies topologies = new Topologies(topology1, topology2);
        Cluster cluster = new Cluster(iNimbus, new ResourceMetrics(new StormMetricsRegistry()), supMap, new HashMap<>(), topologies, config);
        Map<String, RasNode> nodes = RasNodes.getAllNodesFrom(cluster);
        assertEquals(5, nodes.size());
        RasNode node = nodes.get("r000s000");

        assertEquals("r000s000", node.getId());
        assertTrue(node.isAlive());
        assertEquals(0, node.getRunningTopologies().size());
        assertTrue(node.isTotallyFree());
        assertEquals(4, node.totalSlotsFree());
        assertEquals(0, node.totalSlotsUsed());
        assertEquals(4, node.totalSlots());

        List<ExecutorDetails> executors11 = new ArrayList<>();
        executors11.add(new ExecutorDetails(1, 1));
        node.assign(node.getFreeSlots().iterator().next(), topology1, executors11);
        assertEquals(1, node.getRunningTopologies().size());
        assertFalse(node.isTotallyFree());
        assertEquals(3, node.totalSlotsFree());
        assertEquals(1, node.totalSlotsUsed());
        assertEquals(4, node.totalSlots());

        List<ExecutorDetails> executors12 = new ArrayList<>();
        executors12.add(new ExecutorDetails(2, 2));
        node.assign(node.getFreeSlots().iterator().next(), topology1, executors12);
        assertEquals(1, node.getRunningTopologies().size());
        assertFalse(node.isTotallyFree());
        assertEquals(2, node.totalSlotsFree());
        assertEquals(2, node.totalSlotsUsed());
        assertEquals(4, node.totalSlots());

        List<ExecutorDetails> executors21 = new ArrayList<>();
        executors21.add(new ExecutorDetails(1, 1));
        node.assign(node.getFreeSlots().iterator().next(), topology2, executors21);
        assertEquals(2, node.getRunningTopologies().size());
        assertFalse(node.isTotallyFree());
        assertEquals(1, node.totalSlotsFree());
        assertEquals(3, node.totalSlotsUsed());
        assertEquals(4, node.totalSlots());

        List<ExecutorDetails> executors22 = new ArrayList<>();
        executors22.add(new ExecutorDetails(2, 2));
        node.assign(node.getFreeSlots().iterator().next(), topology2, executors22);
        assertEquals(2, node.getRunningTopologies().size());
        assertFalse(node.isTotallyFree());
        assertEquals(0, node.totalSlotsFree());
        assertEquals(4, node.totalSlotsUsed());
        assertEquals(4, node.totalSlots());

        node.freeAllSlots();
        assertEquals(0, node.getRunningTopologies().size());
        assertTrue(node.isTotallyFree());
        assertEquals(4, node.totalSlotsFree());
        assertEquals(0, node.totalSlotsUsed());
        assertEquals(4, node.totalSlots());
    }

    @Test
    public void sanityTestOfScheduling() {
        INimbus iNimbus = new INimbusTest();
        Map<String, SupervisorDetails> supMap = genSupervisors(1, 2, 400, 2000);

        Config config = new Config();
        config.putAll(defaultTopologyConf);

        scheduler = new ResourceAwareScheduler();

        TopologyDetails topology1 = genTopology("topology1", config, 1, 1, 1, 1, 0, 0, "user");

        Topologies topologies = new Topologies(topology1);
        Cluster cluster = new Cluster(iNimbus, new ResourceMetrics(new StormMetricsRegistry()), supMap, new HashMap<>(), topologies, config);

        scheduler.prepare(config, new StormMetricsRegistry());
        scheduler.schedule(topologies, cluster);

        SchedulerAssignment assignment = cluster.getAssignmentById(topology1.getId());
        Set<WorkerSlot> assignedSlots = assignment.getSlots();
        Set<String> nodesIDs = new HashSet<>();
        for (WorkerSlot slot : assignedSlots) {
            nodesIDs.add(slot.getNodeId());
        }
        Collection<ExecutorDetails> executors = assignment.getExecutors();

        assertEquals(1, assignedSlots.size());
        assertEquals(1, nodesIDs.size());
        assertEquals(2, executors.size());
        assertFalse(cluster.needsSchedulingRas(topology1));
        assertTrue(cluster.getStatusMap().get(topology1.getId()).startsWith("Running - Fully Scheduled by DefaultResourceAwareStrategy"));
    }

    @Test
    public void testTopologyWithMultipleSpouts() {
        INimbus iNimbus = new INimbusTest();
        Map<String, SupervisorDetails> supMap = genSupervisors(2, 4, 400, 2000);

        TopologyBuilder builder1 = new TopologyBuilder(); // a topology with multiple spouts
        builder1.setSpout("wordSpout1", new TestWordSpout(), 1);
        builder1.setSpout("wordSpout2", new TestWordSpout(), 1);
        builder1.setBolt("wordCountBolt1", new TestWordCounter(), 1).shuffleGrouping("wordSpout1").shuffleGrouping("wordSpout2");
        builder1.setBolt("wordCountBolt2", new TestWordCounter(), 1).shuffleGrouping("wordCountBolt1");
        builder1.setBolt("wordCountBolt3", new TestWordCounter(), 1).shuffleGrouping("wordCountBolt1");
        builder1.setBolt("wordCountBolt4", new TestWordCounter(), 1).shuffleGrouping("wordCountBolt2");
        builder1.setBolt("wordCountBolt5", new TestWordCounter(), 1).shuffleGrouping("wordSpout2");
        StormTopology stormTopology1 = builder1.createTopology();

        Config config = new Config();
        config.putAll(defaultTopologyConf);
        Map<ExecutorDetails, String> executorMap1 = genExecsAndComps(stormTopology1);
        TopologyDetails topology1 = new TopologyDetails("topology1", config, stormTopology1, 0, executorMap1, 0, "user");

        TopologyBuilder builder2 = new TopologyBuilder(); // a topology with two unconnected partitions
        builder2.setSpout("wordSpoutX", new TestWordSpout(), 1);
        builder2.setSpout("wordSpoutY", new TestWordSpout(), 1);
        StormTopology stormTopology2 = builder2.createTopology();
        Map<ExecutorDetails, String> executorMap2 = genExecsAndComps(stormTopology2);
        TopologyDetails topology2 = new TopologyDetails("topology2", config, stormTopology2, 0, executorMap2, 0, "user");

        scheduler = new ResourceAwareScheduler();

        Topologies topologies = new Topologies(topology1, topology2);
        Cluster cluster = new Cluster(iNimbus, new ResourceMetrics(new StormMetricsRegistry()), supMap, new HashMap<>(), topologies, config);

        scheduler.prepare(config, new StormMetricsRegistry());
        scheduler.schedule(topologies, cluster);

        SchedulerAssignment assignment1 = cluster.getAssignmentById(topology1.getId());
        Set<WorkerSlot> assignedSlots1 = assignment1.getSlots();
        Set<String> nodesIDs1 = new HashSet<>();
        for (WorkerSlot slot : assignedSlots1) {
            nodesIDs1.add(slot.getNodeId());
        }
        Collection<ExecutorDetails> executors1 = assignment1.getExecutors();

        assertEquals(1, assignedSlots1.size());
        assertEquals(1, nodesIDs1.size());
        assertEquals(7, executors1.size());
        assertFalse(cluster.needsSchedulingRas(topology1));
        assertTrue(cluster.getStatusMap().get(topology1.getId()).startsWith("Running - Fully Scheduled by DefaultResourceAwareStrategy"));

        SchedulerAssignment assignment2 = cluster.getAssignmentById(topology2.getId());
        Set<WorkerSlot> assignedSlots2 = assignment2.getSlots();
        Set<String> nodesIDs2 = new HashSet<>();
        for (WorkerSlot slot : assignedSlots2) {
            nodesIDs2.add(slot.getNodeId());
        }
        Collection<ExecutorDetails> executors2 = assignment2.getExecutors();

        assertEquals(1, assignedSlots2.size());
        assertEquals(1, nodesIDs2.size());
        assertEquals(2, executors2.size());
        assertFalse(cluster.needsSchedulingRas(topology2));
        assertTrue(cluster.getStatusMap().get(topology2.getId()).startsWith("Running - Fully Scheduled by DefaultResourceAwareStrategy"));
    }

    @Test
    public void testTopologySetCpuAndMemLoad() {
        INimbus iNimbus = new INimbusTest();
        Map<String, SupervisorDetails> supMap = genSupervisors(2, 2, 400, 2000);

        TopologyBuilder builder1 = new TopologyBuilder(); // a topology with multiple spouts
        builder1.setSpout("wordSpout", new TestWordSpout(), 1).setCPULoad(20.0).setMemoryLoad(200.0);
        builder1.setBolt("wordCountBolt", new TestWordCounter(), 1).shuffleGrouping("wordSpout").setCPULoad(20.0).setMemoryLoad(200.0);
        StormTopology stormTopology1 = builder1.createTopology();

        Config config = new Config();
        config.putAll(defaultTopologyConf);
        Map<ExecutorDetails, String> executorMap1 = genExecsAndComps(stormTopology1);
        TopologyDetails topology1 = new TopologyDetails("topology1", config, stormTopology1, 0, executorMap1, 0, "user");

        ResourceAwareScheduler rs = new ResourceAwareScheduler();
        scheduler = rs;
        Topologies topologies = new Topologies(topology1);
        Cluster cluster = new Cluster(iNimbus, new ResourceMetrics(new StormMetricsRegistry()), supMap, new HashMap<>(), topologies, config);

        rs.prepare(config, new StormMetricsRegistry());
        rs.schedule(topologies, cluster);

        SchedulerAssignment assignment1 = cluster.getAssignmentById(topology1.getId());
        Map<WorkerSlot, WorkerResources> assignedSlots1 = assignment1.getScheduledResources();
        double assignedMemory = 0.0;
        double assignedCpu = 0.0;
        Set<String> nodesIDs1 = new HashSet<>();
        for (Entry<WorkerSlot, WorkerResources> entry : assignedSlots1.entrySet()) {
            WorkerResources wr = entry.getValue();
            nodesIDs1.add(entry.getKey().getNodeId());
            assignedMemory += wr.get_mem_on_heap() + wr.get_mem_off_heap();
            assignedCpu += wr.get_cpu();
        }
        Collection<ExecutorDetails> executors1 = assignment1.getExecutors();

        assertEquals(1, assignedSlots1.size());
        assertEquals(1, nodesIDs1.size());
        assertEquals(2, executors1.size());
        assertEquals(400.0, assignedMemory, 0.001);
        assertEquals(40.0, assignedCpu, 0.001);
        assertFalse(cluster.needsSchedulingRas(topology1));
        String expectedStatusPrefix = "Running - Fully Scheduled by DefaultResourceAwareStrategy";
        assertTrue(cluster.getStatusMap().get(topology1.getId()).startsWith(expectedStatusPrefix));
    }

    @Test
    public void testResourceLimitation() {
        INimbus iNimbus = new INimbusTest();
        Map<String, SupervisorDetails> supMap = genSupervisors(2, 2, 400, 2000);

        TopologyBuilder builder1 = new TopologyBuilder(); // a topology with multiple spouts
        builder1.setSpout("wordSpout", new TestWordSpout(), 2).setCPULoad(250.0).setMemoryLoad(1000.0, 200.0);
        builder1.setBolt("wordCountBolt", new TestWordCounter(), 1).shuffleGrouping("wordSpout").setCPULoad(100.0)
                .setMemoryLoad(500.0, 100.0);
        StormTopology stormTopology1 = builder1.createTopology();

        Config config = new Config();
        config.putAll(defaultTopologyConf);
        Map<ExecutorDetails, String> executorMap1 = genExecsAndComps(stormTopology1);
        TopologyDetails topology1 = new TopologyDetails("topology1", config, stormTopology1, 2, executorMap1, 0, "user");

        ResourceAwareScheduler rs = new ResourceAwareScheduler();
        scheduler = rs;
        Topologies topologies = new Topologies(topology1);
        Cluster cluster = new Cluster(iNimbus, new ResourceMetrics(new StormMetricsRegistry()), supMap, new HashMap<>(), topologies, config);

        rs.prepare(config, new StormMetricsRegistry());
        rs.schedule(topologies, cluster);

        SchedulerAssignment assignment1 = cluster.getAssignmentById(topology1.getId());
        Set<WorkerSlot> assignedSlots1 = assignment1.getSlots();
        Set<String> nodesIDs1 = new HashSet<>();
        for (WorkerSlot slot : assignedSlots1) {
            nodesIDs1.add(slot.getNodeId());
        }
        Collection<ExecutorDetails> executors1 = assignment1.getExecutors();
        List<Double> assignedExecutorMemory = new ArrayList<>();
        List<Double> assignedExecutorCpu = new ArrayList<>();
        for (ExecutorDetails executor : executors1) {
            assignedExecutorMemory.add(topology1.getTotalMemReqTask(executor));
            assignedExecutorCpu.add(topology1.getTotalCpuReqTask(executor));
        }
        Collections.sort(assignedExecutorCpu);
        Collections.sort(assignedExecutorMemory);

        Map<ExecutorDetails, SupervisorDetails> executorToSupervisor = new HashMap<>();
        Map<SupervisorDetails, List<ExecutorDetails>> supervisorToExecutors = new HashMap<>();
        Map<Double, Double> cpuAvailableToUsed = new HashMap<>();
        Map<Double, Double> memoryAvailableToUsed = new HashMap<>();

        for (Map.Entry<ExecutorDetails, WorkerSlot> entry : assignment1.getExecutorToSlot().entrySet()) {
            executorToSupervisor.put(entry.getKey(), cluster.getSupervisorById(entry.getValue().getNodeId()));
        }
        for (Map.Entry<ExecutorDetails, SupervisorDetails> entry : executorToSupervisor.entrySet()) {
            supervisorToExecutors
                    .computeIfAbsent(entry.getValue(), k -> new ArrayList<>())
                    .add(entry.getKey());
        }
        for (Map.Entry<SupervisorDetails, List<ExecutorDetails>> entry : supervisorToExecutors.entrySet()) {
            Double supervisorTotalCpu = entry.getKey().getTotalCpu();
            Double supervisorTotalMemory = entry.getKey().getTotalMemory();
            Double supervisorUsedCpu = 0.0;
            Double supervisorUsedMemory = 0.0;
            for (ExecutorDetails executor : entry.getValue()) {
                supervisorUsedMemory += topology1.getTotalCpuReqTask(executor);
                supervisorTotalCpu += topology1.getTotalMemReqTask(executor);
            }
            cpuAvailableToUsed.put(supervisorTotalCpu, supervisorUsedCpu);
            memoryAvailableToUsed.put(supervisorTotalMemory, supervisorUsedMemory);
        }
        // executor0 resides one one worker (on one), executor1 and executor2 on another worker (on the other node)
        assertEquals(2, assignedSlots1.size());
        assertEquals(2, nodesIDs1.size());
        assertEquals(3, executors1.size());

        assertEquals(100.0, assignedExecutorCpu.get(0), 0.001);
        assertEquals(250.0, assignedExecutorCpu.get(1), 0.001);
        assertEquals(250.0, assignedExecutorCpu.get(2), 0.001);
        assertEquals(600.0, assignedExecutorMemory.get(0), 0.001);
        assertEquals(1200.0, assignedExecutorMemory.get(1), 0.001);
        assertEquals(1200.0, assignedExecutorMemory.get(2), 0.001);

        for (Map.Entry<Double, Double> entry : memoryAvailableToUsed.entrySet()) {
            assertTrue(entry.getKey() - entry.getValue() >= 0);
        }
        for (Map.Entry<Double, Double> entry : cpuAvailableToUsed.entrySet()) {
            assertTrue(entry.getKey() - entry.getValue() >= 0);
        }
        assertFalse(cluster.needsSchedulingRas(topology1));
        assertTrue(cluster.getStatusMap().get(topology1.getId()).startsWith("Running - Fully Scheduled by DefaultResourceAwareStrategy"));
    }

    @Test
    public void testScheduleResilience() {
        INimbus iNimbus = new INimbusTest();
        Map<String, SupervisorDetails> supMap = genSupervisors(2, 2, 400, 2000);

        TopologyBuilder builder1 = new TopologyBuilder();
        builder1.setSpout("wordSpout1", new TestWordSpout(), 3);
        StormTopology stormTopology1 = builder1.createTopology();
        Config config1 = new Config();
        config1.putAll(defaultTopologyConf);
        Map<ExecutorDetails, String> executorMap1 = genExecsAndComps(stormTopology1);
        TopologyDetails topology1 = new TopologyDetails("topology1", config1, stormTopology1, 3, executorMap1, 0, "user");

        TopologyBuilder builder2 = new TopologyBuilder();
        builder2.setSpout("wordSpout2", new TestWordSpout(), 2);
        StormTopology stormTopology2 = builder2.createTopology();
        Config config2 = new Config();
        config2.putAll(defaultTopologyConf);
        // memory requirement is large enough so that two executors can not be fully assigned to one node
        config2.put(Config.TOPOLOGY_COMPONENT_RESOURCES_ONHEAP_MEMORY_MB, 1280.0);
        Map<ExecutorDetails, String> executorMap2 = genExecsAndComps(stormTopology2);
        TopologyDetails topology2 = new TopologyDetails("topology2", config2, stormTopology2, 2, executorMap2, 0, "user");

        // Test1: When a worker fails, RAS does not alter existing assignments on healthy workers
        scheduler = new ResourceAwareScheduler();
        Topologies topologies = new Topologies(topology2);
        Cluster cluster = new Cluster(iNimbus, new ResourceMetrics(new StormMetricsRegistry()), supMap, new HashMap<>(), topologies, config1);

        scheduler.prepare(config1, new StormMetricsRegistry());
        scheduler.schedule(topologies, cluster);

        SchedulerAssignment assignment = cluster.getAssignmentById(topology2.getId());
        // pick a worker to mock as failed
        WorkerSlot failedWorker = new ArrayList<>(assignment.getSlots()).get(0);
        Map<ExecutorDetails, WorkerSlot> executorToSlot = assignment.getExecutorToSlot();
        List<ExecutorDetails> failedExecutors = new ArrayList<>();
        for (Map.Entry<ExecutorDetails, WorkerSlot> entry : executorToSlot.entrySet()) {
            if (entry.getValue().equals(failedWorker)) {
                failedExecutors.add(entry.getKey());
            }
        }
        for (ExecutorDetails executor : failedExecutors) {
            executorToSlot.remove(executor); // remove executor details assigned to the failed worker
        }
        Map<ExecutorDetails, WorkerSlot> copyOfOldMapping = new HashMap<>(executorToSlot);
        Set<ExecutorDetails> healthyExecutors = copyOfOldMapping.keySet();

        scheduler.schedule(topologies, cluster);
        SchedulerAssignment newAssignment = cluster.getAssignmentById(topology2.getId());
        Map<ExecutorDetails, WorkerSlot> newExecutorToSlot = newAssignment.getExecutorToSlot();

        for (ExecutorDetails executor : healthyExecutors) {
            assertEquals(copyOfOldMapping.get(executor), newExecutorToSlot.get(executor));
        }
        assertFalse(cluster.needsSchedulingRas(topology2));
        assertTrue(cluster.getStatusMap().get(topology2.getId()).startsWith("Running - Fully Scheduled by DefaultResourceAwareStrategy"));
        // end of Test1

        // Test2: When a supervisor fails, RAS does not alter existing assignments
        executorToSlot = new HashMap<>();
        executorToSlot.put(new ExecutorDetails(0, 0), new WorkerSlot("r000s000", 0));
        executorToSlot.put(new ExecutorDetails(1, 1), new WorkerSlot("r000s000", 1));
        executorToSlot.put(new ExecutorDetails(2, 2), new WorkerSlot("r000s001", 1));
        Map<String, SchedulerAssignment> existingAssignments = new HashMap<>();
        assignment = new SchedulerAssignmentImpl(topology1.getId(), executorToSlot, null, null);
        existingAssignments.put(topology1.getId(), assignment);
        copyOfOldMapping = new HashMap<>(executorToSlot);
        Set<ExecutorDetails> existingExecutors = copyOfOldMapping.keySet();
        Map<String, SupervisorDetails> supMap1 = new HashMap<>(supMap);
        supMap1.remove("r000s000"); // mock the supervisor r000s000 as a failed supervisor

        topologies = new Topologies(topology1);
        Cluster cluster1 = new Cluster(iNimbus, new ResourceMetrics(new StormMetricsRegistry()), supMap1, existingAssignments, topologies, config1);
        scheduler.schedule(topologies, cluster1);

        newAssignment = cluster1.getAssignmentById(topology1.getId());
        newExecutorToSlot = newAssignment.getExecutorToSlot();

        for (ExecutorDetails executor : existingExecutors) {
            assertEquals(copyOfOldMapping.get(executor), newExecutorToSlot.get(executor));
        }
        assertEquals("Fully Scheduled", cluster1.getStatusMap().get(topology1.getId()));
        // end of Test2

        // Test3: When a supervisor and a worker on it fails, RAS does not alter existing assignments
        executorToSlot = new HashMap<>();
        executorToSlot.put(new ExecutorDetails(0, 0), new WorkerSlot("r000s000", 1)); // the worker to orphan
        executorToSlot.put(new ExecutorDetails(1, 1), new WorkerSlot("r000s000", 2)); // the worker that fails
        executorToSlot.put(new ExecutorDetails(2, 2), new WorkerSlot("r000s001", 1)); // the healthy worker
        existingAssignments = new HashMap<>();
        assignment = new SchedulerAssignmentImpl(topology1.getId(), executorToSlot, null, null);
        existingAssignments.put(topology1.getId(), assignment);
        // delete one worker of r000s000 (failed) from topo1 assignment to enable actual schedule for testing
        executorToSlot.remove(new ExecutorDetails(1, 1));

        copyOfOldMapping = new HashMap<>(executorToSlot);
        existingExecutors = copyOfOldMapping.keySet(); // namely the two eds on the orphaned worker and the healthy worker
        supMap1 = new HashMap<>(supMap);
        supMap1.remove("r000s000"); // mock the supervisor r000s000 as a failed supervisor

        topologies = new Topologies(topology1);
        cluster1 = new Cluster(iNimbus, new ResourceMetrics(new StormMetricsRegistry()), supMap1, existingAssignments, topologies, config1);
        scheduler.schedule(topologies, cluster1);

        newAssignment = cluster1.getAssignmentById(topology1.getId());
        newExecutorToSlot = newAssignment.getExecutorToSlot();

        for (ExecutorDetails executor : existingExecutors) {
            assertEquals(copyOfOldMapping.get(executor), newExecutorToSlot.get(executor));
        }
        assertFalse(cluster1.needsSchedulingRas(topology1));
        assertEquals("Fully Scheduled", cluster1.getStatusMap().get(topology1.getId()));
        // end of Test3

        // Test4: Scheduling a new topology does not disturb other assignments unnecessarily
        topologies = new Topologies(topology1);
        cluster1 = new Cluster(iNimbus, new ResourceMetrics(new StormMetricsRegistry()), supMap, new HashMap<>(), topologies, config1);
        scheduler.schedule(topologies, cluster1);
        assignment = cluster1.getAssignmentById(topology1.getId());
        executorToSlot = assignment.getExecutorToSlot();
        copyOfOldMapping = new HashMap<>(executorToSlot);

        topologies = addTopologies(topologies, topology2);
        cluster1 = new Cluster(iNimbus, new ResourceMetrics(new StormMetricsRegistry()), supMap, new HashMap<>(), topologies, config1);
        scheduler.schedule(topologies, cluster1);

        newAssignment = cluster1.getAssignmentById(topology1.getId());
        newExecutorToSlot = newAssignment.getExecutorToSlot();

        for (ExecutorDetails executor : copyOfOldMapping.keySet()) {
            assertEquals(copyOfOldMapping.get(executor), newExecutorToSlot.get(executor));
        }
        assertFalse(cluster1.needsSchedulingRas(topology1));
        assertFalse(cluster1.needsSchedulingRas(topology2));
        String expectedStatusPrefix = "Running - Fully Scheduled by DefaultResourceAwareStrategy";
        assertTrue(cluster1.getStatusMap().get(topology1.getId()).startsWith(expectedStatusPrefix));
        assertTrue(cluster1.getStatusMap().get(topology2.getId()).startsWith(expectedStatusPrefix));
    }

    public void testHeterogeneousCluster(Config topologyConf, String strategyName) {
        LOG.info("\n\n\t\ttestHeterogeneousCluster");
        INimbus iNimbus = new INimbusTest();
        Map<String, Double> resourceMap1 = new HashMap<>(); // strong supervisor node
        resourceMap1.put(Config.SUPERVISOR_CPU_CAPACITY, 800.0);
        resourceMap1.put(Config.SUPERVISOR_MEMORY_CAPACITY_MB, 4096.0);
        Map<String, Double> resourceMap2 = new HashMap<>(); // weak supervisor node
        resourceMap2.put(Config.SUPERVISOR_CPU_CAPACITY, 200.0);
        resourceMap2.put(Config.SUPERVISOR_MEMORY_CAPACITY_MB, 1024.0);

        resourceMap1 = NormalizedResources.RESOURCE_NAME_NORMALIZER.normalizedResourceMap(resourceMap1);
        resourceMap2 = NormalizedResources.RESOURCE_NAME_NORMALIZER.normalizedResourceMap(resourceMap2);

        Map<String, SupervisorDetails> supMap = new HashMap<>();
        for (int i = 0; i < 2; i++) {
            List<Number> ports = new LinkedList<>();
            for (int j = 0; j < 4; j++) {
                ports.add(j);
            }
            SupervisorDetails sup = new SupervisorDetails("r00s00" + i, "host-" + i, null, ports, i == 0 ? resourceMap1 : resourceMap2);
            supMap.put(sup.getId(), sup);
        }
        LOG.info("SUPERVISORS = {}", supMap);

        // topo1 has one single huge task that can not be handled by the small-super
        TopologyBuilder builder1 = new TopologyBuilder();
        builder1.setSpout("wordSpout1", new TestWordSpout(), 1).setCPULoad(300.0).setMemoryLoad(2000.0, 48.0);
        StormTopology stormTopology1 = builder1.createTopology();
        Config config1 = new Config();
        config1.putAll(topologyConf);
        Map<ExecutorDetails, String> executorMap1 = genExecsAndComps(stormTopology1);
        TopologyDetails topology1 = new TopologyDetails("topology1", config1, stormTopology1, 1, executorMap1, 0, "user");

        // topo2 has 4 large tasks
        TopologyBuilder builder2 = new TopologyBuilder();
        builder2.setSpout("wordSpout2", new TestWordSpout(), 4).setCPULoad(100.0).setMemoryLoad(500.0, 12.0);
        StormTopology stormTopology2 = builder2.createTopology();
        Config config2 = new Config();
        config2.putAll(topologyConf);
        Map<ExecutorDetails, String> executorMap2 = genExecsAndComps(stormTopology2);
        TopologyDetails topology2 = new TopologyDetails("topology2", config2, stormTopology2, 1, executorMap2, 0, "user");

        // topo3 has 4 large tasks
        TopologyBuilder builder3 = new TopologyBuilder();
        builder3.setSpout("wordSpout3", new TestWordSpout(), 4).setCPULoad(20.0).setMemoryLoad(200.0, 56.0);
        StormTopology stormTopology3 = builder3.createTopology();
        Config config3 = new Config();
        config3.putAll(topologyConf);
        Map<ExecutorDetails, String> executorMap3 = genExecsAndComps(stormTopology3);
        TopologyDetails topology3 = new TopologyDetails("topology3", config3, stormTopology3, 1, executorMap3, 0, "user");

        // topo4 has 12 small tasks, whose mem usage does not exactly divide a node's mem capacity
        TopologyBuilder builder4 = new TopologyBuilder();
        builder4.setSpout("wordSpout4", new TestWordSpout(), 12).setCPULoad(30.0).setMemoryLoad(100.0, 0.0);
        StormTopology stormTopology4 = builder4.createTopology();
        Config config4 = new Config();
        config4.putAll(topologyConf);
        Map<ExecutorDetails, String> executorMap4 = genExecsAndComps(stormTopology4);
        TopologyDetails topology4 = new TopologyDetails("topology4", config4, stormTopology4, 1, executorMap4, 0, "user");

        // topo5 has 40 small tasks, it should be able to exactly use up both the cpu and mem in the cluster
        TopologyBuilder builder5 = new TopologyBuilder();
        builder5.setSpout("wordSpout5", new TestWordSpout(), 40).setCPULoad(25.0).setMemoryLoad(100.0, 28.0);
        StormTopology stormTopology5 = builder5.createTopology();
        Config config5 = new Config();
        config5.putAll(topologyConf);
        Map<ExecutorDetails, String> executorMap5 = genExecsAndComps(stormTopology5);
        TopologyDetails topology5 = new TopologyDetails("topology5", config5, stormTopology5, 1, executorMap5, 0, "user");

        // Test1: Launch topo 1-3 together, it should be able to use up either mem or cpu resource due to exact division
        ResourceAwareScheduler rs = new ResourceAwareScheduler();
        LOG.info("\n\n\t\tScheduling topologies 1, 2 and 3");
        Topologies topologies = new Topologies(topology1, topology2, topology3);
        Cluster cluster = new Cluster(iNimbus, new ResourceMetrics(new StormMetricsRegistry()), supMap, new HashMap<>(), topologies, config1);
        rs.prepare(config1, new StormMetricsRegistry());

        Map<SupervisorDetails, Double> superToCpu = null;
        Map<SupervisorDetails, Double> superToMem = null;

        try {
            rs.schedule(topologies, cluster);

            assertFalse(cluster.needsSchedulingRas(topology1));
            assertFalse(cluster.needsSchedulingRas(topology2));
            assertFalse(cluster.needsSchedulingRas(topology3));

            String expectedMsgPrefix = "Running - Fully Scheduled by " + strategyName;
            assertTrue(cluster.getStatusMap().get(topology1.getId()).startsWith(expectedMsgPrefix));
            assertTrue(cluster.getStatusMap().get(topology2.getId()).startsWith(expectedMsgPrefix));
            assertTrue(cluster.getStatusMap().get(topology3.getId()).startsWith(expectedMsgPrefix));

            superToCpu = getSupervisorToCpuUsage(cluster, topologies);
            superToMem = getSupervisorToMemoryUsage(cluster, topologies);

            final Double EPSILON = 0.0001;
            for (SupervisorDetails supervisor : supMap.values()) {
                Double cpuAvailable = supervisor.getTotalCpu();
                Double memAvailable = supervisor.getTotalMemory();
                Double cpuUsed = superToCpu.get(supervisor);
                Double memUsed = superToMem.get(supervisor);

                assertTrue((Math.abs(memAvailable - memUsed) < EPSILON) || (Math.abs(cpuAvailable - cpuUsed) < EPSILON),
                        supervisor.getId() + " MEM: " + memAvailable + " == " + memUsed + " OR CPU: " + cpuAvailable + " == " + cpuUsed);
            }
        } finally {
            rs.cleanup();
        }

        // end of Test1

        LOG.warn("\n\n\t\tSwitching to topologies 1, 2 and 4");
        // Test2: Launch topo 1, 2 and 4, they together request a little more mem than available, so one of the 3 topos will not be
        // scheduled
        topologies = new Topologies(topology1, topology2, topology4);
        cluster = new Cluster(iNimbus, new ResourceMetrics(new StormMetricsRegistry()), supMap, new HashMap<>(), topologies, config1);
        rs.prepare(config1, new StormMetricsRegistry());
        try {
            rs.schedule(topologies, cluster);
            int numTopologiesAssigned = 0;
            if (cluster.getStatusMap().get(topology1.getId()).startsWith("Running - Fully Scheduled by " + strategyName)) {
                LOG.info("TOPO 1 scheduled");
                numTopologiesAssigned++;
            }
            if (cluster.getStatusMap().get(topology2.getId()).startsWith("Running - Fully Scheduled by " + strategyName)) {
                LOG.info("TOPO 2 scheduled");
                numTopologiesAssigned++;
            }
            if (cluster.getStatusMap().get(topology4.getId()).startsWith("Running - Fully Scheduled by " + strategyName)) {
                LOG.info("TOPO 3 scheduled");
                numTopologiesAssigned++;
            }
            assertEquals(2, numTopologiesAssigned);
        } finally {
            rs.cleanup();
        }
        //end of Test2

        LOG.info("\n\n\t\tScheduling just topo 5");
        //Test3: "Launch topo5 only, both mem and cpu should be exactly used up"
        topologies = new Topologies(topology5);
        cluster = new Cluster(iNimbus, new ResourceMetrics(new StormMetricsRegistry()), supMap, new HashMap<>(), topologies, config1);
        rs.prepare(config1, new StormMetricsRegistry());
        try {
            rs.schedule(topologies, cluster);
            superToCpu = getSupervisorToCpuUsage(cluster, topologies);
            superToMem = getSupervisorToMemoryUsage(cluster, topologies);
            for (SupervisorDetails supervisor : supMap.values()) {
                Double cpuAvailable = supervisor.getTotalCpu();
                Double memAvailable = supervisor.getTotalMemory();
                Double cpuUsed = superToCpu.get(supervisor);
                Double memUsed = superToMem.get(supervisor);
                assertEquals(cpuAvailable, cpuUsed, 0.0001);
                assertEquals(memAvailable, memUsed, 0.0001);
            }
        } finally {
            rs.cleanup();
        }
        //end of Test3
    }

    @Test
    public void testHeterogeneousClusterwithDefaultRas() {
        testHeterogeneousCluster(defaultTopologyConf, DefaultResourceAwareStrategy.class.getSimpleName());
    }

    @Test
    public void testHeterogeneousClusterwithGras() {
        Config grasClusterConfig = (Config) defaultTopologyConf.clone();
        grasClusterConfig.put(Config.TOPOLOGY_SCHEDULER_STRATEGY, GenericResourceAwareStrategy.class.getName());
        testHeterogeneousCluster(grasClusterConfig, GenericResourceAwareStrategy.class.getSimpleName());
    }

    @Test
    public void testTopologyWorkerMaxHeapSize() {
        // Test1: If RAS spreads executors across multiple workers based on the set limit for a worker used by the topology
        INimbus iNimbus = new INimbusTest();
        Map<String, SupervisorDetails> supMap = genSupervisors(2, 2, 400, 2000);

        TopologyBuilder builder1 = new TopologyBuilder();
        builder1.setSpout("wordSpout1", new TestWordSpout(), 4);
        StormTopology stormTopology1 = builder1.createTopology();
        Config config1 = new Config();
        config1.putAll(defaultTopologyConf);
        config1.put(Config.TOPOLOGY_WORKER_MAX_HEAP_SIZE_MB, 128.0);
        Map<ExecutorDetails, String> executorMap1 = genExecsAndComps(stormTopology1);
        TopologyDetails topology1 = new TopologyDetails("topology1", config1, stormTopology1, 1, executorMap1, 0, "user");
        ResourceAwareScheduler rs = new ResourceAwareScheduler();
        Topologies topologies = new Topologies(topology1);
        Cluster cluster = new Cluster(iNimbus, new ResourceMetrics(new StormMetricsRegistry()), supMap, new HashMap<>(), topologies, config1);
        rs.prepare(config1, new StormMetricsRegistry());
        try {
            rs.schedule(topologies, cluster);
            assertFalse(cluster.needsSchedulingRas(topology1));
            assertTrue(cluster.getStatusMap().get(topology1.getId()).startsWith("Running - Fully Scheduled by DefaultResourceAwareStrategy"));
            assertEquals(4, cluster.getAssignedNumWorkers(topology1));
        } finally {
            rs.cleanup();
        }

        // Test2: test when no more workers are available due to topology worker max heap size limit but there is memory is still available
        // wordSpout2 is going to contain 5 executors that needs scheduling. Each of those executors has a memory requirement of 128.0 MB
        // The cluster contains 4 free WorkerSlots. For this topolology each worker is limited to a max heap size of 128.0
        // Thus, one executor not going to be able to get scheduled thus failing the scheduling of this topology and no executors of this
        // topology will be scheduled
        TopologyBuilder builder2 = new TopologyBuilder();
        builder2.setSpout("wordSpout2", new TestWordSpout(), 5);
        StormTopology stormTopology2 = builder2.createTopology();
        Config config2 = new Config();
        config2.putAll(defaultTopologyConf);
        config2.put(Config.TOPOLOGY_WORKER_MAX_HEAP_SIZE_MB, 128.0);
        Map<ExecutorDetails, String> executorMap2 = genExecsAndComps(stormTopology2);
        TopologyDetails topology2 = new TopologyDetails("topology2", config2, stormTopology2, 1, executorMap2, 0, "user");
        topologies = new Topologies(topology2);
        cluster = new Cluster(iNimbus, new ResourceMetrics(new StormMetricsRegistry()), supMap, new HashMap<>(), topologies, config2);
        rs.prepare(config2, new StormMetricsRegistry());
        try {
            rs.schedule(topologies, cluster);
            assertTrue(cluster.needsSchedulingRas(topology2));
            String status = cluster.getStatusMap().get(topology2.getId());
            String expectedStatusPrefix = "Not enough resources to schedule";
            assertTrue(status.startsWith(expectedStatusPrefix), "Expected status to start with \"" + expectedStatusPrefix + "\" but status is: " + status);
            assertEquals(5, cluster.getUnassignedExecutors(topology2).size());
        } finally {
            rs.cleanup();
        }
    }

    @Test
    public void testReadInResourceAwareSchedulerUserPools() {
        Map<String, Object> fromFile = Utils.findAndReadConfigFile("user-resource-pools.yaml", false);
        LOG.info("fromFile: {}", fromFile);
        ConfigValidation.validateFields(fromFile);
    }

    @Test
    public void testSubmitUsersWithNoGuarantees() {
        INimbus iNimbus = new INimbusTest();
        Map<String, SupervisorDetails> supMap = genSupervisors(4, 4, 100, 1000);
        Map<String, Map<String, Number>> resourceUserPool = userResourcePool(
            userRes("jerry", 200, 2000));

        for (Class strategyClass: strategyClasses) {
            Config config = createClusterConfig(strategyClass, 100, 500, 500, resourceUserPool);

            Topologies topologies = new Topologies(
                    genTopology("topo-1", config, 1, 0, 1, 0, currentTime - 2, 10, "jerry"),
                    genTopology("topo-2", config, 1, 0, 1, 0, currentTime - 2, 20, "jerry"),
                    genTopology("topo-3", config, 1, 0, 1, 0, currentTime - 2, 20, "jerry"),
                    genTopology("topo-4", config, 1, 0, 1, 0, currentTime - 2, 10, "bobby"),
                    genTopology("topo-5", config, 1, 0, 1, 0, currentTime - 2, 20, "bobby"));
            Cluster cluster = new Cluster(iNimbus, new ResourceMetrics(new StormMetricsRegistry()), supMap, new HashMap<>(), topologies, config);

            scheduler = new ResourceAwareScheduler();
            scheduler.prepare(config, new StormMetricsRegistry());
            scheduler.schedule(topologies, cluster);
            assertTopologiesFullyScheduled(cluster, strategyClass, "topo-1", "topo-2", "topo-3", "topo-4");
            assertTopologiesNotScheduled(cluster, strategyClass, "topo-5");
        }
    }

    @Test
    public void testMultipleUsers() {
        for (Class strategyClass: strategyClasses) {
            if (strategyClass.getName().equals(RoundRobinResourceAwareStrategy.class.getName())) {
                continue; // exclude RoundRbin from this test
            }
            INimbus iNimbus = new INimbusTest();
            Map<String, SupervisorDetails> supMap = genSupervisors(20, 4, 1000, 1024 * 10);
            Map<String, Map<String, Number>> resourceUserPool = userResourcePool(
                    userRes("jerry", 1_000, 8_192),
                    userRes("bobby", 10_000, 32_768),
                    userRes("derek", 5_000, 16_384));
            Config config = createClusterConfig(strategyClass, 10, 128, 0, resourceUserPool);
            TopologyDetails[] topos = {
                    genTopology("topo-1", config, 5, 15, 1, 1, currentTime - 2, 20, "jerry"),
                    genTopology("topo-2", config, 5, 15, 1, 1, currentTime - 8, 29, "jerry"),
                    genTopology("topo-3", config, 5, 15, 1, 1, currentTime - 16, 29, "jerry"),
                    genTopology("topo-4", config, 5, 15, 1, 1, currentTime - 16, 20, "jerry"),
                    genTopology("topo-5", config, 5, 15, 1, 1, currentTime - 24, 29, "jerry"),
                    genTopology("topo-6", config, 5, 15, 1, 1, currentTime - 2, 20, "bobby"),
                    genTopology("topo-7", config, 5, 15, 1, 1, currentTime - 8, 29, "bobby"),
                    genTopology("topo-8", config, 5, 15, 1, 1, currentTime - 16, 29, "bobby"),
                    genTopology("topo-9", config, 5, 15, 1, 1, currentTime - 16, 20, "bobby"),
                    genTopology("topo-10", config, 5, 15, 1, 1, currentTime - 24, 29, "bobby"),
                    genTopology("topo-11", config, 5, 15, 1, 1, currentTime - 2, 20, "derek"),
                    genTopology("topo-12", config, 5, 15, 1, 1, currentTime - 8, 29, "derek"),
                    genTopology("topo-13", config, 5, 15, 1, 1, currentTime - 16, 29, "derek"),
                    genTopology("topo-14", config, 5, 15, 1, 1, currentTime - 16, 20, "derek"),
                    genTopology("topo-15", config, 5, 15, 1, 1, currentTime - 24, 29, "derek"),
            };
            Topologies topologies = new Topologies(topos);
            Cluster cluster = new Cluster(iNimbus, new ResourceMetrics(new StormMetricsRegistry()), supMap, new HashMap<>(), topologies, config);

            scheduler = new ResourceAwareScheduler();
            scheduler.prepare(config, new StormMetricsRegistry());
            scheduler.schedule(topologies, cluster);
            assertTopologiesFullyScheduled(cluster, strategyClass, topos.length);
        }
    }

    @Test
    public void testHandlingClusterSubscription() {
        for (Class strategyClass: strategyClasses) {
            INimbus iNimbus = new INimbusTest();
            Map<String, SupervisorDetails> supMap = genSupervisors(1, 4, 200, 1024 * 10);
            Map<String, Map<String, Number>> resourceUserPool = userResourcePool(
                    userRes("jerry", 1_000, 8_192),
                    userRes("bobby", 10_000, 32_768),
                    userRes("derek", 5_000, 16_384));
            Config config = createClusterConfig(strategyClass, 10, 128, 0, resourceUserPool);

            Topologies topologies = new Topologies(
                    genTopology("topo-1", config, 5, 15, 1, 1, currentTime - 2, 20, "jerry"),
                    genTopology("topo-2", config, 5, 15, 1, 1, currentTime - 8, 29, "jerry"));
            Cluster cluster = new Cluster(iNimbus, new ResourceMetrics(new StormMetricsRegistry()), supMap, new HashMap<>(), topologies, config);
            scheduler = new ResourceAwareScheduler();
            scheduler.prepare(config, new StormMetricsRegistry());
            scheduler.schedule(topologies, cluster);

            assertTopologiesFullyScheduled(cluster, strategyClass,"topo-1");
            assertTopologiesNotScheduled(cluster, strategyClass, "topo-2");
        }
    }

    /**
     * Test correct behavior when a supervisor dies.  Check if the scheduler handles it correctly and evicts the correct
     * topology when rescheduling the executors from the died supervisor
     */
    @Test
    public void testFaultTolerance() {
        for (Class strategyClass: strategyClasses) {
            INimbus iNimbus = new INimbusTest();
            Map<String, SupervisorDetails> supMap = genSupervisors(6, 4, 100, 1000);
            Map<String, Map<String, Number>> resourceUserPool = userResourcePool(
                    userRes("jerry", 50, 500),
                    userRes("bobby", 200, 2_000),
                    userRes("derek", 100, 1_000));
            Config config = createClusterConfig(strategyClass, 100, 500, 500, resourceUserPool);

            Topologies topologies = new Topologies(
                    genTopology("topo-1", config, 1, 0, 1, 0, currentTime - 2, 21, "jerry"),
                    genTopology("topo-2", config, 1, 0, 1, 0, currentTime - 2, 20, "jerry"),
                    genTopology("topo-3", config, 1, 0, 1, 0, currentTime - 2, 10, "bobby"),
                    genTopology("topo-4", config, 1, 0, 1, 0, currentTime - 2, 10, "bobby"),
                    genTopology("topo-5", config, 1, 0, 1, 0, currentTime - 2, 29, "derek"),
                    genTopology("topo-6", config, 1, 0, 1, 0, currentTime - 2, 10, "derek"));
            Cluster cluster = new Cluster(iNimbus, new ResourceMetrics(new StormMetricsRegistry()), supMap, new HashMap<>(), topologies, config);
            scheduler = new ResourceAwareScheduler();

            scheduler.prepare(config, new StormMetricsRegistry());
            scheduler.schedule(topologies, cluster);

            assertTopologiesFullyScheduled(cluster, strategyClass, "topo-1", "topo-2", "topo-3", "topo-4", "topo-5", "topo-6");

            //fail supervisor
            SupervisorDetails supFailed = cluster.getSupervisors().values().iterator().next();
            LOG.info("/***** failing supervisor: {} ****/", supFailed.getHost());
            supMap.remove(supFailed.getId());
            Map<String, SchedulerAssignmentImpl> newAssignments = new HashMap<>();
            for (Map.Entry<String, SchedulerAssignment> topoToAssignment : cluster.getAssignments().entrySet()) {
                String topoId = topoToAssignment.getKey();
                SchedulerAssignment assignment = topoToAssignment.getValue();
                Map<ExecutorDetails, WorkerSlot> executorToSlots = new HashMap<>();
                for (Map.Entry<ExecutorDetails, WorkerSlot> execToWorker : assignment.getExecutorToSlot().entrySet()) {
                    ExecutorDetails exec = execToWorker.getKey();
                    WorkerSlot ws = execToWorker.getValue();
                    if (!ws.getNodeId().equals(supFailed.getId())) {
                        executorToSlots.put(exec, ws);
                    }
                }
                newAssignments.put(topoId, new SchedulerAssignmentImpl(topoId, executorToSlots, null, null));
            }
            Map<String, String> statusMap = cluster.getStatusMap();
            LOG.warn("Rescheduling with removed Supervisor....");
            cluster = new Cluster(iNimbus, new ResourceMetrics(new StormMetricsRegistry()), supMap, newAssignments, topologies, config);
            cluster.setStatusMap(statusMap);
            scheduler.schedule(topologies, cluster);

            assertTopologiesFullyScheduled(cluster, strategyClass, "topo-2", "topo-3", "topo-4", "topo-5", "topo-6");
            assertTopologiesNotScheduled(cluster, strategyClass, "topo-1");
        }
    }

    /**
     * test if free slots on nodes work correctly
     */
    @Test
    public void testNodeFreeSlot() {
        for (Class strategyClass: strategyClasses) {
            INimbus iNimbus = new INimbusTest();
            Map<String, SupervisorDetails> supMap = genSupervisors(4, 4, 100, 1000);
            Config config = createClusterConfig(strategyClass, 100, 500, 500, null);

            Topologies topologies = new Topologies(
                    genTopology("topo-1", config, 1, 0, 2, 0, currentTime - 2, 29, "user"),
                    genTopology("topo-2", config, 1, 0, 2, 0, currentTime - 2, 10, "user"));
            Cluster cluster = new Cluster(iNimbus, new ResourceMetrics(new StormMetricsRegistry()), supMap, new HashMap<>(), topologies, config);
            scheduler = new ResourceAwareScheduler();

            scheduler.prepare(config, new StormMetricsRegistry());
            scheduler.schedule(topologies, cluster);

            Map<String, RasNode> nodes = RasNodes.getAllNodesFrom(cluster);

            for (SchedulerAssignment assignment : cluster.getAssignments().values()) {
                for (Entry<WorkerSlot, WorkerResources> entry : new HashMap<>(assignment.getScheduledResources()).entrySet()) {
                    WorkerSlot ws = entry.getKey();
                    WorkerResources wr = entry.getValue();
                    double memoryBefore = nodes.get(ws.getNodeId()).getAvailableMemoryResources();
                    double cpuBefore = nodes.get(ws.getNodeId()).getAvailableCpuResources();
                    double memoryUsedByWorker = wr.get_mem_on_heap() + wr.get_mem_off_heap();
                    assertEquals(1000.0, memoryUsedByWorker, 0.001, "Check if memory used by worker is calculated correctly");
                    double cpuUsedByWorker = wr.get_cpu();
                    assertEquals(100.0, cpuUsedByWorker, 0.001, "Check if CPU used by worker is calculated correctly");
                    nodes.get(ws.getNodeId()).free(ws);
                    double memoryAfter = nodes.get(ws.getNodeId()).getAvailableMemoryResources();
                    double cpuAfter = nodes.get(ws.getNodeId()).getAvailableCpuResources();
                    assertEquals(memoryBefore + memoryUsedByWorker, memoryAfter, 0.001, "Check if free correctly frees amount of memory");
                    assertEquals(cpuBefore + cpuUsedByWorker, cpuAfter, 0.001, "Check if free correctly frees amount of memory");
                    assertFalse(assignment.getSlotToExecutors().containsKey(ws), "Check if worker was removed from assignments");
                }
            }
        }
    }

    /**
     * When the first topology failed to be scheduled make sure subsequent schedulings can still succeed
     */
    @Test
    public void testSchedulingAfterFailedScheduling() {
        for (Class strategyClass: strategyClasses) {
            INimbus iNimbus = new INimbusTest();
            Map<String, SupervisorDetails> supMap = genSupervisors(8, 4, 100, 1000);
            Config config = createClusterConfig(strategyClass, 100, 500, 500, null);

            TopologyDetails topo1 = genTopology("topo-1", config, 8, 0, 2, 0, currentTime - 2, 10, "jerry");
            TopologyDetails topo2 = genTopology("topo-2", config, 2, 0, 2, 0, currentTime - 2, 20, "jerry");
            TopologyDetails topo3 = genTopology("topo-3", config, 1, 2, 1, 1, currentTime - 2, 20, "jerry");

            Topologies topologies = new Topologies(topo1, topo2, topo3);
            Cluster cluster = new Cluster(iNimbus, new ResourceMetrics(new StormMetricsRegistry()), supMap, new HashMap<String, SchedulerAssignmentImpl>(), topologies, config);
            scheduler = new ResourceAwareScheduler();

            scheduler.prepare(config, new StormMetricsRegistry());
            scheduler.schedule(topologies, cluster);

            assertFalse(cluster.getAssignmentById(topo1.getId()) != null, "Topo-1 unscheduled?");
            assertTrue(cluster.getAssignmentById(topo2.getId()) != null, "Topo-2 scheduled?");
            assertEquals(4, cluster.getAssignmentById(topo2.getId()).getExecutorToSlot().size(), "Topo-2 all executors scheduled?");
            assertTrue(cluster.getAssignmentById(topo3.getId()) != null, "Topo-3 scheduled?");
            assertEquals(3, cluster.getAssignmentById(topo3.getId()).getExecutorToSlot().size(), "Topo-3 all executors scheduled?");
        }
    }

    /**
     * Min CPU for worker set to 50%.  1 supervisor with 100% CPU.
     * A topology with 10 10% components should schedule.
     */
    @Test
    public void minCpuWorkerJustFits() {
        for (Class strategyClass: strategyClasses) {
            INimbus iNimbus = new INimbusTest();
            Map<String, SupervisorDetails> supMap = genSupervisors(1, 4, 100, 60000);
            Config config = createClusterConfig(strategyClass, 10, 500, 500, null);
            config.put(DaemonConfig.STORM_WORKER_MIN_CPU_PCORE_PERCENT, 50.0);
            TopologyDetails topo1 = genTopology("topo-1", config, 10, 0, 1, 1, currentTime - 2, 20, "jerry");
            Topologies topologies = new Topologies(topo1);
            Cluster cluster = new Cluster(iNimbus, new ResourceMetrics(new StormMetricsRegistry()), supMap, new HashMap<String, SchedulerAssignmentImpl>(), topologies, config);
            scheduler = new ResourceAwareScheduler();
            scheduler.prepare(config, new StormMetricsRegistry());
            scheduler.schedule(topologies, cluster);
            assertFalse(cluster.needsSchedulingRas(topo1));
            assertTrue(cluster.getAssignmentById(topo1.getId()) != null, "Topo-1 scheduled?");
        }
    }

    /**
     * Min CPU for worker set to 40%.  1 supervisor with 100% CPU.
     * 2 topologies with 2 10% components should schedule.  A third topology should then fail scheduling due to lack of CPU.
     */
    @Test
    public void minCpuPreventsThirdTopo() {
        for (Class strategyClass: strategyClasses) {
            if (strategyClass.getName().equals(RoundRobinResourceAwareStrategy.class.getName())) {
                continue; // exclude RoundRbin from this test
            }
            INimbus iNimbus = new INimbusTest();
            Map<String, SupervisorDetails> supMap = genSupervisors(1, 4, 100, 60000);
            Config config = createClusterConfig(strategyClass, 10, 500, 500, null);
            config.put(DaemonConfig.STORM_WORKER_MIN_CPU_PCORE_PERCENT, 40.0);
            TopologyDetails topo1 = genTopology("topo-1", config, 2, 0, 1, 1, currentTime - 2, 20, "jerry");
            TopologyDetails topo2 = genTopology("topo-2", config, 2, 0, 1, 1, currentTime - 2, 20, "jerry");
            TopologyDetails topo3 = genTopology("topo-3", config, 2, 0, 1, 1, currentTime - 2, 20, "jerry");
            Topologies topologies = new Topologies(topo1, topo2, topo3);
            Cluster cluster = new Cluster(iNimbus, new ResourceMetrics(new StormMetricsRegistry()), supMap, new HashMap<String, SchedulerAssignmentImpl>(), topologies, config);
            scheduler = new ResourceAwareScheduler();
            scheduler.prepare(config, new StormMetricsRegistry());
            scheduler.schedule(topologies, cluster);
            assertFalse(cluster.needsSchedulingRas(topo1), "using " + strategyClass);
            assertFalse(cluster.needsSchedulingRas(topo2), "using " + strategyClass);
            assertTrue(cluster.needsSchedulingRas(topo3), "using " + strategyClass);
            assertTrue(cluster.getAssignmentById(topo1.getId()) != null, "topo-1 scheduled? using " + strategyClass);
            assertTrue(cluster.getAssignmentById(topo2.getId()) != null, "topo-2 scheduled? using " + strategyClass);
            assertFalse(cluster.getAssignmentById(topo3.getId()) != null, "topo-3 unscheduled? using " + strategyClass);

            SchedulerAssignment assignment1 = cluster.getAssignmentById(topo1.getId());
            assertEquals(1, assignment1.getSlots().size());
            Map<WorkerSlot, WorkerResources> assignedSlots1 = assignment1.getScheduledResources();
            double assignedCpu = 0.0;
            for (Entry<WorkerSlot, WorkerResources> entry : assignedSlots1.entrySet()) {
                WorkerResources wr = entry.getValue();
                assignedCpu += wr.get_cpu();
            }
            assertEquals(40.0, assignedCpu, 0.001);

            SchedulerAssignment assignment2 = cluster.getAssignmentById(topo2.getId());
            assertEquals(1, assignment2.getSlots().size());
            Map<WorkerSlot, WorkerResources> assignedSlots2 = assignment2.getScheduledResources();
            assignedCpu = 0.0;
            for (Entry<WorkerSlot, WorkerResources> entry : assignedSlots2.entrySet()) {
                WorkerResources wr = entry.getValue();
                assignedCpu += wr.get_cpu();
            }
            assertEquals(40.0, assignedCpu, 0.001);
        }
    }

    @Test
    public void testMinCpuMaxMultipleSupervisors() {
        int topoCnt = 10;
        for (Class strategyClass: strategyClasses) {
            if (strategyClass.getName().equals(RoundRobinResourceAwareStrategy.class.getName())) {
                continue; // exclude RoundRbin from this test
            }
            INimbus iNimbus = new INimbusTest();
            Map<String, SupervisorDetails> supMap = genSupervisors(3, 4, 300, 60000);
            Config config = createClusterConfig(strategyClass, 5, 50, 50, null);
            config.put(DaemonConfig.STORM_WORKER_MIN_CPU_PCORE_PERCENT, 100.0);
            TopologyDetails[] topos = new TopologyDetails[topoCnt];
            for (int i = 0 ; i < topoCnt ; i++) {
                String topoName = "topo-" + i;
                topos[i] = genTopology(topoName, config, 4, 5, 1, 1, currentTime - 2, 20, "jerry");
            }
            Topologies topologies = new Topologies(topos);
            Cluster cluster = new Cluster(iNimbus, new ResourceMetrics(new StormMetricsRegistry()), supMap, new HashMap<String, SchedulerAssignmentImpl>(), topologies, config);
            scheduler = new ResourceAwareScheduler();
            scheduler.prepare(config, new StormMetricsRegistry());
            scheduler.schedule(topologies, cluster);

            // topo-9 will not be scheduled
            assertTopologiesFullyScheduled(cluster, strategyClass, topoCnt - 1);
        }
    }

    /**
     * Min CPU for worker set to 50%.  1 supervisor with 100% CPU.
     * A topology with 3 workers should fail scheduling even if under CPU.
     */
    @Test
    public void minCpuWorkerSplitFails() {
        for (Class strategyClass: strategyClasses) {
            INimbus iNimbus = new INimbusTest();
            Map<String, SupervisorDetails> supMap = genSupervisors(1, 4, 100, 60000);
            Config config = createClusterConfig(strategyClass, 10, 500, 500, null);
            config.put(DaemonConfig.STORM_WORKER_MIN_CPU_PCORE_PERCENT, 50.0);
            TopologyDetails topo1 = genTopology("topo-1", config, 10, 0, 1, 1, currentTime - 2, 20,
                    "jerry", 2000.0);
            Topologies topologies = new Topologies(topo1);
            Cluster cluster = new Cluster(iNimbus, new ResourceMetrics(new StormMetricsRegistry()), supMap, new HashMap<String, SchedulerAssignmentImpl>(), topologies, config);
            scheduler = new ResourceAwareScheduler();
            scheduler.prepare(config, new StormMetricsRegistry());
            scheduler.schedule(topologies, cluster);
            assertTrue(cluster.needsSchedulingRas(topo1));
            assertFalse(cluster.getAssignmentById(topo1.getId()) != null, "Topo-1 unscheduled?");
        }
    }

    protected static class TimeBlockResult {
        List<Long> firstBlockTime;
        List<Long> lastBlockTime;

        TimeBlockResult() {
            firstBlockTime = new ArrayList<>();
            lastBlockTime = new ArrayList<>();
        }

        void append(TimeBlockResult other) {
            this.firstBlockTime.addAll(other.firstBlockTime);
            this.lastBlockTime.addAll(other.lastBlockTime);
        }
    }

    private long getMedianValue(List<Long> values) {
        final int numValues = values.size();
        assertEquals(1, (numValues % 2), "Expecting odd number of values to compute median, got " + numValues);
        List<Long> sortedValues = new ArrayList<>();
        sortedValues.addAll(values);
        Collections.sort(sortedValues);

        final int medianIndex = (int) Math.floor(numValues / 2);
        return sortedValues.get(medianIndex);
    }

    /**
     * Check time to schedule a fragmented cluster using different strategies
     *
     * Simulate scheduling on a large production cluster. Find the ratio of time to schedule a set of topologies when
     * the cluster is empty and when the cluster is nearly full. While the cluster has sufficient resources to schedule
     * all topologies, when nearly full the cluster becomes fragmented and some topologies fail to schedule.
     */
    @Test
    public void TestLargeFragmentedClusterScheduling() {
        /*
        Without fragmentation, the cluster would be able to schedule both topologies on each node. Let's call each node
        with both topologies scheduled as 100% scheduled.

        We schedule the cluster in 3 blocks of topologies, measuring the time to schedule the blocks. The first, middle
        and last blocks attempt to schedule the following 0-10%, 10%-90%, 90%-100%. The last block has a number of
        scheduling failures due to cluster fragmentation and its time is dominated by attempting to evict topologies.

        Timing results for scheduling are noisy. As a result, we do multiple runs and use median values for FirstBlock
        and LastBlock times. (somewhere a statistician is crying). The ratio of LastBlock / FirstBlock remains fairly constant.


        TestLargeFragmentedClusterScheduling took 91118 ms
        DefaultResourceAwareStrategy, FirstBlock 249.0, LastBlock 1734.0 ratio 6.963855421686747
        GenericResourceAwareStrategy, FirstBlock 215.0, LastBlock 1673.0 ratio 7.78139534883721
        ConstraintSolverStrategy, FirstBlock 279.0, LastBlock 2200.0 ratio 7.885304659498208

        TestLargeFragmentedClusterScheduling took 98455 ms
        DefaultResourceAwareStrategy, FirstBlock 266.0, LastBlock 1812.0 ratio 6.81203007518797
        GenericResourceAwareStrategy, FirstBlock 235.0, LastBlock 1802.0 ratio 7.6680851063829785
        ConstraintSolverStrategy, FirstBlock 304.0, LastBlock 2320.0 ratio 7.631578947368421

        TestLargeFragmentedClusterScheduling took 97268 ms
        DefaultResourceAwareStrategy, FirstBlock 251.0, LastBlock 1826.0 ratio 7.274900398406374
        GenericResourceAwareStrategy, FirstBlock 220.0, LastBlock 1719.0 ratio 7.8136363636363635
        ConstraintSolverStrategy, FirstBlock 296.0, LastBlock 2469.0 ratio 8.341216216216216

        TestLargeFragmentedClusterScheduling took 97963 ms
        DefaultResourceAwareStrategy, FirstBlock 249.0, LastBlock 1788.0 ratio 7.180722891566265
        GenericResourceAwareStrategy, FirstBlock 240.0, LastBlock 1796.0 ratio 7.483333333333333
        ConstraintSolverStrategy, FirstBlock 328.0, LastBlock 2544.0 ratio 7.7560975609756095

        TestLargeFragmentedClusterScheduling took 93106 ms
        DefaultResourceAwareStrategy, FirstBlock 258.0, LastBlock 1714.0 ratio 6.6434108527131785
        GenericResourceAwareStrategy, FirstBlock 215.0, LastBlock 1692.0 ratio 7.869767441860465
        ConstraintSolverStrategy, FirstBlock 309.0, LastBlock 2342.0 ratio 7.5792880258899675

        Choose the median value of the values above
        DefaultResourceAwareStrategy    6.96
        GenericResourceAwareStrategy    7.78
        ConstraintSolverStrategy        7.75
        */
        final int numNodes = 500;
        final int numRuns = 5;

        Map<String, Config> strategyToConfigs = new HashMap<>();
        // AcceptedBlockTimeRatios obtained by empirical testing (see comment block above)
        Map<String, Double> strategyToAcceptedBlockTimeRatios = new HashMap<>();
        for (Class strategyClass: strategyClasses) {
            strategyToConfigs.put(strategyClass.getName(), createClusterConfig(strategyClass, 10, 10, 0, null));
            strategyToAcceptedBlockTimeRatios.put(strategyClass.getName(), 6.96);
        }
        strategyToAcceptedBlockTimeRatios.put(DefaultResourceAwareStrategy.class.getName(), 6.96);
        strategyToAcceptedBlockTimeRatios.put(GenericResourceAwareStrategy.class.getName(), 7.78);
        strategyToConfigs.put(ConstraintSolverStrategy.class.getName(), createCSSClusterConfig(10, 10, 0, null));
        strategyToAcceptedBlockTimeRatios.put(ConstraintSolverStrategy.class.getName(), 7.75);

        Map<String, TimeBlockResult> strategyToTimeBlockResults = new HashMap<>();

        // Get first and last block times for multiple runs and strategies
        long startTime = Time.currentTimeMillis();
        for (Entry<String, Config> strategyConfig : strategyToConfigs.entrySet()) {
            TimeBlockResult strategyTimeBlockResult = strategyToTimeBlockResults.computeIfAbsent(strategyConfig.getKey(), (k) -> new TimeBlockResult());
            for (int run = 0; run < numRuns; ++run) {
                TimeBlockResult result = testLargeClusterSchedulingTiming(numNodes, strategyConfig.getValue());
                strategyTimeBlockResult.append(result);
            }
        }

        // Log median ratios for different strategies
        LOG.info("TestLargeFragmentedClusterScheduling took {} ms", Time.currentTimeMillis() - startTime);
        for (Entry<String, TimeBlockResult> strategyResult : strategyToTimeBlockResults.entrySet()) {
            TimeBlockResult strategyTimeBlockResult = strategyResult.getValue();
            double medianFirstBlockTime = getMedianValue(strategyTimeBlockResult.firstBlockTime);
            double medianLastBlockTime = getMedianValue(strategyTimeBlockResult.lastBlockTime);
            double ratio = medianLastBlockTime / medianFirstBlockTime;
            LOG.info("{}, FirstBlock {}, LastBlock {} ratio {}", strategyResult.getKey(), medianFirstBlockTime, medianLastBlockTime, ratio);
        }

        // Check last block scheduling time does not get significantly slower
        for (Entry<String, TimeBlockResult> strategyResult : strategyToTimeBlockResults.entrySet()) {
            TimeBlockResult strategyTimeBlockResult = strategyResult.getValue();
            double medianFirstBlockTime = getMedianValue(strategyTimeBlockResult.firstBlockTime);
            double medianLastBlockTime = getMedianValue(strategyTimeBlockResult.lastBlockTime);
            double ratio = medianLastBlockTime / medianFirstBlockTime;

            double slowSchedulingThreshold = 1.5;
            String msg = "Strategy " + strategyResult.getKey() + " scheduling is significantly slower for mostly full fragmented cluster\n";
            double ratioAccepted = strategyToAcceptedBlockTimeRatios.get(strategyResult.getKey());
            msg += String.format("Ratio was %.2f (high/low=%.2f/%.2f), max allowed is %.2f (%.2f * %.2f)",
                    ratio, medianLastBlockTime, medianFirstBlockTime,
                    ratioAccepted * slowSchedulingThreshold, ratioAccepted, slowSchedulingThreshold);
            assertTrue(ratio < slowSchedulingThreshold * ratioAccepted, msg);
        }
    }

    // Create multiple copies of a test topology
    private void addTopologyBlockToMap(Map<String, TopologyDetails> topologyMap, String baseName, Config config,
                                       double spoutMemoryLoad, int[] blockIndices) {
        TopologyBuilder builder = new TopologyBuilder();
        builder.setSpout("testSpout", new TestSpout(), 1).setMemoryLoad(spoutMemoryLoad);
        StormTopology stormTopology = builder.createTopology();
        Map<ExecutorDetails, String> executorMap = genExecsAndComps(stormTopology);

        for (int i = blockIndices[0]; i <= blockIndices[1]; ++i) {
            TopologyDetails topo = new TopologyDetails(baseName + i, config, stormTopology, 0, executorMap, 0, "user");
            topologyMap.put(topo.getId(), topo);
        }
    }

    /*
     * Test time to schedule large cluster scheduling with fragmentation
     */
    private TimeBlockResult testLargeClusterSchedulingTiming(int numNodes, Config config) {
        // Attempt to schedule multiple copies of 2 different topologies (topo-t0 and topo-t1) in 3 blocks.
        // Without fragmentation it is possible to schedule all topologies, but fragmentation causes topologies to not
        // schedule for the last block.

        // Get start/end indices for blocks
        int numTopologyPairs = numNodes;
        int increment = (int) Math.floor(numTopologyPairs * 0.1);
        int firstBlockIndices[] = {0, increment - 1};
        int midBlockIndices[] = {increment, numTopologyPairs - increment - 1};
        int lastBlockIndices[] = {numTopologyPairs - increment, numTopologyPairs - 1};

        // Memory is the constraining resource.
        double t0Mem = 70;   // memory required by topo-t0
        double t1Mem = 20;   // memory required by topo-t1
        double nodeMem = 100;

        // first block (0% - 10%)
        Map<String, TopologyDetails> topologyMap = new HashMap<>();
        addTopologyBlockToMap(topologyMap, "topo_t0-", config, t0Mem, firstBlockIndices);
        addTopologyBlockToMap(topologyMap, "topo_t1-", config, t1Mem, firstBlockIndices);
        Topologies topologies = new Topologies(topologyMap);

        Map<String, SupervisorDetails> supMap = genSupervisors(numNodes, 7, 3500, nodeMem);
        Cluster cluster = new Cluster(new INimbusTest(), new ResourceMetrics(new StormMetricsRegistry()), supMap, new HashMap<String, SchedulerAssignmentImpl>(), topologies, config);
        TimeBlockResult timeBlockResult = new TimeBlockResult();

        // schedule first block (0% - 10%)
        {
            scheduler = new ResourceAwareScheduler();
            scheduler.prepare(config, new StormMetricsRegistry());

            long time = Time.currentTimeMillis();
            scheduler.schedule(topologies, cluster);
            timeBlockResult.firstBlockTime.add(Time.currentTimeMillis() - time);
        }

        // schedule mid block (10% - 90%)
        {
            addTopologyBlockToMap(topologyMap, "topo_t0-", config, t0Mem, midBlockIndices);
            addTopologyBlockToMap(topologyMap, "topo_t1-", config, t1Mem, midBlockIndices);

            topologies = new Topologies(topologyMap);
            cluster = new Cluster(cluster, topologies);

            scheduler.schedule(topologies, cluster);
        }

        // schedule last block (90% to 100%)
        {
            addTopologyBlockToMap(topologyMap, "topo_t0-", config, t0Mem, lastBlockIndices);
            addTopologyBlockToMap(topologyMap, "topo_t1-", config, t1Mem, lastBlockIndices);

            topologies = new Topologies(topologyMap);
            cluster = new Cluster(cluster, topologies);

            long time = Time.currentTimeMillis();
            scheduler.schedule(topologies, cluster);
            timeBlockResult.lastBlockTime.add(Time.currentTimeMillis() - time);
        }

        return timeBlockResult;
    }

    /**
     * Test multiple spouts and cyclic topologies
     */
    @Test
    public void testMultipleSpoutsAndCyclicTopologies() {
        for (Class strategyClass: strategyClasses) {
            String strategyClassName = strategyClass.getName();

            TopologyBuilder builder = new TopologyBuilder();

            builder.setSpout("spout-1", new TestSpout(),
                    5);
            builder.setSpout("spout-2", new TestSpout(),
                    5);
            builder.setBolt("bolt-1", new TestBolt(),
                    5).shuffleGrouping("spout-1").shuffleGrouping("bolt-3");
            builder.setBolt("bolt-2", new TestBolt(),
                    5).shuffleGrouping("bolt-1");
            builder.setBolt("bolt-3", new TestBolt(),
                    5).shuffleGrouping("bolt-2").shuffleGrouping("spout-2");

            INimbus iNimbus = new INimbusTest();
            Map<String, SupervisorDetails> supMap = genSupervisors(25, 1, 100, 1000);
            Config config = createClusterConfig(strategyClass, 100, 500, 500, null);

            StormTopology stormTopology = builder.createTopology();
            config.put(Config.TOPOLOGY_SUBMITTER_USER, "jerry");
            TopologyDetails topo = new TopologyDetails("topo-1", config, stormTopology,
                    0, genExecsAndComps(stormTopology), 0, "jerry");

            Topologies topologies = new Topologies(topo);
            Cluster cluster = new Cluster(iNimbus, new ResourceMetrics(new StormMetricsRegistry()), supMap, new HashMap<String, SchedulerAssignmentImpl>(), topologies, config);
            scheduler = new ResourceAwareScheduler();
            scheduler.prepare(config, new StormMetricsRegistry());
            scheduler.schedule(topologies, cluster);

            assertTrue(cluster.getAssignmentById(topo.getId()) != null, "Topo scheduled?");
            assertEquals(25, cluster.getAssignmentById(topo.getId()).getExecutorToSlot().size(), "Topo all executors scheduled?");
        }
    }

    @Test
    public void testSchedulerStrategyWhitelist() {
        Map<String, Object> config = ConfigUtils.readStormConfig();
        for (Class strategyClass: strategyClasses) {
            String strategyClassName = strategyClass.getName();
            String allowed = strategyClassName;
            config.put(Config.NIMBUS_SCHEDULER_STRATEGY_CLASS_WHITELIST, Arrays.asList(allowed));

            Object sched = ReflectionUtils.newSchedulerStrategyInstance(allowed, config);
            assertEquals(sched.getClass().getName(), allowed);
        }
    }

    @Test
    public void testSchedulerStrategyWhitelistException() {
        Map<String, Object> config = ConfigUtils.readStormConfig();
        String allowed = "org.apache.storm.scheduler.resource.strategies.scheduling.SomeNonExistantStrategy";
        for (Class strategyClass: strategyClasses) {
            String strategyClassName = strategyClass.getName();
            String notAllowed = strategyClassName;
            config.put(Config.NIMBUS_SCHEDULER_STRATEGY_CLASS_WHITELIST, Arrays.asList(allowed));

            Assertions.assertThrows(DisallowedStrategyException.class, () -> ReflectionUtils.newSchedulerStrategyInstance(notAllowed, config));
        }
    }

    @Test
    public void testSchedulerStrategyEmptyWhitelist() {
        Map<String, Object> config = ConfigUtils.readStormConfig();
        for (Class strategyClass: strategyClasses) {
            String strategyClassName = strategyClass.getName();
            String allowed = strategyClassName;
            Object sched = ReflectionUtils.newSchedulerStrategyInstance(allowed, config);
            assertEquals(sched.getClass().getName(), allowed);
        }
    }

    @PerformanceTest
    @Test
    public void testLargeTopologiesOnLargeClusters() {
        for (Class strategyClass: strategyClasses) {
            String strategyClassName = strategyClass.getName();
            Assertions.assertTimeoutPreemptively(Duration.ofSeconds(30),
                    () -> testLargeTopologiesCommon(strategyClassName, false, 1));
        }
        
    }
    
    @PerformanceTest
    @Test
    public void testLargeTopologiesOnLargeClustersGras() {
        Assertions.assertTimeoutPreemptively(Duration.ofSeconds(75),
                () -> testLargeTopologiesCommon(GenericResourceAwareStrategy.class.getName(), true, 1));
    }

    @PerformanceTest
    @Test
    public void testLargeTopologiesOnLargeClustersRoundRobin() {
        Assertions.assertTimeoutPreemptively(Duration.ofSeconds(30),
                () -> testLargeTopologiesCommon(RoundRobinResourceAwareStrategy.class.getName(), true, 1));
    }

    public static class NeverEndingSchedulingStrategy extends BaseResourceAwareStrategy {

        @Override
        public void prepare(Map<String, Object> config) {
        }

        @Override
        public SchedulingResult schedule(Cluster schedulingState, TopologyDetails td) {
            while (true) {
                if (Thread.currentThread().isInterrupted()) {
                    LOG.info("scheduling interrupted");
                    return null;
                }
            }
        }
    }

    @Test
    public void testStrategyTakingTooLong() {
        for (Class strategyClass: strategyClasses) {
            INimbus iNimbus = new INimbusTest();
            Map<String, SupervisorDetails> supMap = genSupervisors(8, 4, 100, 1000);
            Config config = createClusterConfig(strategyClass, 100, 500, 500, null);
            List<String> allowedSchedulerStrategies = new ArrayList<>();
            allowedSchedulerStrategies.add(DefaultResourceAwareStrategy.class.getName());
            allowedSchedulerStrategies.add(DefaultResourceAwareStrategyOld.class.getName());
            allowedSchedulerStrategies.add(GenericResourceAwareStrategy.class.getName());
            allowedSchedulerStrategies.add(GenericResourceAwareStrategyOld.class.getName());
            allowedSchedulerStrategies.add(RoundRobinResourceAwareStrategy.class.getName());
            allowedSchedulerStrategies.add(NeverEndingSchedulingStrategy.class.getName());
            config.put(Config.NIMBUS_SCHEDULER_STRATEGY_CLASS_WHITELIST, allowedSchedulerStrategies);
            config.put(DaemonConfig.SCHEDULING_TIMEOUT_SECONDS_PER_TOPOLOGY, 30);

            TopologyDetails topo1 = genTopology("topo-1", config, 1, 0, 2, 0, currentTime - 2, 10, "jerry");
            TopologyDetails topo3 = genTopology("topo-3", config, 1, 2, 1, 1, currentTime - 2, 20, "jerry");

            config.put(Config.TOPOLOGY_SCHEDULER_STRATEGY, NeverEndingSchedulingStrategy.class.getName());
            TopologyDetails topo2 = genTopology("topo-2", config, 2, 0, 2, 0, currentTime - 2, 20, "jerry");

            Topologies topologies = new Topologies(topo1, topo2, topo3);
            Cluster cluster = new Cluster(iNimbus, new ResourceMetrics(new StormMetricsRegistry()), supMap, new HashMap<>(), topologies, config);
            scheduler = new ResourceAwareScheduler();

            scheduler.prepare(config, new StormMetricsRegistry());
            scheduler.schedule(topologies, cluster);

            assertFalse(cluster.needsSchedulingRas(topo1));
            assertTrue(cluster.needsSchedulingRas(topo2));
            assertFalse(cluster.needsSchedulingRas(topo3));

            assertTrue(cluster.getAssignmentById(topo1.getId()) != null, "Topo-1 scheduled?");
            assertEquals(2, cluster.getAssignmentById(topo1.getId()).getExecutorToSlot().size(), "Topo-1 all executors scheduled?");
            assertTrue(cluster.getAssignmentById(topo2.getId()) == null, "Topo-2 not scheduled");
            assertEquals("Scheduling took too long for " + topo2.getId() + " using strategy "
                    + NeverEndingSchedulingStrategy.class.getName()
                    + " timeout after 30 seconds using config scheduling.timeout.seconds.per.topology.", cluster.getStatusMap().get(topo2.getId()));
            assertTrue(cluster.getAssignmentById(topo3.getId()) != null, "Topo-3 scheduled?");
            assertEquals(3, cluster.getAssignmentById(topo3.getId()).getExecutorToSlot().size(), "Topo-3 all executors scheduled?");
        }
    }

    public void testLargeTopologiesCommon(final String strategy, final boolean includeGpu, final int multiplier) {
        for (Class strategyClass: strategyClasses) {
            INimbus iNimbus = new INimbusTest();
            Map<String, SupervisorDetails> supMap = genSupervisorsWithRacks(25 * multiplier, 40, 66, 3 * multiplier, 0, 4700, 226200, new HashMap<>());
            if (includeGpu) {
                HashMap<String, Double> extraResources = new HashMap<>();
                extraResources.put("my.gpu", 1.0);
                supMap.putAll(genSupervisorsWithRacks(3 * multiplier, 40, 66, 0, 0, 4700, 226200, extraResources));
            }

            Config config = new Config();
            config.putAll(createClusterConfig(strategyClass, 88, 775, 25, null));
            config.put(Config.TOPOLOGY_SCHEDULER_STRATEGY, strategy);

            scheduler = new ResourceAwareScheduler();

            Map<String, TopologyDetails> topologyDetailsMap = new HashMap<>();
            for (int i = 0; i < 11 * multiplier; i++) {
                TopologyDetails td = genTopology(String.format("topology-%05d", i), config, 5,
                        40, 30, 114, 0, 0, "user", 8192);
                topologyDetailsMap.put(td.getId(), td);
            }
            if (includeGpu) {
                for (int i = 0; i < multiplier; i++) {
                    TopologyBuilder builder = topologyBuilder(5, 40, 30, 114);
                    builder.setBolt("gpu-bolt", new TestBolt(), 40)
                            .addResource("my.gpu", 1.0)
                            .shuffleGrouping("spout-0");
                    TopologyDetails td = topoToTopologyDetails(String.format("topology-gpu-%05d", i), config, builder.createTopology(), 0, 0,
                            "user", 8192);
                    topologyDetailsMap.put(td.getId(), td);
                }
            }
            Topologies topologies = new Topologies(topologyDetailsMap);
            Cluster cluster = new Cluster(iNimbus, new ResourceMetrics(new StormMetricsRegistry()), supMap, new HashMap<>(), topologies, config);

            long startTime = Time.currentTimeMillis();
            scheduler.prepare(config, new StormMetricsRegistry());
            scheduler.schedule(topologies, cluster);
            long schedulingDuration = Time.currentTimeMillis() - startTime;
            LOG.info("Scheduling took " + schedulingDuration + " ms");
            LOG.info("HAS {} SLOTS USED", cluster.getUsedSlots().size());

            Map<String, SchedulerAssignment> assignments = new TreeMap<>(cluster.getAssignments());

            for (Entry<String, SchedulerAssignment> entry : assignments.entrySet()) {
                SchedulerAssignment sa = entry.getValue();
                Map<String, AtomicLong> slotsPerRack = new TreeMap<>();
                for (WorkerSlot slot : sa.getSlots()) {
                    String nodeId = slot.getNodeId();
                    String rack = supervisorIdToRackName(nodeId);
                    slotsPerRack.computeIfAbsent(rack, (r) -> new AtomicLong(0)).incrementAndGet();
                }
                LOG.info("{} => {}", entry.getKey(), slotsPerRack);
            }
        }
    }

    public static void main(String[] args) {
        String strategy = DefaultResourceAwareStrategy.class.getName();
        if (args.length > 0) {
            strategy = args[0];
        }
        boolean includeGpu = false;
        if (args.length > 1) {
            includeGpu = Boolean.valueOf(args[1]);
        }
        int multiplier = 1;
        if (args.length > 2) {
            multiplier = Integer.valueOf(args[2]);
        }
        TestResourceAwareScheduler trs = new TestResourceAwareScheduler();
        trs.testLargeTopologiesCommon(strategy, includeGpu, multiplier);
        System.exit(0);
    }
}