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

import java.util.ArrayList;
import java.util.Collection;
import java.util.Collections;
import java.util.HashMap;
import java.util.HashSet;
import java.util.Iterator;
import java.util.LinkedList;
import java.util.List;
import java.util.Map;
import java.util.Objects;
import java.util.Set;
import java.util.concurrent.Callable;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;

import jakarta.inject.Inject;

import org.apache.ambari.server.AmbariException;
import org.apache.ambari.server.actionmanager.HostRoleCommand;
import org.apache.ambari.server.actionmanager.HostRoleStatus;
import org.apache.ambari.server.api.services.stackadvisor.StackAdvisorBlueprintProcessor;
import org.apache.ambari.server.configuration.Configuration;
import org.apache.ambari.server.controller.AmbariServer;
import org.apache.ambari.server.controller.KerberosHelper;
import org.apache.ambari.server.controller.RequestStatusResponse;
import org.apache.ambari.server.controller.internal.ArtifactResourceProvider;
import org.apache.ambari.server.controller.internal.BaseClusterRequest;
import org.apache.ambari.server.controller.internal.CalculatedStatus;
import org.apache.ambari.server.controller.internal.ProvisionClusterRequest;
import org.apache.ambari.server.controller.internal.RequestImpl;
import org.apache.ambari.server.controller.internal.ScaleClusterRequest;
import org.apache.ambari.server.controller.internal.Stack;
import org.apache.ambari.server.controller.spi.NoSuchParentResourceException;
import org.apache.ambari.server.controller.spi.Request;
import org.apache.ambari.server.controller.spi.RequestStatus;
import org.apache.ambari.server.controller.spi.Resource;
import org.apache.ambari.server.controller.spi.ResourceAlreadyExistsException;
import org.apache.ambari.server.controller.spi.ResourceProvider;
import org.apache.ambari.server.controller.spi.SystemException;
import org.apache.ambari.server.controller.spi.UnsupportedPropertyException;
import org.apache.ambari.server.events.AmbariEvent;
import org.apache.ambari.server.events.ClusterConfigFinishedEvent;
import org.apache.ambari.server.events.ClusterProvisionStartedEvent;
import org.apache.ambari.server.events.ClusterProvisionedEvent;
import org.apache.ambari.server.events.HostsRemovedEvent;
import org.apache.ambari.server.events.RequestFinishedEvent;
import org.apache.ambari.server.events.publishers.AmbariEventPublisher;
import org.apache.ambari.server.orm.dao.HostRoleCommandStatusSummaryDTO;
import org.apache.ambari.server.orm.dao.SettingDAO;
import org.apache.ambari.server.orm.entities.SettingEntity;
import org.apache.ambari.server.orm.entities.StageEntity;
import org.apache.ambari.server.security.authorization.AuthorizationHelper;
import org.apache.ambari.server.serveraction.kerberos.KDCType;
import org.apache.ambari.server.state.Host;
import org.apache.ambari.server.state.SecurityType;
import org.apache.ambari.server.state.host.HostImpl;
import org.apache.ambari.server.state.quicklinksprofile.QuickLinksProfile;
import org.apache.ambari.server.topology.addservice.ResourceProviderAdapter;
import org.apache.ambari.server.topology.tasks.ConfigureClusterTask;
import org.apache.ambari.server.topology.tasks.ConfigureClusterTaskFactory;
import org.apache.ambari.server.topology.validators.TopologyValidatorService;
import org.apache.ambari.server.utils.ManagedThreadPoolExecutor;
import org.apache.ambari.server.utils.RetryHelper;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.google.common.collect.ImmutableMap;
import com.google.common.collect.ImmutableSet;
import com.google.common.eventbus.Subscribe;
import com.google.inject.Singleton;
import com.google.inject.persist.Transactional;

/**
 * Manages all cluster provisioning actions on the cluster topology.
 */
//todo: cluster isolation
@Singleton
public class TopologyManager {

