/*
 * 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.storm.scheduler;

import java.util.ArrayList;
import java.util.Collection;
import java.util.Collections;
import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Map.Entry;
import java.util.Set;
import java.util.stream.Collectors;
import org.apache.storm.Config;
import org.apache.storm.Constants;
import org.apache.storm.DaemonConfig;
import org.apache.storm.daemon.nimbus.TopologyResources;
import org.apache.storm.generated.SharedMemory;
import org.apache.storm.generated.WorkerResources;
import org.apache.storm.networktopography.DNSToSwitchMapping;
import org.apache.storm.networktopography.DefaultRackDNSToSwitchMapping;
import org.apache.storm.scheduler.resource.normalization.NormalizedResourceOffer;
import org.apache.storm.scheduler.resource.normalization.NormalizedResourceRequest;
import org.apache.storm.scheduler.resource.normalization.NormalizedResources;
import org.apache.storm.scheduler.resource.normalization.ResourceMetrics;
import org.apache.storm.shade.com.google.common.annotations.VisibleForTesting;
import org.apache.storm.utils.ConfigUtils;
import org.apache.storm.utils.ObjectReader;
import org.apache.storm.utils.ReflectionUtils;
import org.apache.storm.utils.Utils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * The current state of the storm cluster.  Cluster is not currently thread safe.
 */
public class Cluster implements ISchedulingState {
    private static final Logger LOG = LoggerFactory.getLogger(Cluster.class);

    /**
     * key: supervisor id, value: supervisor details.
     */
    private final Map<String, SupervisorDetails> supervisors = new HashMap<>();
    /**
     * key: rack, value: nodes in that rack.
     */
    private final Map<String, List<String>> networkTopography = new HashMap<>();
    /**
     * key: topologyId, value: topology's current assignments.
     */
    private final Map<String, SchedulerAssignmentImpl> assignments = new HashMap<>();
    /**
     * key topologyId, Value: scheduler's status.
     */
    private final Map<String, String> status = new HashMap<>();
    /**
     * A map from hostname to supervisor ids.
     */
    private final Map<String, List<String>> hostToId = new HashMap<>();
    private final Map<String, Object> conf;
    private final Topologies topologies;
    private final Map<String, Map<WorkerSlot, NormalizedResourceRequest>> nodeToScheduledResourcesCache;
    private final Map<String, Map<String, Double>> nodeToScheduledOffHeapNodeMemoryCache;   // node -> topologyId -> double
    private final Map<String, Set<WorkerSlot>> nodeToUsedSlotsCache;
    private final Map<String, NormalizedResourceRequest> totalResourcesPerNodeCache = new HashMap<>();
    private final ResourceMetrics resourceMetrics;
    private SchedulerAssignmentImpl assignment;
    private Set<String> blackListedHosts = new HashSet<>();
    private INimbus inimbus;
    private double minWorkerCpu = 0.0;

    private static <K, V> Map<K, V> makeMap(String key) {
        return new HashMap<>();
    }

    private static <K> Set<K> makeSet(String key) {
        return new HashSet<>();
    }

    public Cluster(
        INimbus nimbus,
        ResourceMetrics resourceMetrics,
        Map<String, SupervisorDetails> supervisors,
        Map<String, ? extends SchedulerAssignment> map,
        Topologies topologies,
        Map<String, Object> conf) {
        this(nimbus, resourceMetrics, supervisors, map, topologies, conf, null, null, null);
    }

    /**
     * Copy constructor.
     */
    public Cluster(Cluster src) {
        this(
            src.inimbus,
            src.resourceMetrics,
            src.supervisors,
            src.assignments,
            src.topologies,
            new HashMap<>(src.conf),
            src.status,
            src.blackListedHosts,
            src.networkTopography);
    }

    /**
     * Testing Constructor that takes an existing cluster and replaces the topologies in it.
     *
     * @param src        the original cluster
     * @param topologies the new topolgoies to use
     */
    @VisibleForTesting
    public Cluster(Cluster src, Topologies topologies) {
        this(
            src.inimbus,
            src.resourceMetrics,
            src.supervisors,
            src.assignments,
            topologies,
            new HashMap<>(src.conf),
            src.status,
            src.blackListedHosts,
            src.networkTopography);
    }

