/*
 * 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.ambari.server.state.cluster;

import static java.util.stream.Collectors.toList;

import java.text.MessageFormat;
import java.util.ArrayList;
import java.util.Collection;
import java.util.Collections;
import java.util.Date;
import java.util.HashMap;
import java.util.HashSet;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.Map.Entry;
import java.util.Objects;
import java.util.Set;
import java.util.TreeMap;
import java.util.TreeSet;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.ConcurrentSkipListMap;
import java.util.concurrent.CopyOnWriteArrayList;
import java.util.concurrent.locks.ReadWriteLock;
import java.util.function.Function;
import java.util.stream.Collectors;

import javax.annotation.Nullable;

import jakarta.persistence.EntityManager;
import jakarta.persistence.RollbackException;

import org.apache.ambari.annotations.Experimental;
import org.apache.ambari.annotations.ExperimentalFeature;
import org.apache.ambari.server.AmbariException;
import org.apache.ambari.server.ConfigGroupNotFoundException;
import org.apache.ambari.server.ObjectNotFoundException;
import org.apache.ambari.server.ParentObjectNotFoundException;
import org.apache.ambari.server.RoleCommand;
import org.apache.ambari.server.ServiceComponentHostNotFoundException;
import org.apache.ambari.server.ServiceComponentNotFoundException;
import org.apache.ambari.server.ServiceNotFoundException;
import org.apache.ambari.server.agent.ExecutionCommand.KeyNames;
import org.apache.ambari.server.agent.stomp.HostLevelParamsHolder;
import org.apache.ambari.server.api.services.AmbariMetaInfo;
import org.apache.ambari.server.controller.AmbariManagementController;
import org.apache.ambari.server.controller.AmbariSessionManager;
import org.apache.ambari.server.controller.ClusterResponse;
import org.apache.ambari.server.controller.ConfigurationResponse;
import org.apache.ambari.server.controller.MaintenanceStateHelper;
import org.apache.ambari.server.controller.RootService;
import org.apache.ambari.server.controller.ServiceConfigVersionResponse;
import org.apache.ambari.server.controller.internal.BlueprintConfigurationProcessor;
import org.apache.ambari.server.controller.internal.DeleteHostComponentStatusMetaData;
import org.apache.ambari.server.events.AmbariEvent.AmbariEventType;
import org.apache.ambari.server.events.ClusterConfigChangedEvent;
import org.apache.ambari.server.events.ClusterEvent;
import org.apache.ambari.server.events.ClusterProvisionedEvent;
import org.apache.ambari.server.events.ConfigsUpdateEvent;
import org.apache.ambari.server.events.jpa.EntityManagerCacheInvalidationEvent;
import org.apache.ambari.server.events.jpa.JPAEvent;
import org.apache.ambari.server.events.publishers.AmbariEventPublisher;
import org.apache.ambari.server.events.publishers.JPAEventPublisher;
import org.apache.ambari.server.events.publishers.STOMPUpdatePublisher;
import org.apache.ambari.server.logging.LockFactory;
import org.apache.ambari.server.metadata.RoleCommandOrder;
import org.apache.ambari.server.metadata.RoleCommandOrderProvider;
import org.apache.ambari.server.orm.RequiresSession;
import org.apache.ambari.server.orm.cache.HostConfigMapping;
import org.apache.ambari.server.orm.dao.AlertDefinitionDAO;
import org.apache.ambari.server.orm.dao.AlertDispatchDAO;
import org.apache.ambari.server.orm.dao.ClusterDAO;
import org.apache.ambari.server.orm.dao.ClusterStateDAO;
import org.apache.ambari.server.orm.dao.HostComponentDesiredStateDAO;
import org.apache.ambari.server.orm.dao.HostConfigMappingDAO;
import org.apache.ambari.server.orm.dao.HostDAO;
import org.apache.ambari.server.orm.dao.HostVersionDAO;
import org.apache.ambari.server.orm.dao.ServiceConfigDAO;
import org.apache.ambari.server.orm.dao.StackDAO;
import org.apache.ambari.server.orm.dao.TopologyRequestDAO;
import org.apache.ambari.server.orm.dao.UpgradeDAO;
import org.apache.ambari.server.orm.entities.ClusterConfigEntity;
import org.apache.ambari.server.orm.entities.ClusterEntity;
import org.apache.ambari.server.orm.entities.ClusterServiceEntity;
import org.apache.ambari.server.orm.entities.ClusterStateEntity;
import org.apache.ambari.server.orm.entities.ConfigGroupEntity;
import org.apache.ambari.server.orm.entities.HostComponentDesiredStateEntity;
import org.apache.ambari.server.orm.entities.HostEntity;
import org.apache.ambari.server.orm.entities.HostVersionEntity;
import org.apache.ambari.server.orm.entities.PermissionEntity;
import org.apache.ambari.server.orm.entities.PrivilegeEntity;
import org.apache.ambari.server.orm.entities.RepositoryVersionEntity;
import org.apache.ambari.server.orm.entities.RequestScheduleEntity;
import org.apache.ambari.server.orm.entities.ResourceEntity;
import org.apache.ambari.server.orm.entities.ServiceConfigEntity;
import org.apache.ambari.server.orm.entities.StackEntity;
import org.apache.ambari.server.orm.entities.TopologyRequestEntity;
import org.apache.ambari.server.orm.entities.UpgradeEntity;
import org.apache.ambari.server.stack.upgrade.orchestrate.UpgradeContext;
import org.apache.ambari.server.stack.upgrade.orchestrate.UpgradeContextFactory;
import org.apache.ambari.server.state.BlueprintProvisioningState;
import org.apache.ambari.server.state.Cluster;
import org.apache.ambari.server.state.ClusterHealthReport;
import org.apache.ambari.server.state.Clusters;
import org.apache.ambari.server.state.Config;
import org.apache.ambari.server.state.ConfigFactory;
import org.apache.ambari.server.state.ConfigHelper;
import org.apache.ambari.server.state.DesiredConfig;
import org.apache.ambari.server.state.Host;
import org.apache.ambari.server.state.HostHealthStatus;
import org.apache.ambari.server.state.MaintenanceState;
import org.apache.ambari.server.state.PropertyInfo;
import org.apache.ambari.server.state.RepositoryVersionState;
import org.apache.ambari.server.state.SecurityType;
import org.apache.ambari.server.state.Service;
import org.apache.ambari.server.state.ServiceComponent;
import org.apache.ambari.server.state.ServiceComponentHost;
import org.apache.ambari.server.state.ServiceComponentHostEvent;
import org.apache.ambari.server.state.ServiceComponentHostEventType;
import org.apache.ambari.server.state.ServiceFactory;
import org.apache.ambari.server.state.ServiceInfo;
import org.apache.ambari.server.state.StackId;
import org.apache.ambari.server.state.StackInfo;
import org.apache.ambari.server.state.State;
import org.apache.ambari.server.state.configgroup.ConfigGroup;
import org.apache.ambari.server.state.configgroup.ConfigGroupFactory;
import org.apache.ambari.server.state.fsm.InvalidStateTransitionException;
import org.apache.ambari.server.state.repository.ClusterVersionSummary;
import org.apache.ambari.server.state.repository.VersionDefinitionXml;
import org.apache.ambari.server.state.scheduler.RequestExecution;
import org.apache.ambari.server.state.scheduler.RequestExecutionFactory;
import org.apache.ambari.server.topology.STOMPComponentsDeleteHandler;
import org.apache.ambari.server.topology.TopologyRequest;
import org.apache.ambari.spi.ClusterInformation;
import org.apache.ambari.spi.RepositoryType;
import org.apache.ambari.spi.RepositoryVersion;
import org.apache.commons.collections4.CollectionUtils;
import org.apache.commons.collections4.MapUtils;
import org.apache.commons.lang.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.google.common.base.Functions;
import com.google.common.base.Predicate;
import com.google.common.collect.HashMultimap;
import com.google.common.collect.ImmutableMap;
import com.google.common.collect.Iterables;
import com.google.common.collect.ListMultimap;
import com.google.common.collect.Maps;
import com.google.common.collect.Multimap;
import com.google.common.collect.Sets;
import com.google.common.eventbus.Subscribe;
import com.google.inject.Inject;
import com.google.inject.Injector;
import com.google.inject.assistedinject.Assisted;
import com.google.inject.persist.Transactional;

public class ClusterImpl implements Cluster {

  private static final Logger LOG = LoggerFactory.getLogger(ClusterImpl.class);
  private static final Logger configChangeLog = LoggerFactory.getLogger("configchange");

  /**
   * Prefix for cluster session attributes name.
   */
  private static final String CLUSTER_SESSION_ATTRIBUTES_PREFIX = "cluster_session_attributes:";

  @Inject
  private Clusters clusters;

  private StackId desiredStackVersion;

  private final ConcurrentSkipListMap<String, Service> services = new ConcurrentSkipListMap<>();

  /**
   * [ Config Type -> [ Config Version Tag -> Config ] ]
   */
  private final ConcurrentMap<String, ConcurrentMap<String, Config>> allConfigs = new ConcurrentHashMap<>();

  /**
   * [ ServiceName -> [ ServiceComponentName -> [ HostName -> [ ... ] ] ] ]
   */
  private final ConcurrentMap<String, ConcurrentMap<String, ConcurrentMap<String, ServiceComponentHost>>> serviceComponentHosts = new ConcurrentHashMap<>();

  /**
   * [ HostName -> [ ... ] ]
   */
  private final ConcurrentMap<String, List<ServiceComponentHost>> serviceComponentHostsByHost = new ConcurrentHashMap<>();

  /**
   * Map of existing config groups
   */
  private final Map<Long, ConfigGroup> clusterConfigGroups = new ConcurrentHashMap<>();

  /**
   * Map of Request schedules for this cluster
   */
  private final Map<Long, RequestExecution> requestExecutions = new ConcurrentHashMap<>();

  private final ReadWriteLock clusterGlobalLock;

  /**
   * The unique ID of the {@link @ClusterEntity}.
   */
  private final long clusterId;

  private String clusterName;

  @Inject
  private ClusterDAO clusterDAO;

  @Inject
  private ClusterStateDAO clusterStateDAO;

  @Inject
  private HostDAO hostDAO;

  @Inject
  private HostVersionDAO hostVersionDAO;

  @Inject
  private ServiceFactory serviceFactory;

  @Inject
  private ConfigFactory configFactory;

  @Inject
  private LockFactory lockFactory;

  @Inject
  private HostConfigMappingDAO hostConfigMappingDAO;

  @Inject
  private ConfigGroupFactory configGroupFactory;

  @Inject
  private RequestExecutionFactory requestExecutionFactory;

  @Inject
  private ConfigHelper configHelper;

  @Inject
  private MaintenanceStateHelper maintenanceStateHelper;

  @Inject
  private AmbariMetaInfo ambariMetaInfo;

  @Inject
  private AmbariManagementController controller;

  @Inject
  private ServiceConfigDAO serviceConfigDAO;

  @Inject
  private AlertDefinitionDAO alertDefinitionDAO;

  @Inject
  private AlertDispatchDAO alertDispatchDAO;

  @Inject
  private UpgradeDAO upgradeDAO;

  @Inject
  private AmbariSessionManager sessionManager;

  @Inject
  private TopologyRequestDAO topologyRequestDAO;

  @Inject
  private STOMPComponentsDeleteHandler STOMPComponentsDeleteHandler;

  @Inject
  private HostLevelParamsHolder hostLevelParamsHolder;

  /**
   * Data access object used for looking up stacks from the database.
   */
  @Inject
  private StackDAO stackDAO;

  private volatile Multimap<String, String> serviceConfigTypes;

  /**
   * Used to publish events relating to cluster CRUD operations and to receive
   * information about cluster operations.
   */
  private AmbariEventPublisher eventPublisher;

  /**
   * Used for broadcasting {@link JPAEvent}s.
   */
  @Inject
  private JPAEventPublisher jpaEventPublisher;

  /**
   * Used for getting instances of {@link RoleCommand} for this cluster.
   */
  @Inject
  private RoleCommandOrderProvider roleCommandOrderProvider;

  /**
   * Used to create instances of {@link UpgradeContext} with injected
   * dependencies. The {@link UpgradeContext} is used to populate the command
   * with upgrade information on the command/role maps if the upgrade is
   * suspended.
   */
  @Inject
  private UpgradeContextFactory upgradeContextFactory;

  @Inject
  private STOMPUpdatePublisher STOMPUpdatePublisher;

  @Inject
  private HostComponentDesiredStateDAO hostComponentDesiredStateDAO;

  /**
   * A simple cache for looking up {@code cluster-env} properties for a cluster.
   * This map is changed whenever {{cluster-env}} is changed and we receive a
   * {@link ClusterConfigChangedEvent}.
   */
  private Map<String, String> m_clusterPropertyCache = new ConcurrentHashMap<>();

  @Inject
  public ClusterImpl(@Assisted ClusterEntity clusterEntity, Injector injector,
      AmbariEventPublisher eventPublisher)
      throws AmbariException {

    clusterId = clusterEntity.getClusterId();
    clusterName = clusterEntity.getClusterName();

    injector.injectMembers(this);

    clusterGlobalLock = lockFactory.newReadWriteLock("clusterGlobalLock");

    loadStackVersion();
    loadServices();
    loadServiceHostComponents();

    // cache configurations before loading configuration groups
    cacheConfigurations();
    loadConfigGroups();

    loadRequestExecutions();

    if (desiredStackVersion != null && !StringUtils.isEmpty(desiredStackVersion.getStackName()) && !
      StringUtils.isEmpty(desiredStackVersion.getStackVersion())) {
      loadServiceConfigTypes();
    }

    // register to receive stuff
    eventPublisher.register(this);
    this.eventPublisher = eventPublisher;
  }

  private void loadServiceConfigTypes() throws AmbariException {
    try {
      serviceConfigTypes = collectServiceConfigTypesMapping();
    } catch (AmbariException e) {
      LOG.error("Cannot load stack info:", e);
      throw e;
    }
    LOG.info("Service config types loaded: {}", serviceConfigTypes);
  }

  /**
   * Construct config type to service name mapping
   * @throws AmbariException when stack or its part not found
   */
  private Multimap<String, String> collectServiceConfigTypesMapping() throws AmbariException {
    Multimap<String, String> serviceConfigTypes = HashMultimap.create();

    Map<String, ServiceInfo> serviceInfoMap = null;
    try {
      serviceInfoMap = ambariMetaInfo.getServices(desiredStackVersion.getStackName(), desiredStackVersion.getStackVersion());
    } catch (ParentObjectNotFoundException e) {
      LOG.error("Service config versioning disabled due to exception: ", e);
      return serviceConfigTypes;
    }
    for (Entry<String, ServiceInfo> entry : serviceInfoMap.entrySet()) {
      String serviceName = entry.getKey();
      ServiceInfo serviceInfo = entry.getValue();
      Set<String> configTypes = serviceInfo.getConfigTypeAttributes().keySet();
      for (String configType : configTypes) {
        serviceConfigTypes.put(serviceName, configType);
      }
    }

    return serviceConfigTypes;
  }

  /**
   * Make sure we load all the service host components.
   * We need this for live status checks.
   */
  private void loadServiceHostComponents() {
    for (Entry<String, Service> serviceKV : services.entrySet()) {
      /* get all the service component hosts **/
      Service service = serviceKV.getValue();
      if (!serviceComponentHosts.containsKey(service.getName())) {
        serviceComponentHosts.put(service.getName(), new ConcurrentHashMap<>());
      }

      for (Entry<String, ServiceComponent> svcComponent : service.getServiceComponents().entrySet()) {
        ServiceComponent comp = svcComponent.getValue();
        String componentName = svcComponent.getKey();
        if (!serviceComponentHosts.get(service.getName()).containsKey(componentName)) {
          serviceComponentHosts.get(service.getName()).put(componentName, new ConcurrentHashMap<>());
        }

        // Get Service Host Components
        for (Entry<String, ServiceComponentHost> svchost : comp.getServiceComponentHosts().entrySet()) {
          String hostname = svchost.getKey();
          ServiceComponentHost svcHostComponent = svchost.getValue();
          if (!serviceComponentHostsByHost.containsKey(hostname)) {
            serviceComponentHostsByHost.put(hostname, new CopyOnWriteArrayList<>());
          }

          List<ServiceComponentHost> compList = serviceComponentHostsByHost.get(hostname);
          compList.add(svcHostComponent);

          if (!serviceComponentHosts.get(service.getName()).get(componentName).containsKey(
              hostname)) {
            serviceComponentHosts.get(service.getName()).get(componentName).put(hostname,
                svcHostComponent);
          }
        }
      }
    }
  }

  private void loadServices() {
    ClusterEntity clusterEntity = getClusterEntity();
    if (CollectionUtils.isEmpty(clusterEntity.getClusterServiceEntities())) {
      return;
    }

    for (ClusterServiceEntity serviceEntity : clusterEntity.getClusterServiceEntities()) {
      StackId stackId = getCurrentStackVersion();
      try {
        if (ambariMetaInfo.getService(stackId.getStackName(),
          stackId.getStackVersion(), serviceEntity.getServiceName()) != null) {
          services.put(serviceEntity.getServiceName(),
            serviceFactory.createExisting(this, serviceEntity));
        }
      } catch (AmbariException e) {
        LOG.error(String.format(
          "Can not get service info: stackName=%s, stackVersion=%s, serviceName=%s",
          stackId.getStackName(), stackId.getStackVersion(),
          serviceEntity.getServiceName()));
      }
    }
  }

  private void loadConfigGroups() {
    ClusterEntity clusterEntity = getClusterEntity();
    if (!clusterEntity.getConfigGroupEntities().isEmpty()) {
      for (ConfigGroupEntity configGroupEntity : clusterEntity.getConfigGroupEntities()) {
        clusterConfigGroups.put(configGroupEntity.getGroupId(),
            configGroupFactory.createExisting(this, configGroupEntity));
      }
    }
  }

  private void loadRequestExecutions() {
    ClusterEntity clusterEntity = getClusterEntity();
    if (!clusterEntity.getRequestScheduleEntities().isEmpty()) {
      for (RequestScheduleEntity scheduleEntity : clusterEntity.getRequestScheduleEntities()) {
        requestExecutions.put(scheduleEntity.getScheduleId(),
            requestExecutionFactory.createExisting(this, scheduleEntity));
      }
    }
  }

  @Override
  public void addConfigGroup(ConfigGroup configGroup) throws AmbariException {
    String hostList = "";
    if(LOG.isDebugEnabled()) {
      if (configGroup.getHosts() != null) {
        for (Host host : configGroup.getHosts().values()) {
          hostList += host.getHostName() + ", ";
        }
      }
    }

    LOG.debug("Adding a new Config group, clusterName = {}, groupName = {}, tag = {} with hosts {}",
      getClusterName(), configGroup.getName(), configGroup.getTag(), hostList);

    if (clusterConfigGroups.containsKey(configGroup.getId())) {
      // The loadConfigGroups will load all groups to memory
      LOG.debug("Config group already exists, clusterName = {}, groupName = {}, groupId = {}, tag = {}",
        getClusterName(), configGroup.getName(), configGroup.getId(), configGroup.getTag());
    } else {
      clusterConfigGroups.put(configGroup.getId(), configGroup);
    }
  }

  @Override
  public Map<Long, ConfigGroup> getConfigGroups() {
    return Collections.unmodifiableMap(clusterConfigGroups);
  }

  @Override
  public Map<Long, ConfigGroup> getConfigGroupsByHostname(String hostname)
    throws AmbariException {
    Map<Long, ConfigGroup> configGroups = new HashMap<>();

    for (Entry<Long, ConfigGroup> groupEntry : clusterConfigGroups.entrySet()) {
      Long id = groupEntry.getKey();
      ConfigGroup group = groupEntry.getValue();
      for (Host host : group.getHosts().values()) {
        if (StringUtils.equals(hostname, host.getHostName())) {
          configGroups.put(id, group);
          break;
        }
      }
    }
    return configGroups;
  }

  @Override
  public ConfigGroup getConfigGroupsById(Long configId) {
    return clusterConfigGroups.get(configId);
  }

  @Override
  public Map<Long, ConfigGroup> getConfigGroupsByServiceName(String serviceName) {
    Map<Long, ConfigGroup> configGroups = new HashMap<>();

    for (Entry<Long, ConfigGroup> groupEntry : clusterConfigGroups.entrySet()) {
      Long id = groupEntry.getKey();
      ConfigGroup group = groupEntry.getValue();
      if (StringUtils.equals(serviceName, group.getServiceName())) {
        configGroups.put(id, group);
      }
    }
    return configGroups;
  }

  @Override
  public void addRequestExecution(RequestExecution requestExecution) throws AmbariException {
    LOG.info("Adding a new request schedule" + ", clusterName = " + getClusterName() + ", id = "
        + requestExecution.getId() + ", description = " + requestExecution.getDescription());

    if (requestExecutions.containsKey(requestExecution.getId())) {
      LOG.debug("Request schedule already exists, clusterName = {}, id = {}, description = {}",
        getClusterName(), requestExecution.getId(), requestExecution.getDescription());
    } else {
      requestExecutions.put(requestExecution.getId(), requestExecution);
    }
  }

  @Override
  public Map<Long, RequestExecution> getAllRequestExecutions() {
    return Collections.unmodifiableMap(requestExecutions);
  }

  @Override
  public void deleteRequestExecution(Long id) throws AmbariException {
    RequestExecution requestExecution = requestExecutions.get(id);
    if (requestExecution == null) {
      throw new AmbariException("Request schedule does not exists, " + "id = " + id);
    }
    LOG.info("Deleting request schedule" + ", clusterName = " + getClusterName() + ", id = "
        + requestExecution.getId() + ", description = " + requestExecution.getDescription());

    requestExecution.delete();
    requestExecutions.remove(id);
  }

  @Override
  public void deleteConfigGroup(Long id) throws AmbariException {
    ConfigGroup configGroup = clusterConfigGroups.get(id);
    if (configGroup == null) {
      throw new ConfigGroupNotFoundException(getClusterName(), id.toString());
    }

    LOG.debug("Deleting Config group, clusterName = {}, groupName = {}, groupId = {}, tag = {}",
      getClusterName(), configGroup.getName(), configGroup.getId(), configGroup.getTag());

    configGroup.delete();
    clusterConfigGroups.remove(id);

    configHelper.updateAgentConfigs(Collections.singleton(configGroup.getClusterName()));
  }

  public ServiceComponentHost getServiceComponentHost(String serviceName,
      String serviceComponentName, String hostname) throws AmbariException {
    if (!serviceComponentHosts.containsKey(serviceName)
        || !serviceComponentHosts.get(serviceName).containsKey(
            serviceComponentName)
        || !serviceComponentHosts.get(serviceName).get(serviceComponentName).containsKey(
            hostname)) {
      throw new ServiceComponentHostNotFoundException(getClusterName(),
          serviceName, serviceComponentName, hostname);
    }
    return serviceComponentHosts.get(serviceName).get(serviceComponentName).get(hostname);
  }

  @Override
  public List<ServiceComponentHost> getServiceComponentHosts() {
    List<ServiceComponentHost> serviceComponentHosts = new ArrayList<>();
    if (!serviceComponentHostsByHost.isEmpty()) {
      for (List<ServiceComponentHost> schList : serviceComponentHostsByHost.values()) {
        serviceComponentHosts.addAll(schList);
      }
    }
    return Collections.unmodifiableList(serviceComponentHosts);
  }

  @Override
  public String getClusterName() {
    return clusterName;
  }

  @Override
  public void setClusterName(String clusterName) {
    String oldName = null;
    ClusterEntity clusterEntity = getClusterEntity();
    oldName = clusterEntity.getClusterName();
    clusterEntity.setClusterName(clusterName);

    // RollbackException possibility if UNIQUE constraint violated
    clusterEntity = clusterDAO.merge(clusterEntity);
    clusters.updateClusterName(oldName, clusterName);
    this.clusterName = clusterName;

    // if the name changed, fire an event
    if (!StringUtils.equals(oldName, clusterName)) {
      ClusterEvent clusterNameChangedEvent = new ClusterEvent(AmbariEventType.CLUSTER_RENAME, clusterId);
      eventPublisher.publish(clusterNameChangedEvent);
    }
  }

  @Override
  public Long getResourceId() {
    ClusterEntity clusterEntity = getClusterEntity();

    ResourceEntity resourceEntity = clusterEntity.getResource();
    if (resourceEntity == null) {
      LOG.warn(
          "There is no resource associated with this cluster:\n\tCluster Name: {}\n\tCluster ID: {}",
          getClusterName(), getClusterId());
      return null;
    } else {
      return resourceEntity.getId();
    }
  }

  @Override
  @Transactional
  public void addServiceComponentHosts(Collection<ServiceComponentHost> serviceComponentHosts) throws AmbariException {
    for (ServiceComponentHost serviceComponentHost : serviceComponentHosts) {
      Service service = getService(serviceComponentHost.getServiceName());
      ServiceComponent serviceComponent = service.getServiceComponent(serviceComponentHost.getServiceComponentName());
      serviceComponent.addServiceComponentHost(serviceComponentHost);
    }
  }

  public void addServiceComponentHost(ServiceComponentHost svcCompHost)
      throws AmbariException {
    if (LOG.isDebugEnabled()) {
      LOG.debug("Trying to add component {} of service {} on {} to the cache",
          svcCompHost.getServiceComponentName(), svcCompHost.getServiceName(),
          svcCompHost.getHostName());
    }

    final String hostname = svcCompHost.getHostName();
    final String serviceName = svcCompHost.getServiceName();
    final String componentName = svcCompHost.getServiceComponentName();

    Set<Cluster> cs = clusters.getClustersForHost(hostname);

    boolean clusterFound = false;
    Iterator<Cluster> iter = cs.iterator();
    while (iter.hasNext()) {
      Cluster c = iter.next();
      if (c.getClusterId() == getClusterId()) {
        clusterFound = true;
        break;
      }
    }

    if (!clusterFound) {
      throw new AmbariException("Host does not belong this cluster"
          + ", hostname=" + hostname + ", clusterName=" + getClusterName()
          + ", clusterId=" + getClusterId());
    }

    if (!serviceComponentHosts.containsKey(serviceName)) {
      serviceComponentHosts.put(serviceName, new ConcurrentHashMap<>());
    }

    if (!serviceComponentHosts.get(serviceName).containsKey(componentName)) {
      serviceComponentHosts.get(serviceName).put(componentName, new ConcurrentHashMap<>());
    }

    if (serviceComponentHosts.get(serviceName).get(componentName).containsKey(
        hostname)) {
      throw new AmbariException("Duplicate entry for ServiceComponentHost"
          + ", serviceName=" + serviceName + ", serviceComponentName"
          + componentName + ", hostname= " + hostname);
    }

    if (!serviceComponentHostsByHost.containsKey(hostname)) {
      serviceComponentHostsByHost.put(hostname, new CopyOnWriteArrayList<>());
    }

    if (LOG.isDebugEnabled()) {
      LOG.debug("Adding a new ServiceComponentHost, clusterName={}, clusterId={}, serviceName={}, serviceComponentName{}, hostname= {}",
        getClusterName(), getClusterId(), serviceName, componentName, hostname);
    }

    serviceComponentHosts.get(serviceName).get(componentName).put(hostname,
      svcCompHost);

    serviceComponentHostsByHost.get(hostname).add(svcCompHost);
  }

  @Override
  public void removeServiceComponentHost(ServiceComponentHost svcCompHost)
    throws AmbariException {
    if (LOG.isDebugEnabled()) {
      LOG.debug(
          "Trying to remove component {} of service {} on {} from the cache",
          svcCompHost.getServiceComponentName(), svcCompHost.getServiceName(),
          svcCompHost.getHostName());
    }

    final String hostname = svcCompHost.getHostName();
    final String serviceName = svcCompHost.getServiceName();
    final String componentName = svcCompHost.getServiceComponentName();
    Set<Cluster> cs = clusters.getClustersForHost(hostname);

    boolean clusterFound = false;
    Iterator<Cluster> iter = cs.iterator();
    while (iter.hasNext()) {
      Cluster c = iter.next();
      if (c.getClusterId() == getClusterId()) {
        clusterFound = true;
        break;
      }
    }

    if (!clusterFound) {
      throw new AmbariException("Host does not belong this cluster"
          + ", hostname=" + hostname + ", clusterName=" + getClusterName()
          + ", clusterId=" + getClusterId());
    }

    if (!serviceComponentHosts.containsKey(serviceName)
        || !serviceComponentHosts.get(serviceName).containsKey(componentName)
        || !serviceComponentHosts.get(serviceName).get(componentName).containsKey(
            hostname)) {
      throw new AmbariException("Invalid entry for ServiceComponentHost"
          + ", serviceName=" + serviceName + ", serviceComponentName"
          + componentName + ", hostname= " + hostname);
    }

    if (!serviceComponentHostsByHost.containsKey(hostname)) {
      throw new AmbariException("Invalid host entry for ServiceComponentHost"
          + ", serviceName=" + serviceName + ", serviceComponentName"
          + componentName + ", hostname= " + hostname);
    }

    ServiceComponentHost schToRemove = null;
    for (ServiceComponentHost sch : serviceComponentHostsByHost.get(hostname)) {
      if (sch.getServiceName().equals(serviceName)
          && sch.getServiceComponentName().equals(componentName)
          && sch.getHostName().equals(hostname)) {
        schToRemove = sch;
        break;
      }
    }

    if (schToRemove == null) {
      LOG.warn("Unavailable in per host cache. ServiceComponentHost"
        + ", serviceName=" + serviceName
        + ", serviceComponentName" + componentName
        + ", hostname= " + hostname);
    }

    if (LOG.isDebugEnabled()) {
      LOG.debug("Removing a ServiceComponentHost, clusterName={}, clusterId={}, serviceName={}, serviceComponentName{}, hostname= {}",
        getClusterName(), getClusterId(), serviceName, componentName, hostname);
    }

    serviceComponentHosts.get(serviceName).get(componentName).remove(hostname);
    if (schToRemove != null) {
      serviceComponentHostsByHost.get(hostname).remove(schToRemove);
    }
  }

  @Override
  public long getClusterId() {
    // Add cluster creates the managed entity before creating the Cluster
    // instance so id would not be null.
    return clusterId;
  }

  @Override
  public List<ServiceComponentHost> getServiceComponentHosts(String hostname) {
    List<ServiceComponentHost> serviceComponentHosts = serviceComponentHostsByHost.get(hostname);
    if (null != serviceComponentHosts) {
      return new CopyOnWriteArrayList<>(serviceComponentHosts);
    }

    return new ArrayList<>();
  }

  @Override
  public Map<String, Set<String>> getServiceComponentHostMap(Set<String> hostNames, Set<String> serviceNames) {
    Map<String, Set<String>> componentHostMap = new TreeMap<>();

    Collection<Host> hosts = getHosts();

    if(hosts != null) {
      for (Host host : hosts) {
        String hostname = host.getHostName();

        // If this host is not filtered out, continue processing
        if ((hostNames == null) || hostNames.contains(hostname)) {
          List<ServiceComponentHost> serviceComponentHosts = getServiceComponentHosts(hostname);

          if (serviceComponentHosts != null) {
            for (ServiceComponentHost sch : serviceComponentHosts) {
              // If the service for this ServiceComponentHost is not filtered out, continue processing
              if ((serviceNames == null) || serviceNames.contains(sch.getServiceName())) {
                String component = sch.getServiceComponentName();
                Set<String> componentHosts = componentHostMap.get(component);

                if (componentHosts == null) {
                  componentHosts = new TreeSet<>();
                  componentHostMap.put(component, componentHosts);
                }

                componentHosts.add(hostname);
              }
            }
          }
        }
      }
    }

    return componentHostMap;
  }

  @Override
