class TopologyManager {

  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() != null) {
      submitKerberosDescriptorAsArtifact(clusterName, securityConfiguration.getDescriptor());
    }

    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());
  }

  public  TopologyManager();
  @Inject public  TopologyManager(Configuration configuration);

}