  /**
   * internal token for topology related async tasks
   */
  public static final String INTERNAL_AUTH_TOKEN = "internal_topology_token";

  public static final String INITIAL_CONFIG_TAG = "INITIAL";
  public static final String TOPOLOGY_RESOLVED_TAG = "TOPOLOGY_RESOLVED";
  public static final String KDC_ADMIN_CREDENTIAL = "kdc.admin.credential";

  private PersistedState persistedState;

  /**
   * Single threaded executor to execute async tasks. At the moment it's only used to execute ConfigureClusterTask.
   */
  private final ExecutorService executor = Executors.newSingleThreadExecutor();

  /**
   * Thread pool size for topology task executors.
   */
  private int topologyTaskExecutorThreadPoolSize;
  /**
   * There is one ExecutorService for each cluster to execute TopologyTasks.
   * TopologyTasks are submitted into ExecutorService for each cluster,
   * however the ExecutorService is started only after cluster configuration is finished.
   */
  private final Map<Long, ManagedThreadPoolExecutor> topologyTaskExecutorServiceMap = new HashMap<>();

  private Collection<String> hostsToIgnore = new HashSet<>();
  private final List<HostImpl> availableHosts = new LinkedList<>();
  private final Map<String, LogicalRequest> reservedHosts = new HashMap<>();
  private final Map<Long, LogicalRequest> allRequests = new HashMap<>();
  // priority is given to oldest outstanding requests
  private final Collection<LogicalRequest> outstandingRequests = new ArrayList<>();
  //todo: currently only support a single cluster
  private Map<Long, ClusterTopology> clusterTopologyMap = new HashMap<>();

  @Inject
  private StackAdvisorBlueprintProcessor stackAdvisorBlueprintProcessor;

  @Inject
  private LogicalRequestFactory logicalRequestFactory;

  @Inject
  private AmbariContext ambariContext;

  private final Object initializationLock = new Object();

  @Inject
  private SecurityConfigurationFactory securityConfigurationFactory;

  @Inject
  private ConfigureClusterTaskFactory configureClusterTaskFactory;

  @Inject
  private AmbariEventPublisher ambariEventPublisher;

  @Inject
  private SettingDAO settingDAO;

  @Inject
  private TopologyValidatorService topologyValidatorService;

  /**
   * A boolean not cached thread-local (volatile) to prevent double-checked
   * locking on the synchronized keyword.
   */
  private volatile boolean isInitialized;

  private final static Logger LOG = LoggerFactory.getLogger(TopologyManager.class);

  /**
   * Stores request that belongs to blueprint creation
   */
  private Map<Long, LogicalRequest> clusterProvisionWithBlueprintCreateRequests = new HashMap<>();
  /**
   * Flag to show whether blueprint is already finished or not. It is used for shortcuts.
   */
  private Map<Long, Boolean> clusterProvisionWithBlueprintCreationFinished = new HashMap<>();

  public TopologyManager() {
    topologyTaskExecutorThreadPoolSize = 1;
  }

  @Inject
  public TopologyManager(Configuration configuration) {
    topologyTaskExecutorThreadPoolSize = configuration.getParallelTopologyTaskCreationThreadPoolSize();
    if (!configuration.isParallelTopologyTaskCreationEnabled()) {
      topologyTaskExecutorThreadPoolSize = 1;
    }
  }

  // executed by the IoC framework after creating the object (guice)
  @Inject
  private void register() {
    ambariEventPublisher.register(this);
  }

  @Inject
  private void setPersistedState() {
    persistedState = ambariContext.getPersistedTopologyState();
  }