    private Cluster(
        INimbus nimbus,
        ResourceMetrics resourceMetrics,
        Map<String, SupervisorDetails> supervisors,
        Map<String, ? extends SchedulerAssignment> assignments,
        Topologies topologies,
        Map<String, Object> conf,
        Map<String, String> status,
        Set<String> blackListedHosts,
        Map<String, List<String>> networkTopography) {
        this.inimbus = nimbus;
        this.resourceMetrics = resourceMetrics;
        this.supervisors.putAll(supervisors);
        this.nodeToScheduledResourcesCache = new HashMap<>(this.supervisors.size());
        this.nodeToScheduledOffHeapNodeMemoryCache = new HashMap<>();
        this.nodeToUsedSlotsCache = new HashMap<>(this.supervisors.size());

        for (Map.Entry<String, SupervisorDetails> entry : supervisors.entrySet()) {
            String nodeId = entry.getKey();
            SupervisorDetails supervisor = entry.getValue();
            String host = supervisor.getHost();
            List<String> ids = hostToId.computeIfAbsent(host, k -> new ArrayList<>());
            ids.add(nodeId);
        }
        this.conf = conf;
        this.topologies = topologies;
        this.minWorkerCpu = ObjectReader.getDouble(conf.get(DaemonConfig.STORM_WORKER_MIN_CPU_PCORE_PERCENT), 0.0);

        ArrayList<String> supervisorHostNames = new ArrayList<>();
        for (SupervisorDetails s : supervisors.values()) {
            supervisorHostNames.add(s.getHost());
        }

        //Initialize the network topography
        if (networkTopography == null || networkTopography.isEmpty()) {
            //Initialize the network topography
            String clazz = (String) conf.get(Config.STORM_NETWORK_TOPOGRAPHY_PLUGIN);
            if (clazz == null || clazz.isEmpty()) {
                clazz = DefaultRackDNSToSwitchMapping.class.getName();
            }
            DNSToSwitchMapping topographyMapper = ReflectionUtils.newInstance(clazz);

            Map<String, String> resolvedSuperVisors = topographyMapper.resolve(supervisorHostNames);
            for (Map.Entry<String, String> entry : resolvedSuperVisors.entrySet()) {
                String hostName = entry.getKey();
                String rack = entry.getValue();
                List<String> nodesForRack = this.networkTopography.computeIfAbsent(rack, k -> new ArrayList<>());
                nodesForRack.add(hostName);
            }
        } else {
            this.networkTopography.putAll(networkTopography);
        }

        if (status != null) {
            this.status.putAll(status);
        }

        if (blackListedHosts != null) {
            this.blackListedHosts.addAll(blackListedHosts);
        }
        setAssignments(assignments, true);
    }

    /**
     * Get heap memory usage for a worker's main process and logwriter process.
     *
     * @param topConf - the topology config
     * @return the assigned memory (in MB)
     */
    public static double getAssignedMemoryForSlot(final Map<String, Object> topConf) {
        double totalWorkerMemory = 0.0;
        final Integer topologyWorkerDefaultMemoryAllocation = 768;

        List<String> topologyWorkerGcChildopts = ConfigUtils.getValueAsList(
            Config.TOPOLOGY_WORKER_GC_CHILDOPTS, topConf);
        List<String> workerGcChildopts = ConfigUtils.getValueAsList(
            Config.WORKER_GC_CHILDOPTS, topConf);
        Double memGcChildopts = null;
        memGcChildopts = Utils.parseJvmHeapMemByChildOpts(
            topologyWorkerGcChildopts, null);
        if (memGcChildopts == null) {
            memGcChildopts = Utils.parseJvmHeapMemByChildOpts(
                workerGcChildopts, null);
        }

        List<String> topologyWorkerChildopts = ConfigUtils.getValueAsList(
            Config.TOPOLOGY_WORKER_CHILDOPTS, topConf);
        Double memTopologyWorkerChildopts = Utils.parseJvmHeapMemByChildOpts(
            topologyWorkerChildopts, null);

        List<String> workerChildopts = ConfigUtils.getValueAsList(
            Config.WORKER_CHILDOPTS, topConf);
        Double memWorkerChildopts = Utils.parseJvmHeapMemByChildOpts(
            workerChildopts, null);

        if (memGcChildopts != null) {
            totalWorkerMemory += memGcChildopts;
        } else if (memTopologyWorkerChildopts != null) {
            totalWorkerMemory += memTopologyWorkerChildopts;
        } else if (memWorkerChildopts != null) {
            totalWorkerMemory += memWorkerChildopts;
        } else {
            Object workerHeapMemoryMb = topConf.get(
                Config.WORKER_HEAP_MEMORY_MB);
            totalWorkerMemory += ObjectReader.getInt(
                workerHeapMemoryMb, topologyWorkerDefaultMemoryAllocation);
        }

        List<String> topoWorkerLwChildopts = ConfigUtils.getValueAsList(
            Config.TOPOLOGY_WORKER_LOGWRITER_CHILDOPTS, topConf);
        if (topoWorkerLwChildopts != null) {
            totalWorkerMemory += Utils.parseJvmHeapMemByChildOpts(
                topoWorkerLwChildopts, 0.0);
        }
        return totalWorkerMemory;
    }

    /**
     * Check if the given topology is allowed for modification right now. If not throw an IllegalArgumentException else go on.
     *
     * @param topologyId the id of the topology to check
     */
    protected void assertValidTopologyForModification(String topologyId) {
        //NOOP
    }

