class EmrHelperServiceImpl implements EmrHelperService {

    EmrClusterCreateDto emrCreateClusterAwsSpecificStepsImpl(EmrClusterCreateRequest request, EmrClusterDefinition emrClusterDefinition,
        EmrClusterAlternateKeyDto emrClusterAlternateKeyDto, EmrParamsDto emrParamsDto)
    {
        // RunJobFlow creates and starts running a new cluster
        // The RunJobFlow request can contain InstanceFleets parameters or InstanceGroups (InstanceDefinitions) parameters, but not both
        // If instance group definitions are specified, find best price and update definition.
        // Else instance fleet definitions are specified. If minimum ip filter is greater than 0, find valid subnets and update definition.
        if (!emrHelper.isInstanceDefinitionsEmpty(emrClusterDefinition.getInstanceDefinitions()))
        {
            emrPricingHelper.updateEmrClusterDefinitionWithBestPrice(emrClusterAlternateKeyDto, emrClusterDefinition, emrParamsDto);
        }
        else
        {
            updateEmrClusterDefinitionWithValidInstanceFleetSubnets(emrClusterAlternateKeyDto, emrClusterDefinition, emrParamsDto);
        }

        // The cluster ID record.
        String clusterId = null;

        // Is EMR cluster already existing.
        Boolean emrClusterAlreadyExists = null;

        // Is EMR cluster created.
        Boolean emrClusterCreated = null;

        // EMR cluster status string.
        String emrClusterStatus = null;

        // If the dryRun flag is null or false. This is the default option if no flag is given.
        if (!Boolean.TRUE.equals(request.isDryRun()))
        {
            /*
             * Create the cluster only if the cluster does not already exist.
             * If the cluster is created, record the newly created cluster ID.
             * If the cluster already exists, record the existing cluster ID.
             * If there is any error while attempting to check for existing cluster or create a new one, handle the exception to throw appropriate exception.
             */
            String clusterName = emrHelper
                .buildEmrClusterName(emrClusterAlternateKeyDto.getNamespace(), emrClusterAlternateKeyDto.getEmrClusterDefinitionName(),
                    emrClusterAlternateKeyDto.getEmrClusterName());
            try
            {
                // Synchronizing this block of code to prevent duplicate cluster creation.
                synchronized (this)
                {
                    LOGGER.info("Entering synchronized block.");

                    // Try to get an active EMR cluster by its name.
                    ClusterSummary clusterSummary =
                        emrDao.getActiveEmrClusterByNameAndAccountId(clusterName, emrClusterDefinition.getAccountId(), emrParamsDto);

                    // If cluster does not already exist.
                    if (clusterSummary == null)
                    {
                        clusterId = emrDao.createEmrCluster(clusterName, emrClusterDefinition, emrParamsDto);
                        emrClusterCreated = true;
                        emrClusterStatus = emrDao.getEmrClusterStatusById(clusterId, emrParamsDto);

                    }
                    // If the cluster already exists.
                    else
                    {
                        clusterId = clusterSummary.getId();
                        emrClusterCreated = false;
                        emrClusterAlreadyExists = true;

                        // If the cluster already exists use the status from the get active EMR cluster by name and account id method call.
                        emrClusterStatus = clusterSummary.getStatus().getState();
                    }

                    LOGGER.info("Exiting synchronized block.");
                }
            }
            catch (AmazonServiceException ex)
            {
                awsServiceHelper.handleAmazonException(ex, "An Amazon exception occurred while creating EMR cluster with name \"" + clusterName + "\".");
            }
        }
        // If the dryRun flag is true and not null
        else
        {
            emrClusterCreated = false;
        }

        return new EmrClusterCreateDto(clusterId, emrClusterAlreadyExists, emrClusterCreated, emrClusterStatus);
    }

}