  //todo: can't call in constructor.
  //todo: Very important that this occurs prior to any usage
  private void ensureInitialized() {
    if (!isInitialized) {
      synchronized (initializationLock) {
        if (!isInitialized) {
          replayRequests(persistedState.getAllRequests());
          // ensure KERBEROS_CLIENT is present in each hostgroup even if it's not in original BP
          for(ClusterTopology clusterTopology : clusterTopologyMap.values()) {
            if (clusterTopology.isClusterKerberosEnabled() && isKerberosClientInstallAllowed(clusterTopology)) {
              addKerberosClient(clusterTopology);
            }
          }
          isInitialized = true;
        }

      }
    }
  }

  /**
   * Called when heartbeat processing finishes
   * @param event
   */
  @Subscribe
  public void onRequestFinished(RequestFinishedEvent event) {
    if(event.getType() != AmbariEvent.AmbariEventType.REQUEST_FINISHED
            || clusterProvisionWithBlueprintCreateRequests.isEmpty()
            || Boolean.TRUE.equals(clusterProvisionWithBlueprintCreationFinished.get(event.getClusterId()))) {
      return;
    }

    if(isClusterProvisionWithBlueprintFinished(event.getClusterId())) {
      clusterProvisionWithBlueprintCreationFinished.put(event.getClusterId(), Boolean.TRUE);
      LogicalRequest provisionRequest = clusterProvisionWithBlueprintCreateRequests.get(event.getClusterId());
      if(isLogicalRequestSuccessful(provisionRequest)) {
        LOG.info("Cluster creation request id={} using Blueprint {} successfully completed for cluster id={}",
                clusterProvisionWithBlueprintCreateRequests.get(event.getClusterId()).getRequestId(),
                clusterTopologyMap.get(event.getClusterId()).getBlueprint().getName(),
                event.getClusterId());
        ambariEventPublisher.publish(new ClusterProvisionedEvent(event.getClusterId()));
      } else {
        LOG.info("Cluster creation request id={} using Blueprint {} failed for cluster id={}",
                clusterProvisionWithBlueprintCreateRequests.get(event.getClusterId()).getRequestId(),
                clusterTopologyMap.get(event.getClusterId()).getBlueprint().getName(),
                event.getClusterId());
      }
    }
  }

  /**
   * Returns if provision request for a cluster is tracked
   * @param clusterId
   * @return
   */
  public boolean isClusterProvisionWithBlueprintTracked(long clusterId) {
    return clusterProvisionWithBlueprintCreateRequests.containsKey(clusterId);
  }

  /**
   * Returns if the provision request for a cluster is finished.
   * Note that this method returns false if the request is not tracked.
   * See {@link TopologyManager#isClusterProvisionWithBlueprintTracked(long)}
   * @param clusterId
   * @return
   */
  public boolean isClusterProvisionWithBlueprintFinished(long clusterId) {
    if(!isClusterProvisionWithBlueprintTracked(clusterId)) {
      return false; // no blueprint request is running
    }
    // shortcut
    if(clusterProvisionWithBlueprintCreationFinished.containsKey(clusterId) && clusterProvisionWithBlueprintCreationFinished.get(clusterId)) {
      return true;
    }
    return isLogicalRequestFinished(clusterProvisionWithBlueprintCreateRequests.get(clusterId));
  }