    @Override
    public Topologies getTopologies() {
        return topologies;
    }

    @Override
    public Set<String> getBlacklistedHosts() {
        return blackListedHosts;
    }

    /**
     * Set the list of hosts that are blacklisted.
     *
     * @param hosts the new hosts that are blacklisted.
     */
    public void setBlacklistedHosts(Set<String> hosts) {
        if (hosts == blackListedHosts) {
            //NOOP
            return;
        }
        blackListedHosts.clear();
        blackListedHosts.addAll(hosts);
    }

    public void blacklistHost(String host) {
        blackListedHosts.add(host);
    }

    @Override
    public boolean isBlackListed(String supervisorId) {
        return blackListedHosts.contains(getHost(supervisorId));
    }

    @Override
    public boolean isBlacklistedHost(String host) {
        return blackListedHosts.contains(host);
    }

    @Override
    public String getHost(String supervisorId) {
        return inimbus.getHostName(supervisors, supervisorId);
    }

    @Override
    public List<TopologyDetails> needsSchedulingTopologies() {
        List<TopologyDetails> ret = new ArrayList<>();
        for (TopologyDetails topology : getTopologies()) {
            if (needsScheduling(topology)) {
                ret.add(topology);
            }
        }

        return ret;
    }

    @Override
    public boolean needsScheduling(TopologyDetails topology) {
        int desiredNumWorkers = topology.getNumWorkers();
        int assignedNumWorkers = this.getAssignedNumWorkers(topology);
        return desiredNumWorkers > assignedNumWorkers || getUnassignedExecutors(topology).size() > 0;
    }

    @Override
    public boolean needsSchedulingRas(TopologyDetails topology) {
        return getUnassignedExecutors(topology).size() > 0;
    }

    @Override
    public Map<ExecutorDetails, String> getNeedsSchedulingExecutorToComponents(
        TopologyDetails topology) {
        Collection<ExecutorDetails> allExecutors = new HashSet<>(topology.getExecutors());

        SchedulerAssignment assignment = assignments.get(topology.getId());
        if (assignment != null) {
            allExecutors.removeAll(assignment.getExecutors());
        }
        return topology.selectExecutorToComponent(allExecutors);
    }

    @Override
    public Map<String, List<ExecutorDetails>> getNeedsSchedulingComponentToExecutors(
        TopologyDetails topology) {
        Map<ExecutorDetails, String> executorToComponents =
            getNeedsSchedulingExecutorToComponents(topology);
        Map<String, List<ExecutorDetails>> componentToExecutors = new HashMap<>();
        for (Map.Entry<ExecutorDetails, String> entry : executorToComponents.entrySet()) {
            ExecutorDetails executor = entry.getKey();
            String component = entry.getValue();
            if (!componentToExecutors.containsKey(component)) {
                componentToExecutors.put(component, new ArrayList<>());
            }

            componentToExecutors.get(component).add(executor);
        }

        return componentToExecutors;
    }

    @Override
    public Set<Integer> getUsedPorts(SupervisorDetails supervisor) {
        return nodeToUsedSlotsCache.computeIfAbsent(supervisor.getId(), Cluster::makeSet)
            .stream()
            .map(WorkerSlot::getPort)
            .collect(Collectors.toSet());
    }

    @Override
    public Set<Integer> getAvailablePorts(SupervisorDetails supervisor) {
        Set<Integer> usedPorts = this.getUsedPorts(supervisor);

        Set<Integer> ret = new HashSet<>();
        ret.addAll(getAssignablePorts(supervisor));
        ret.removeAll(usedPorts);

        return ret;
    }

    @Override
    public Set<Integer> getAssignablePorts(SupervisorDetails supervisor) {
        if (isBlackListed(supervisor.getId())) {
            return Collections.emptySet();
        }
        return supervisor.getAllPorts();
    }

    @Override
    public List<WorkerSlot> getNonBlacklistedAvailableSlots(List<String> blacklistedSupervisorIds) {
        List<WorkerSlot> slots = new ArrayList<>();
        for (SupervisorDetails supervisor : this.supervisors.values()) {
            if (!isBlackListed(supervisor.getId()) && !blacklistedSupervisorIds.contains(supervisor.getId())) {
                slots.addAll(getAvailableSlots(supervisor));
            }
        }

        return slots;
    }

    @Override
    public List<WorkerSlot> getAvailableSlots() {
        List<WorkerSlot> slots = new ArrayList<>();
        for (SupervisorDetails supervisor : this.supervisors.values()) {
            slots.addAll(this.getAvailableSlots(supervisor));
        }

        return slots;
    }

