@Override
    public void prepare(ExecutorDetails exec) {
        this.exec = exec;
    }
@Test
    public void testPreferRackWithTopoExecutors() {
        INimbus iNimbus = new INimbusTest();
        double compPcore = 100;
        double compOnHeap = 775;
        double compOffHeap = 25;
        int topo1NumSpouts = 1;
        int topo1NumBolts = 5;
        int topo1SpoutParallelism = 100;
        int topo1BoltParallelism = 200;
        int topo2NumSpouts = 1;
        int topo2NumBolts = 5;
        int topo2SpoutParallelism = 10;
        int topo2BoltParallelism = 20;
        final int numRacks = 3;
        final int numSupersPerRack = 10;
        final int numPortsPerSuper = 6;
        final int numZonesPerHost = 1;
        final double numaResourceMultiplier = 1.0;
        int rackStartNum = 0;
        int supStartNum = 0;
        long compPerRack = (topo1NumSpouts * topo1SpoutParallelism + topo1NumBolts * topo1BoltParallelism
                + topo2NumSpouts * topo2SpoutParallelism); // enough for topo1 but not topo1+topo2
        long compPerSuper =  compPerRack / numSupersPerRack;
        double cpuPerSuper = compPcore * compPerSuper;
        double memPerSuper = (compOnHeap + compOffHeap) * compPerSuper;
        double topo1MaxHeapSize = memPerSuper;
        double topo2MaxHeapSize = memPerSuper;
        final String topoName1 = "topology1";
        final String topoName2 = "topology2";

        Map<String, SupervisorDetails> supMap = genSupervisorsWithRacksAndNuma(
                numRacks, numSupersPerRack, numZonesPerHost, numPortsPerSuper, rackStartNum, supStartNum,
                cpuPerSuper, memPerSuper, Collections.emptyMap(), numaResourceMultiplier);
        TestDNSToSwitchMapping testDNSToSwitchMapping = new TestDNSToSwitchMapping(supMap.values());

        Config config = new Config();
        config.putAll(createGrasClusterConfig(compPcore, compOnHeap, compOffHeap, null, null));
        config.put(Config.TOPOLOGY_SCHEDULER_STRATEGY, GenericResourceAwareStrategy.class.getName());

        IScheduler scheduler = new ResourceAwareScheduler();
        scheduler.prepare(config, new StormMetricsRegistry());

        TopologyDetails td1 = genTopology(topoName1, config, topo1NumSpouts,
                topo1NumBolts, topo1SpoutParallelism, topo1BoltParallelism, 0, 0, "user", topo1MaxHeapSize);

        //Schedule the topo1 topology and ensure it fits on 1 rack
        Topologies topologies = new Topologies(td1);
        Cluster cluster = new Cluster(iNimbus, new ResourceMetrics(new StormMetricsRegistry()), supMap, new HashMap<>(), topologies, config);
        cluster.setNetworkTopography(testDNSToSwitchMapping.getRackToHosts());

        scheduler.schedule(topologies, cluster);
        Set<String> assignedRacks = cluster.getAssignedRacks(td1.getId());
        assertEquals(1, assignedRacks.size(), "Racks for topology=" + td1.getId() + " is " + assignedRacks);

        TopologyBuilder builder = topologyBuilder(topo2NumSpouts, topo2NumBolts, topo2SpoutParallelism, topo2BoltParallelism);
        TopologyDetails td2 = topoToTopologyDetails(topoName2, config, builder.createTopology(), 0, 0,"user", topo2MaxHeapSize);

        //Now schedule GPU but with the simple topology in place.
        topologies = new Topologies(td1, td2);
        cluster = new Cluster(cluster, topologies);
        scheduler.schedule(topologies, cluster);

        assignedRacks = cluster.getAssignedRacks(td1.getId(), td2.getId());
        assertEquals(2, assignedRacks.size(), "Racks for topologies=" + td1.getId() + "/" + td2.getId() + " is " + assignedRacks);

        // topo2 gets scheduled on its own rack because it is empty and available
        assignedRacks = cluster.getAssignedRacks(td2.getId());
        assertEquals(1, assignedRacks.size(), "Racks for topologies=" + td2.getId() + " is " + assignedRacks);

        // now unassign topo2, expect only one rack to be in use; free some slots and reschedule topo1 some topo1 executors
        cluster.unassign(td2.getId());
        assignedRacks = cluster.getAssignedRacks(td2.getId());
        assertEquals(0, assignedRacks.size(),
            "After unassigning topology " + td2.getId() + ", racks for topology=" + td2.getId() + " is " + assignedRacks);
        assignedRacks = cluster.getAssignedRacks(td1.getId());
        assertEquals(1, assignedRacks.size(),
            "After unassigning topology " + td2.getId() + ", racks for topology=" + td1.getId() + " is " + assignedRacks);
        assertFalse(cluster.needsSchedulingRas(td1),
            "Topology " + td1.getId() + " should be fully assigned before freeing slots");
        freeSomeWorkerSlots(cluster);
        assertTrue(cluster.needsSchedulingRas(td1),
            "Topology " + td1.getId() + " should need scheduling after freeing slots");

        // then reschedule executors
        scheduler.schedule(topologies, cluster);

        // only one rack should be in use by topology1
        assignedRacks = cluster.getAssignedRacks(td1.getId());
        assertEquals(1, assignedRacks.size(),
            "After reassigning topology " + td2.getId() + ", racks for topology=" + td1.getId() + " is " + assignedRacks);
    }