  public RequestStatusResponse provisionCluster(final ProvisionClusterRequest request) throws InvalidTopologyException, AmbariException {
    ensureInitialized();

    final ClusterTopology topology = new ClusterTopologyImpl(ambariContext, request);
    final String clusterName = request.getClusterName();
    final Stack stack = topology.getBlueprint().getStack();
    final String repoVersion = request.getRepositoryVersion();
    final Long repoVersionID = request.getRepositoryVersionId();

    // get the id prior to creating ambari resources which increments the counter
    final Long provisionId = ambariContext.getNextRequestId();

    SecurityType securityType = null;
    Credential credential = null;

    SecurityConfiguration securityConfiguration = processSecurityConfiguration(request);

    if (securityConfiguration != null && securityConfiguration.getType() == SecurityType.KERBEROS) {
      securityType = SecurityType.KERBEROS;

      if (isKerberosClientInstallAllowed(topology)) {
        addKerberosClient(topology);
      }

      // refresh default stack config after adding KERBEROS_CLIENT component to topology
      topology.getBlueprint().getConfiguration().setParentConfiguration(stack.getConfiguration(topology.getBlueprint().getServices()));

      credential = request.getCredentialsMap().get(KDC_ADMIN_CREDENTIAL);
      if (credential == null) {
        throw new InvalidTopologyException(KDC_ADMIN_CREDENTIAL + " is missing from request.");
      }
    }

    topologyValidatorService.validateTopologyConfiguration(topology);


    // create resources
    ambariContext.createAmbariResources(topology, clusterName, securityType, repoVersion, repoVersionID);

    if (securityConfiguration != null) {
      securityConfiguration.getDescriptor().ifPresent(descriptor -> submitKerberosDescriptorAsArtifact(clusterName, descriptor));
    }

    if (credential != null) {
      submitCredential(clusterName, credential);
    }

    long clusterId = ambariContext.getClusterId(clusterName);
    topology.setClusterId(clusterId);
    request.setClusterId(clusterId);
    // set recommendation strategy
    topology.setConfigRecommendationStrategy(request.getConfigRecommendationStrategy());
    // set provision action requested
    topology.setProvisionAction(request.getProvisionAction());


    // create task executor for TopologyTasks
    getOrCreateTopologyTaskExecutor(clusterId);

    // persist request
    LogicalRequest logicalRequest = RetryHelper.executeWithRetry(new Callable<LogicalRequest>() {
        @Override
        public LogicalRequest call() throws Exception {
          LogicalRequest logicalRequest = processAndPersistProvisionClusterTopologyRequest(request, topology, provisionId);
          return logicalRequest;
        }
      }
    );

    clusterTopologyMap.put(clusterId, topology);

    addClusterConfigRequest(logicalRequest, topology, new ClusterConfigurationRequest(ambariContext, topology, true,
      stackAdvisorBlueprintProcessor, securityType == SecurityType.KERBEROS));

    // Process the logical request
    processRequest(request, topology, logicalRequest);

    //todo: this should be invoked as part of a generic lifecycle event which could possibly
    //todo: be tied to cluster state

    ambariContext.persistInstallStateForUI(clusterName, stack.getName(), stack.getVersion());
    clusterProvisionWithBlueprintCreateRequests.put(clusterId, logicalRequest);
    ambariEventPublisher.publish(new ClusterProvisionStartedEvent(clusterId));
    return getRequestStatus(logicalRequest.getRequestId());
  }

  /**
   * The Kerberos Client component is unnecessarily installed via Blueprints when kerberos-env/kdc-type is "none".
   * The Blueprint TopologyManager should only force the Kerberos client to be installed if Kerberos is enabled
   * and kerberos_env/kdc_type is not "none" or when kerberos_env/manage_identities is true.
   *
   * @param topology the Cluster Topology which provides the topology's Configuration object.
   * @return true if kerberos_env/kdc_type is not "none" or when kerberos_env/manage_identities is true
   */
  private boolean isKerberosClientInstallAllowed(final ClusterTopology topology) {
    final org.apache.ambari.server.topology.Configuration topologyConfig = topology.getBlueprint().getConfiguration();
    final String kdc_type = topologyConfig.getPropertyValue(KerberosHelper.KERBEROS_ENV, KerberosHelper.KDC_TYPE);
    final String manage_identities = topologyConfig.getPropertyValue(KerberosHelper.KERBEROS_ENV, KerberosHelper.MANAGE_IDENTITIES);

    return KDCType.NONE != KDCType.translate(kdc_type) || Boolean.parseBoolean(manage_identities);
  }