    @Override
    public List<WorkerSlot> getAvailableSlots(SupervisorDetails supervisor) {
        Set<Integer> ports = this.getAvailablePorts(supervisor);
        List<WorkerSlot> slots = new ArrayList<>(ports.size());

        for (Integer port : ports) {
            slots.add(new WorkerSlot(supervisor.getId(), port));
        }

        return slots;
    }

    @Override
    public List<WorkerSlot> getAssignableSlots(SupervisorDetails supervisor) {
        Set<Integer> ports = this.getAssignablePorts(supervisor);
        List<WorkerSlot> slots = new ArrayList<>(ports.size());

        for (Integer port : ports) {
            slots.add(new WorkerSlot(supervisor.getId(), port));
        }

        return slots;
    }

    @Override
    public List<WorkerSlot> getAssignableSlots() {
        List<WorkerSlot> slots = new ArrayList<>();
        for (SupervisorDetails supervisor : this.supervisors.values()) {
            slots.addAll(this.getAssignableSlots(supervisor));
        }

        return slots;
    }

    @Override
    public Collection<ExecutorDetails> getUnassignedExecutors(TopologyDetails topology) {
        if (topology == null) {
            return new ArrayList<>(0);
        }

        Collection<ExecutorDetails> ret = new HashSet<>(topology.getExecutors());

        SchedulerAssignment assignment = getAssignmentById(topology.getId());
        if (assignment != null) {
            Set<ExecutorDetails> assignedExecutors = assignment.getExecutors();
            ret.removeAll(assignedExecutors);
        }

        return ret;
    }

    @Override
    public int getAssignedNumWorkers(TopologyDetails topology) {
        SchedulerAssignment assignment =
            topology != null ? this.getAssignmentById(topology.getId()) : null;
        if (assignment == null) {
            return 0;
        }

        Set<WorkerSlot> slots = new HashSet<>();
        slots.addAll(assignment.getExecutorToSlot().values());
        return slots.size();
    }

    @Override
    public NormalizedResourceOffer getAvailableResources(SupervisorDetails sd) {
        NormalizedResourceOffer ret = new NormalizedResourceOffer(sd.getTotalResources());
        for (SchedulerAssignment assignment: assignments.values()) {
            for (Entry<WorkerSlot, WorkerResources> entry: assignment.getScheduledResources().entrySet()) {
                if (sd.getId().equals(entry.getKey().getNodeId())) {
                    ret.remove(entry.getValue(), getResourceMetrics());
                }
            }
        }
        return ret;
    }

    private void addResource(Map<String, Double> resourceMap, String resourceName, Double valueToBeAdded) {
        if (!resourceMap.containsKey(resourceName)) {
            resourceMap.put(resourceName, 0.0);
        }
        Double currentPresent = resourceMap.get(resourceName);
        resourceMap.put(resourceName, currentPresent + valueToBeAdded);
    }

    private WorkerResources calculateWorkerResources(
        TopologyDetails td, Collection<ExecutorDetails> executors) {
        NormalizedResourceRequest totalResources = new NormalizedResourceRequest();
        Map<String, Double> sharedTotalResources = new HashMap<>();
        for (ExecutorDetails exec : executors) {
            NormalizedResourceRequest allResources = td.getTotalResources(exec);
            if (allResources == null) {
                continue;
            }
            totalResources.add(allResources);
        }
        for (SharedMemory shared : td.getSharedMemoryRequests(executors)) {
            totalResources.addOffHeap(shared.get_off_heap_worker());
            totalResources.addOnHeap(shared.get_on_heap());

            addResource(
                sharedTotalResources,
                Constants.COMMON_OFFHEAP_MEMORY_RESOURCE_NAME, shared.get_off_heap_worker()
            );
            addResource(
                sharedTotalResources,
                Constants.COMMON_ONHEAP_MEMORY_RESOURCE_NAME, shared.get_on_heap()
            );
        }
        sharedTotalResources = NormalizedResources.RESOURCE_NAME_NORMALIZER.normalizedResourceMap(sharedTotalResources);

        Map<String, Double> totalResourcesMap = totalResources.toNormalizedMap();
        Double cpu = totalResources.getTotalCpu();
        if (cpu < minWorkerCpu) {
            cpu = minWorkerCpu;
            totalResourcesMap.put(Constants.COMMON_CPU_RESOURCE_NAME, cpu);
        }

        WorkerResources ret = new WorkerResources();
        ret.set_resources(totalResourcesMap);
        ret.set_shared_resources(sharedTotalResources);
        ret.set_cpu(cpu);

        ret.set_mem_off_heap(totalResources.getOffHeapMemoryMb());
        ret.set_mem_on_heap(totalResources.getOnHeapMemoryMb());
        ret.set_shared_mem_off_heap(
            sharedTotalResources.getOrDefault(
                Constants.COMMON_OFFHEAP_MEMORY_RESOURCE_NAME, 0.0)
        );
        ret.set_shared_mem_on_heap(
            sharedTotalResources.getOrDefault(
                Constants.COMMON_ONHEAP_MEMORY_RESOURCE_NAME, 0.0)
        );
        return ret;
    }