  @Subscribe
  public void onClusterConfigFinishedEvent(ClusterConfigFinishedEvent event) {
    ManagedThreadPoolExecutor taskExecutor = topologyTaskExecutorServiceMap.get(event.getClusterId());
    if (taskExecutor == null) {
      LOG.error("Can't find executor service taskQueue not found for cluster: {} ", event.getClusterName());
    } else {
      LOG.info("Starting topology task ExecutorService for cluster: {}", event.getClusterName());
      taskExecutor.start();
    }
  }


  /**
   * Saves the quick links profile to the DB as an Ambari setting. Creates a new setting entity or updates the existing
   * one.
   * @param quickLinksProfileJson the quicklinks profile in Json format
   */
  void saveOrUpdateQuickLinksProfile(String quickLinksProfileJson) {
    SettingEntity settingEntity = settingDAO.findByName(QuickLinksProfile.SETTING_NAME_QUICKLINKS_PROFILE);
    // create new
    if (null == settingEntity) {
      settingEntity = new SettingEntity();
      settingEntity.setName(QuickLinksProfile.SETTING_NAME_QUICKLINKS_PROFILE);
      settingEntity.setSettingType(QuickLinksProfile.SETTING_TYPE_AMBARI_SERVER);
      settingEntity.setContent(quickLinksProfileJson);
      settingEntity.setUpdatedBy(AuthorizationHelper.getAuthenticatedName());
      settingEntity.setUpdateTimestamp(System.currentTimeMillis());
      settingDAO.create(settingEntity);
    }
    // update existing
    else {
      settingEntity.setContent(quickLinksProfileJson);
      settingEntity.setUpdatedBy(AuthorizationHelper.getAuthenticatedName());
      settingEntity.setUpdateTimestamp(System.currentTimeMillis());
      settingDAO.merge(settingEntity);
    }
  }

  private static void submitCredential(String clusterName, Credential credential) {
    ResourceProvider provider = AmbariContext.getClusterController().ensureResourceProvider(Resource.Type.Credential);
    Map<String, Object> credentialProperties = ResourceProviderAdapter.createCredentialRequestProperties(clusterName, credential);
    Request request = new RequestImpl(ImmutableSet.of(), ImmutableSet.of(credentialProperties), ImmutableMap.of(), null);
    String baseMessage = String.format("Failed to add credential %s to cluster %s", credential.getAlias(), clusterName);
    try {
      RequestStatus status = provider.createResources(request);
      if (status.getStatus() != RequestStatus.Status.Complete) {
        String msg = String.format("%s, received status: %s", baseMessage, status.getStatus());
        LOG.error(msg);
        throw new RuntimeException(msg);
      }
    } catch (ResourceAlreadyExistsException | SystemException | UnsupportedPropertyException | NoSuchParentResourceException e) {
      String msg = String.format("%s, %s", baseMessage, e);
      LOG.error(msg);
      throw new RuntimeException(msg, e);
    }
  }

  /**
   * Retrieve security info from Blueprint if missing from Cluster Template request.
   *
   * @param request
   * @return
   */
  private SecurityConfiguration processSecurityConfiguration(ProvisionClusterRequest request) {
    LOG.debug("Getting security configuration from the request ...");
    SecurityConfiguration securityConfiguration = request.getSecurityConfiguration();

    if (securityConfiguration == null) {
      // todo - perform this logic at request creation instead!
      LOG.debug("There's no security configuration in the request, retrieving it from the associated blueprint");
      securityConfiguration = request.getBlueprint().getSecurity();
      if (securityConfiguration != null && securityConfiguration.getType() == SecurityType.KERBEROS &&
          securityConfiguration.getDescriptorReference() != null) {
        securityConfiguration = securityConfigurationFactory.loadSecurityConfigurationByReference
          (securityConfiguration.getDescriptorReference());
      }
    }
    return securityConfiguration;
  }

  private void submitKerberosDescriptorAsArtifact(String clusterName, Map<?,?> descriptor) {

    ResourceProvider artifactProvider =
        ambariContext.getClusterController().ensureResourceProvider(Resource.Type.Artifact);

    Map<String, Object> properties = ResourceProviderAdapter.createKerberosDescriptorRequestProperties(clusterName);
    Map<String, String> requestInfoProps = ImmutableMap.of(
      Request.REQUEST_INFO_BODY_PROPERTY, ArtifactResourceProvider.toArtifactDataJson(descriptor)
    );
    Request request = new RequestImpl(Collections.emptySet(), Collections.singleton(properties), requestInfoProps, null);

    try {
      RequestStatus status = artifactProvider.createResources(request);
      try {
        while (status.getStatus() != RequestStatus.Status.Complete) {
          LOG.info("Waiting for kerberos_descriptor artifact creation.");
          Thread.sleep(100);
        }
      } catch (InterruptedException e) {
        LOG.info("Wait for resource creation interrupted!");
      }

      if (status.getStatus() != RequestStatus.Status.Complete) {
        throw new RuntimeException("Failed to attach kerberos_descriptor artifact to cluster!");
      }
    } catch (SystemException | UnsupportedPropertyException | NoSuchParentResourceException e) {
      throw new RuntimeException("Failed to attach kerberos_descriptor artifact to cluster: " + e);
    } catch (ResourceAlreadyExistsException e) {
      throw new RuntimeException("Failed to attach kerberos_descriptor artifact to cluster as resource already exists.");
    }

  }

  public RequestStatusResponse scaleHosts(final ScaleClusterRequest request)
      throws InvalidTopologyException, AmbariException {

    ensureInitialized();
    LOG.info("TopologyManager.scaleHosts: Entering");
    String clusterName = request.getClusterName();
    long clusterId = ambariContext.getClusterId(clusterName);
    final ClusterTopology topology = clusterTopologyMap.get(clusterId);
    if (topology == null) {
      throw new InvalidTopologyException("Unable to retrieve cluster topology for cluster. This is most likely a " +
          "result of trying to scale a cluster via the API which was created using " +
          "the Ambari UI. At this time only clusters created via the API using a " +
          "blueprint can be scaled with this API.  If the cluster was originally created " +
          "via the API as described above, please file a Jira for this matter.");
    }

    hostNameCheck(request, topology);
    request.setClusterId(clusterId);
    if (ambariContext.isTopologyResolved(clusterId)) {
      getOrCreateTopologyTaskExecutor(clusterId).start();
    }

    // this registers/updates all request host groups
    topology.update(request);

    final Long requestId = ambariContext.getNextRequestId();
    LogicalRequest logicalRequest = RetryHelper.executeWithRetry(new Callable<LogicalRequest>() {
         @Override
         public LogicalRequest call() throws Exception {
           LogicalRequest logicalRequest = processAndPersistTopologyRequest(request, topology, requestId);

           return logicalRequest;
         }
       }
    );
    processRequest(request, topology, logicalRequest);
    return getRequestStatus(logicalRequest.getRequestId());
  }

  public void removePendingHostRequests(String clusterName, long requestId) {
    ensureInitialized();
    LOG.info("TopologyManager.removePendingHostRequests: Entering");

    long clusterId = 0;
    try {
      clusterId = ambariContext.getClusterId(clusterName);
    } catch (AmbariException e) {
      LOG.error("Unable to retrieve clusterId", e);
      throw new IllegalArgumentException("Unable to retrieve clusterId");
    }
    ClusterTopology topology = clusterTopologyMap.get(clusterId);
    if (topology == null) {
      throw new IllegalArgumentException("Unable to retrieve cluster topology for cluster");
    }

    LogicalRequest logicalRequest = allRequests.get(requestId);
    if (logicalRequest == null) {
      throw new IllegalArgumentException("No Logical Request found for requestId: " + requestId);
    }

    Collection<HostRequest> pendingHostRequests = logicalRequest.removePendingHostRequests(null);

    if (!logicalRequest.hasPendingHostRequests()) {
      outstandingRequests.remove(logicalRequest);
    }
    if (logicalRequest.getHostRequests().isEmpty()) {
      allRequests.remove(requestId);
    }

    persistedState.removeHostRequests(requestId, pendingHostRequests);

    // set current host count to number of currently connected hosts
    for (HostGroupInfo currentHostGroupInfo : topology.getHostGroupInfo().values()) {
      currentHostGroupInfo.setRequestedCount(currentHostGroupInfo.getHostNames().size());
    }

    LOG.info("TopologyManager.removePendingHostRequests: Exit");
  }