    @Override
    public boolean wouldFit(
        WorkerSlot ws,
        ExecutorDetails exec,
        TopologyDetails td,
        NormalizedResourceOffer resourcesAvailable,
        double maxHeap) {

        NormalizedResourceRequest requestedResources = td.getTotalResources(exec);
        if (!resourcesAvailable.couldFit(minWorkerCpu, requestedResources)) {
            return false;
        }

        double currentTotal = 0.0;
        double currentCpuTotal = 0.0;

        Set<ExecutorDetails> wouldBeAssigned = new HashSet<>();
        wouldBeAssigned.add(exec);
        SchedulerAssignmentImpl assignment = assignments.get(td.getId());

        if (assignment != null) {
            Collection<ExecutorDetails> currentlyAssigned = assignment.getSlotToExecutors().get(ws);
            if (currentlyAssigned != null) {
                wouldBeAssigned.addAll(currentlyAssigned);
                WorkerResources wrCurrent = calculateWorkerResources(td, currentlyAssigned);
                currentTotal = wrCurrent.get_mem_off_heap() + wrCurrent.get_mem_on_heap();
                currentCpuTotal = wrCurrent.get_cpu();
            }

            currentTotal += calculateSharedOffHeapNodeMemory(ws.getNodeId(), assignment, td);
        }

        WorkerResources wrAfter = calculateWorkerResources(td, wouldBeAssigned);
        double afterTotal = wrAfter.get_mem_off_heap() + wrAfter.get_mem_on_heap();
        afterTotal += calculateSharedOffHeapNodeMemory(ws.getNodeId(), assignment, td, exec);

        double afterOnHeap = wrAfter.get_mem_on_heap();
        double afterCpuTotal = wrAfter.get_cpu();

        double cpuAdded = afterCpuTotal - currentCpuTotal;
        double cpuAvailable = resourcesAvailable.getTotalCpu();

        if (cpuAdded > cpuAvailable) {
            if (LOG.isTraceEnabled()) {
                LOG.trace("Could not schedule {}:{} on {} not enough CPU {} > {}",
                        td.getName(),
                        exec,
                        ws,
                        cpuAdded,
                        cpuAvailable);
            }
            return false;
        }

        double memoryAdded = afterTotal - currentTotal;
        double memoryAvailable = resourcesAvailable.getTotalMemoryMb();

        if (memoryAdded > memoryAvailable) {
            if (LOG.isTraceEnabled()) {
                LOG.trace("Could not schedule {}:{} on {} not enough Mem {} > {}",
                          td.getName(),
                          exec,
                          ws,
                          memoryAdded,
                          memoryAvailable);
            }
            return false;
        }
        if (afterOnHeap > maxHeap) {
            if (LOG.isTraceEnabled()) {
                LOG.trace("Could not schedule {}:{} on {} HEAP would be too large {} > {}",
                          td.getName(),
                          exec,
                          ws,
                          afterOnHeap,
                          maxHeap);
            }
            return false;
        }
        return true;
    }

    /**
     * Assign the slot to the executors for this topology.
     *
     * @throws RuntimeException if the specified slot is already occupied.
     */
    public void assign(WorkerSlot slot, String topologyId, Collection<ExecutorDetails> executors) {
        assertValidTopologyForModification(topologyId);
        if (isSlotOccupied(slot)) {
            throw new RuntimeException(
                "slot: [" + slot.getNodeId() + ", " + slot.getPort() + "] is already occupied.");
        }

        TopologyDetails td = topologies.getById(topologyId);
        if (td == null) {
            throw new IllegalArgumentException(
                "Trying to schedule for topo "
                + topologyId
                + " but that is not a known topology "
                + topologies.getAllIds());
        }
        WorkerResources resources = calculateWorkerResources(td, executors);
        SchedulerAssignmentImpl assignment = assignments.get(topologyId);
        if (assignment == null) {
            assignment = new SchedulerAssignmentImpl(topologyId);
            assignments.put(topologyId, assignment);
        } else {
            for (ExecutorDetails executor : executors) {
                if (assignment.isExecutorAssigned(executor)) {
                    throw new RuntimeException(
                        "Attempting to assign executor: "
                        + executor
                        + " of topology: "
                        + topologyId
                        + " to workerslot: "
                        + slot
                        + ". The executor is already assigned to workerslot: "
                        + assignment.getExecutorToSlot().get(executor)
                        + ". The executor must unassigned before it can be assigned to another slot!");
                }
            }
        }

        assignment.assign(slot, executors, resources);
        String nodeId = slot.getNodeId();
        double sharedOffHeapNodeMemory = calculateSharedOffHeapNodeMemory(nodeId, assignment, td);
        assignment.setTotalSharedOffHeapNodeMemory(nodeId, sharedOffHeapNodeMemory);
        updateCachesForWorkerSlot(slot, resources, topologyId, sharedOffHeapNodeMemory);
        totalResourcesPerNodeCache.remove(slot.getNodeId());
    }