  /**
   * Removes topology host requests matched to the given host.  If the parent
   * request has no more child host requests, then it is also removed.
   * This is used when hosts are deleted from the cluster.
   *
   * @param hostName the host name for which requests should be removed
   */
  public void removeHostRequests(String hostName) {
    ensureInitialized();

    for (Iterator<LogicalRequest> iter = allRequests.values().iterator(); iter.hasNext(); ) {
      LogicalRequest logicalRequest = iter.next();
      Collection<HostRequest> removed = logicalRequest.removeHostRequestByHostName(hostName);
      if (!logicalRequest.hasPendingHostRequests()) {
        outstandingRequests.remove(logicalRequest);
      }
      if (logicalRequest.getHostRequests().isEmpty()) {
        iter.remove();
      }
      if (!removed.isEmpty()) {
        persistedState.removeHostRequests(logicalRequest.getRequestId(), removed);
      }
    }
  }

  /**
   * Creates and persists a {@see PersistedTopologyRequest} and a {@see LogicalRequest} for the provided
   * provision cluster request and topology.
   * @param request Provision cluster request to create a logical request for.
   * @param topology Cluster topology
   * @param logicalRequestId The Id for the created logical request
   * @return Logical request created.
   */
  @Transactional
  protected LogicalRequest processAndPersistProvisionClusterTopologyRequest(ProvisionClusterRequest request, ClusterTopology topology, Long logicalRequestId)
    throws InvalidTopologyException, AmbariException {

    if (null != request.getQuickLinksProfileJson()) {
      saveOrUpdateQuickLinksProfile(request.getQuickLinksProfileJson());
    }

    LogicalRequest logicalRequest = processAndPersistTopologyRequest(request, topology, logicalRequestId);

    return logicalRequest;

  }


  /**
   * Creates and persists a {@see PersistedTopologyRequest} and a {@see LogicalRequest} for the provided request and topology.
   * @param request {@see ProvisionClusterRequest} or {@see ScaleClusterRequest} to create a logical request for.
   * @param topology Cluster topology
   * @param logicalRequestId The Id for the created logical request
   * @return Logical request created.
   */
  @Transactional
  protected  LogicalRequest processAndPersistTopologyRequest(BaseClusterRequest request, ClusterTopology topology, Long logicalRequestId)
    throws InvalidTopologyException, AmbariException {
    PersistedTopologyRequest persistedRequest = persistedState.persistTopologyRequest(request);

    LogicalRequest logicalRequest = createLogicalRequest(persistedRequest, topology, logicalRequestId);

    return logicalRequest;
  }

  private void hostNameCheck(ScaleClusterRequest request, ClusterTopology topology) throws InvalidTopologyException {
    Set<String> hostNames = new HashSet<>();
    for(Map.Entry<String, HostGroupInfo> entry : request.getHostGroupInfo().entrySet()) {
      hostNames.addAll(entry.getValue().getHostNames());
    }
    for(String hostName : hostNames) {
      // check if host exists already
      if(topology.getHostGroupForHost(hostName) != null) {
        throw new InvalidTopologyException("Host " + hostName + " cannot be added, because it is already in the cluster");
      }
    }
  }