    /**
     * Assign everything for the given topology.
     *
     * @param assignment the new assignment to make
     */
    public void assign(SchedulerAssignment assignment, boolean ignoreSingleExceptions) {
        String id = assignment.getTopologyId();
        assertValidTopologyForModification(id);
        Map<WorkerSlot, Collection<ExecutorDetails>> slotToExecs = assignment.getSlotToExecutors();
        for (Entry<WorkerSlot, Collection<ExecutorDetails>> entry : slotToExecs.entrySet()) {
            try {
                assign(entry.getKey(), id, entry.getValue());
            } catch (RuntimeException e) {
                if (!ignoreSingleExceptions) {
                    throw e;
                }
            }
        }
    }

    /**
     * Calculate the amount of shared off heap node memory on a given node with the given assignment.
     *
     * @param nodeId     the id of the node
     * @param assignment the current assignment
     * @param td         the topology details
     * @return the amount of shared off heap node memory for that node in MB
     */
    private double calculateSharedOffHeapNodeMemory(String nodeId, SchedulerAssignmentImpl assignment, TopologyDetails td) {
        return calculateSharedOffHeapNodeMemory(nodeId, assignment, td, null);
    }

    private double calculateSharedOffHeapNodeMemory(
        String nodeId, SchedulerAssignmentImpl assignment, TopologyDetails td, ExecutorDetails extra) {
        Set<ExecutorDetails> executorsOnNode = new HashSet<>();
        if (assignment != null) {
            for (Entry<WorkerSlot, Collection<ExecutorDetails>> entry : assignment.getSlotToExecutors().entrySet()) {
                if (nodeId.equals(entry.getKey().getNodeId())) {
                    executorsOnNode.addAll(entry.getValue());
                }
            }
        }
        if (extra != null) {
            executorsOnNode.add(extra);
        }
        //Now check for overlap on the node
        double memorySharedWithinNode = 0.0;
        for (SharedMemory shared : td.getSharedMemoryRequests(executorsOnNode)) {
            memorySharedWithinNode += shared.get_off_heap_node();
        }
        return memorySharedWithinNode;
    }

    /**
     * Free the specified slot.
     *
     * @param slot the slot to free
     */
    public void freeSlot(WorkerSlot slot) {
        // remove the slot from the existing assignments
        for (SchedulerAssignmentImpl assignment : assignments.values()) {
            if (assignment.isSlotOccupied(slot)) {
                assertValidTopologyForModification(assignment.getTopologyId());
                assignment.unassignBySlot(slot);
                String nodeId = slot.getNodeId();
                TopologyDetails td = topologies.getById(assignment.getTopologyId());
                assignment.setTotalSharedOffHeapNodeMemory(
                    nodeId, calculateSharedOffHeapNodeMemory(nodeId, assignment, td));
                nodeToScheduledResourcesCache.computeIfAbsent(nodeId, Cluster::makeMap).put(slot, new NormalizedResourceRequest());
                nodeToUsedSlotsCache.computeIfAbsent(nodeId, Cluster::makeSet).remove(slot);
            }
        }
        //Invalidate the cache as something on the node changed
        totalResourcesPerNodeCache.remove(slot.getNodeId());
    }

    /**
     * free the slots.
     *
     * @param slots multiple slots to free
     */
    public void freeSlots(Collection<WorkerSlot> slots) {
        if (slots != null) {
            for (WorkerSlot slot : slots) {
                freeSlot(slot);
            }
        }
    }

    @Override
    public boolean isSlotOccupied(WorkerSlot slot) {
        return nodeToUsedSlotsCache.computeIfAbsent(slot.getNodeId(), Cluster::makeSet).contains(slot);
    }

    @Override
    public SchedulerAssignment getAssignmentById(String topologyId) {
        if (assignments.containsKey(topologyId)) {
            return assignments.get(topologyId);
        }

        return null;
    }

    @Override
    public Collection<WorkerSlot> getUsedSlotsByTopologyId(String topologyId) {
        SchedulerAssignmentImpl assignment = assignments.get(topologyId);
        if (assignment == null) {
            return Collections.emptySet();
        }
        return assignment.getSlots();
    }

    @Override
    public SupervisorDetails getSupervisorById(String nodeId) {
        return supervisors.get(nodeId);
    }

    @Override
    public Collection<WorkerSlot> getUsedSlots() {
        return nodeToUsedSlotsCache.values().stream().flatMap(Set::stream).collect(Collectors.toSet());
    }

    @Override
    public List<SupervisorDetails> getSupervisorsByHost(String host) {
        List<String> nodeIds = this.hostToId.get(host);
        List<SupervisorDetails> ret = new ArrayList<>();

        if (nodeIds != null) {
            for (String nodeId : nodeIds) {
                ret.add(this.getSupervisorById(nodeId));
            }
        }

        return ret;
    }

    @Override
    public Map<String, SchedulerAssignment> getAssignments() {
        return new HashMap<>(assignments);
    }

    /**
     * Set assignments for cluster.
     */
    public void setAssignments(
        Map<String, ? extends SchedulerAssignment> newAssignments, boolean ignoreSingleExceptions) {
        if (newAssignments == assignments) {
            //NOOP
            return;
        }
        for (SchedulerAssignment assignment : newAssignments.values()) {
            assertValidTopologyForModification(assignment.getTopologyId());
        }
        for (SchedulerAssignment assignment : assignments.values()) {
            assertValidTopologyForModification(assignment.getTopologyId());
        }
        assignments.clear();
        totalResourcesPerNodeCache.clear();
        nodeToScheduledResourcesCache.values().forEach(Map::clear);
        nodeToUsedSlotsCache.values().forEach(Set::clear);
        for (SchedulerAssignment assignment : newAssignments.values()) {
            assign(assignment, ignoreSingleExceptions);
        }
    }

    @Override
    public Map<String, SupervisorDetails> getSupervisors() {
        return this.supervisors;
    }

    @Override
    public NormalizedResourceOffer getNonBlacklistedClusterAvailableResources(Collection<String> blacklistedSupervisorIds) {
        NormalizedResourceOffer available = new NormalizedResourceOffer();
        for (SupervisorDetails sup : supervisors.values()) {
            if (!isBlackListed(sup.getId()) && !blacklistedSupervisorIds.contains(sup.getId())) {
                available.add(sup.getTotalResources());
                available.remove(getAllScheduledResourcesForNode(sup.getId()), getResourceMetrics());
            }
        }
        return available;
    }

    @Override
    public double getClusterTotalCpuResource() {
        double sum = 0.0;
        for (SupervisorDetails sup : supervisors.values()) {
            sum += sup.getTotalCpu();
        }
        return sum;
    }

    @Override
    public double getClusterTotalMemoryResource() {
        double sum = 0.0;
        for (SupervisorDetails sup : supervisors.values()) {
            sum += sup.getTotalMemory();
        }
        return sum;
    }

    @Override
    public Map<String, List<String>> getNetworkTopography() {
        return networkTopography;
    }

    @VisibleForTesting
    public void setNetworkTopography(Map<String, List<String>> networkTopography) {
        this.networkTopography.clear();
        this.networkTopography.putAll(networkTopography);
    }

    /**
     * set scheduler status for a topology.
     */
    public void setStatus(TopologyDetails td, String statusMessage) {
        setStatus(td.getId(), statusMessage);
    }

    /**
     * set scheduler status for a topology.
     */
    public void setStatus(String topologyId, String statusMessage) {
        assertValidTopologyForModification(topologyId);
        LOG.info("STATUS - {} {}", topologyId, statusMessage);
        status.put(topologyId, statusMessage);
    }

    public void setStatusIfAbsent(String topologyId, String statusMessage) {
        assertValidTopologyForModification(topologyId);
        status.putIfAbsent(topologyId, statusMessage);
    }

    @Override
    public Map<String, String> getStatusMap() {
        return status;
    }

    /**
     * set scheduler status map.
     */
    public void setStatusMap(Map<String, String> statusMap) {
        if (statusMap == this.status) {
            return; //This is a NOOP
        }
        for (String topologyId : statusMap.keySet()) {
            assertValidTopologyForModification(topologyId);
        }
        for (String topologyId : status.keySet()) {
            assertValidTopologyForModification(topologyId);
        }
        this.status.clear();
        this.status.putAll(statusMap);
    }

    public String getStatus(String topoId) {
        return status.get(topoId);
    }

    @Override
    public Map<String, TopologyResources> getTopologyResourcesMap() {
        Map<String, TopologyResources> ret = new HashMap<>(assignments.size());
        for (TopologyDetails td : topologies.getTopologies()) {
            String topoId = td.getId();
            SchedulerAssignmentImpl assignment = assignments.get(topoId);
            ret.put(topoId, new TopologyResources(td, assignment));
        }
        return ret;
    }