  public void onHostRegistered(HostImpl host, boolean associatedWithCluster) {
    ensureInitialized();
    LOG.info("TopologyManager.onHostRegistered: Entering");
    if (associatedWithCluster || isHostIgnored(host.getHostName())) {
      LOG.info("TopologyManager.onHostRegistered: host = {} is already associated with the cluster or is currently being processed", host.getHostName());
      return;
    }

    boolean matchedToRequest = false;
    String hostName = host.getHostName();
    // The lock ordering in this method must always be the same ordering as TopologyManager.processRequest
    // TODO: Locking strategies for TopologyManager should be reviewed and possibly rewritten in a future release
    synchronized (availableHosts) {
      synchronized (reservedHosts) {
        if (reservedHosts.containsKey(hostName)) {
          LogicalRequest request = reservedHosts.remove(hostName);
          HostOfferResponse response = request.offer(host);
          if (response.getAnswer() != HostOfferResponse.Answer.ACCEPTED) {
            throw new RuntimeException("LogicalRequest declined host offer of explicitly requested host: " + hostName);
          }

          LOG.info("TopologyManager.onHostRegistered: processing accepted host offer for reserved host = {}", hostName);
          processAcceptedHostOffer(getClusterTopology(request.getClusterId()), response, host);
          matchedToRequest = true;
        }
      }

      // can be true if host was reserved
      if (!matchedToRequest) {
        synchronized (outstandingRequests) {
          Iterator<LogicalRequest> outstandingRequestIterator = outstandingRequests.iterator();
          while (!matchedToRequest && outstandingRequestIterator.hasNext()) {
            LogicalRequest request = outstandingRequestIterator.next();
            HostOfferResponse hostOfferResponse = request.offer(host);
            switch (hostOfferResponse.getAnswer()) {
              case ACCEPTED:
                matchedToRequest = true;
                LOG.info("TopologyManager.onHostRegistered: processing accepted host offer for matched host = {}", hostName);
                processAcceptedHostOffer(getClusterTopology(request.getClusterId()), hostOfferResponse, host);
                break;
              case DECLINED_DONE:
                LOG.info("TopologyManager.onHostRegistered: DECLINED_DONE received for host = {}", hostName);
                outstandingRequestIterator.remove();
                break;
              case DECLINED_PREDICATE:
                LOG.info("TopologyManager.onHostRegistered: DECLINED_PREDICATE received for host = {}", hostName);
                break;
            }
          }
        }
      }

      if (!matchedToRequest) {
        boolean addToAvailableList = true;
        for (HostImpl registered : availableHosts) {
          if (Objects.equals(registered.getHostId(), host.getHostId())) {
            LOG.info("Host {} re-registered, will not be added to the available hosts list", hostName);
            addToAvailableList = false;
            break;
          }
        }

        if (addToAvailableList) {
          LOG.info("TopologyManager: Queueing available host {}", hostName);
          availableHosts.add(host);
        }
      }
    }
  }

  /**
   * Through this method {@see TopologyManager} gets notified when a connection to a host in the cluster is lost.
   * The passed host will be excluded from scheduling any tasks onto it as it can't be reached.
   * @param host
   */
  public void onHostHeartBeatLost(Host host) {
    if (AmbariServer.getController() == null) {
      return;
    }
    ensureInitialized();
    synchronized (availableHosts) {
      LOG.info("Hearbeat for host {} lost thus removing it from available hosts.", host.getHostName());
      availableHosts.remove(host);
    }
  }

  public LogicalRequest getRequest(long requestId) {
    ensureInitialized();
    return allRequests.get(requestId);
  }

  public Collection<LogicalRequest> getRequests(Collection<Long> requestIds) {
    ensureInitialized();
    if (requestIds.isEmpty()) {
      return allRequests.values();
    } else {
      Collection<LogicalRequest> matchingRequests = new ArrayList<>();
      for (long id : requestIds) {
        LogicalRequest request = allRequests.get(id);
        if (request != null) {
          matchingRequests.add(request);
        }
      }
      return matchingRequests;
    }
  }

  /**
   * Currently we are just returning all stages for all requests