    @Override
    public Map<String, SupervisorResources> getSupervisorsResourcesMap() {
        Map<String, SupervisorResources> ret = new HashMap<>();
        for (SupervisorDetails sd : supervisors.values()) {
            ret.put(sd.getId(), new SupervisorResources(sd.getTotalMemory(), sd.getTotalCpu(), sd.getTotalGenericResources(),
                0, 0, new HashMap<>()));
        }
        for (SchedulerAssignmentImpl assignment : assignments.values()) {
            for (Entry<WorkerSlot, WorkerResources> entry :
                assignment.getScheduledResources().entrySet()) {
                String id = entry.getKey().getNodeId();
                SupervisorResources sr = ret.get(id);
                if (sr == null) {
                    sr = new SupervisorResources(0, 0, new HashMap<>(),
                        0, 0, new HashMap<>());

                }
                sr = sr.add(entry.getValue());
                ret.put(id, sr);
            }
            Map<String, Double> nodeIdToSharedOffHeap = assignment.getNodeIdToTotalSharedOffHeapNodeMemory();
            if (nodeIdToSharedOffHeap != null) {
                for (Entry<String, Double> entry : nodeIdToSharedOffHeap.entrySet()) {
                    String id = entry.getKey();
                    SupervisorResources sr = ret.get(id);
                    if (sr == null) {
                        sr = new SupervisorResources(0, 0, new HashMap<>(),
                            0, 0, new HashMap<>());
                    }
                    sr = sr.addMem(entry.getValue());
                    ret.put(id, sr);
                }
            }
        }
        return ret;
    }

    @Override
    public Map<String, Map<WorkerSlot, WorkerResources>> getWorkerResourcesMap() {
        HashMap<String, Map<WorkerSlot, WorkerResources>> ret = new HashMap<>();
        for (Entry<String, SchedulerAssignmentImpl> entry : assignments.entrySet()) {
            ret.put(entry.getKey(), entry.getValue().getScheduledResources());
        }
        return ret;
    }

    @Override
    public WorkerResources getWorkerResources(WorkerSlot ws) {
        WorkerResources ret = null;
        for (SchedulerAssignmentImpl assignment : assignments.values()) {
            ret = assignment.getScheduledResources().get(ws);
            if (ret != null) {
                break;
            }
        }
        return ret;
    }

    /**
     * This method updates ScheduledResources and UsedSlots cache for given workerSlot.
     */
    private void updateCachesForWorkerSlot(WorkerSlot workerSlot, WorkerResources workerResources, String topologyId,
                                           Double sharedOffHeapNodeMemory) {
        String nodeId = workerSlot.getNodeId();
        NormalizedResourceRequest normalizedResourceRequest = new NormalizedResourceRequest();
        normalizedResourceRequest.add(workerResources);
        nodeToScheduledResourcesCache.computeIfAbsent(nodeId, Cluster::makeMap).put(workerSlot, normalizedResourceRequest);
        nodeToScheduledOffHeapNodeMemoryCache.computeIfAbsent(nodeId, Cluster::makeMap).put(topologyId, sharedOffHeapNodeMemory);
        nodeToUsedSlotsCache.computeIfAbsent(nodeId, Cluster::makeSet).add(workerSlot);
    }

    public ResourceMetrics getResourceMetrics() {
        return resourceMetrics;
    }

    @Override
    public NormalizedResourceRequest getAllScheduledResourcesForNode(String nodeId) {
        return totalResourcesPerNodeCache.computeIfAbsent(nodeId, (nid) -> {
            // executor resources
            NormalizedResourceRequest totalScheduledResources = new NormalizedResourceRequest();
            for (NormalizedResourceRequest req : nodeToScheduledResourcesCache.computeIfAbsent(nodeId, Cluster::makeMap).values()) {
                totalScheduledResources.add(req);
            }
            // shared off heap node memory
            for (Double offHeapNodeMemory
                    : nodeToScheduledOffHeapNodeMemoryCache.computeIfAbsent(nid, Cluster::makeMap).values()) {
                totalScheduledResources.addOffHeap(offHeapNodeMemory);
            }

            return totalScheduledResources;
        });
    }

    @Override
    public double getScheduledMemoryForNode(String nodeId) {
        return getAllScheduledResourcesForNode(nodeId).getTotalMemoryMb();
    }

    @Override
    public double getScheduledCpuForNode(String nodeId) {
        return getAllScheduledResourcesForNode(nodeId).getTotalCpu();
    }

    public INimbus getINimbus() {
        return this.inimbus;
    }

    @Override
    public Map<String, Object> getConf() {
        return this.conf;
    }

    /**
     * Unassign everything for the given topology id.
     *
     * @param topoId the is of the topology to unassign
     */
    public void unassign(String topoId) {
        assertValidTopologyForModification(topoId);
        freeSlots(getUsedSlotsByTopologyId(topoId));
    }

    /**
     * Update the assignments and status from the other cluster.
     *
     * @param other the cluster to get the assignments and status from
     */
    public void updateFrom(Cluster other) {
        for (SchedulerAssignment assignment : other.getAssignments().values()) {
            assertValidTopologyForModification(assignment.getTopologyId());
        }
        setAssignments(other.getAssignments(), false);
        setStatusMap(other.getStatusMap());
    }

    public double getMinWorkerCpu() {
        return minWorkerCpu;
    }
}