class EmrPricingHelper extends AwsHelper {

    public void updateEmrClusterDefinitionWithBestPrice(EmrClusterAlternateKeyDto emrClusterAlternateKeyDto, EmrClusterDefinition emrClusterDefinition,
        AwsParamsDto awsParamsDto)
    {
        EmrVpcPricingState emrVpcPricingState = new EmrVpcPricingState();

        // Get total count of instances this definition will attempt to create
        int totalInstanceCount = getTotalInstanceCount(emrClusterDefinition);

        // Get the subnet information
        List<Subnet> subnets = getSubnets(emrClusterDefinition, awsParamsDto);
        for (Subnet subnet : subnets)
        {
            emrVpcPricingState.getSubnetAvailableIpAddressCounts().put(subnet.getSubnetId(), subnet.getAvailableIpAddressCount());
        }
        // Filter out subnets with not enough available IPs
        removeSubnetsWithAvailableIpsLessThan(subnets, totalInstanceCount);

        if (subnets.isEmpty())
        {
            LOGGER.info(String.format("Insufficient IP availability. namespace=\"%s\" emrClusterDefinitionName=\"%s\" emrClusterName=\"%s\" " +
                    "totalRequestedInstanceCount=%s emrVpcPricingState=%s", emrClusterAlternateKeyDto.getNamespace(),
                emrClusterAlternateKeyDto.getEmrClusterDefinitionName(), emrClusterAlternateKeyDto.getEmrClusterName(), totalInstanceCount,
                jsonHelper.objectToJson(emrVpcPricingState)));
            throw new ObjectNotFoundException(String.format(
                "There are no subnets in the current VPC which have sufficient IP addresses available to run your " +
                    "clusters. Try expanding the list of subnets or try again later. requestedInstanceCount=%s%n%s", totalInstanceCount,
                emrVpcPricingStateFormatter.format(emrVpcPricingState)));
        }

        // Best prices are accumulated in this list
        List<EmrClusterPriceDto> emrClusterPrices = new ArrayList<>();

        InstanceDefinition masterInstanceDefinition = getMasterInstanceDefinition(emrClusterDefinition);
        InstanceDefinition coreInstanceDefinition = getCoreInstanceDefinition(emrClusterDefinition);
        InstanceDefinition taskInstanceDefinition = getTaskInstanceDefinition(emrClusterDefinition);

        Set<String> requestedInstanceTypes = new HashSet<>();

        String masterInstanceType = masterInstanceDefinition.getInstanceType();
        requestedInstanceTypes.add(masterInstanceType);

        if (coreInstanceDefinition != null)
        {
            String coreInstanceType = coreInstanceDefinition.getInstanceType();
            requestedInstanceTypes.add(coreInstanceType);
        }

        if (taskInstanceDefinition != null)
        {
            String taskInstanceType = taskInstanceDefinition.getInstanceType();
            requestedInstanceTypes.add(taskInstanceType);
        }

        // Get AZs for the subnets
        for (AvailabilityZone availabilityZone : getAvailabilityZones(subnets, awsParamsDto))
        {
            // Create a mapping of instance types to prices for more efficient, in-memory lookup
            // When AWS does not return any spot price history for an instance type in an availability zone, the algorithm will not use that availability zone
            // when selecting the lowest price.
            Map<String, BigDecimal> instanceTypeSpotPrices = getInstanceTypeSpotPrices(availabilityZone, requestedInstanceTypes, awsParamsDto);

            emrVpcPricingState.getSpotPricesPerAvailabilityZone().put(availabilityZone.getZoneName(), instanceTypeSpotPrices);

            // Get and compare master price
            BigDecimal masterSpotPrice = instanceTypeSpotPrices.get(masterInstanceType);
            Ec2PriceDto masterPrice = getBestInstancePrice(masterSpotPrice, masterInstanceDefinition);

            // Get and compare core price
            Ec2PriceDto corePrice = null;
            if (coreInstanceDefinition != null)
            {
                String coreInstanceType = coreInstanceDefinition.getInstanceType();
                BigDecimal coreSpotPrice = instanceTypeSpotPrices.get(coreInstanceType);
                corePrice = getBestInstancePrice(coreSpotPrice, coreInstanceDefinition);
            }

            // Get and compare task price
            Ec2PriceDto taskPrice = null;
            if (taskInstanceDefinition != null)
            {
                String taskInstanceType = taskInstanceDefinition.getInstanceType();
                BigDecimal taskSpotPrice = instanceTypeSpotPrices.get(taskInstanceType);
                taskPrice = getBestInstancePrice(taskSpotPrice, taskInstanceDefinition);
            }

            // If prices were found
            if (masterPrice != null && (coreInstanceDefinition == null || corePrice != null) && (taskInstanceDefinition == null || taskPrice != null))
            {
                // Add the pricing result to the result list
                emrClusterPrices.add(createEmrClusterPrice(availabilityZone, masterPrice, corePrice, taskPrice));
            }

            // If prices were not found for either master or core, this AZ cannot satisfy the search criteria. Ignore this AZ.
        }

        if (emrClusterPrices.isEmpty())
        {
            LOGGER.info(String.format("No subnets which satisfied the best price search criteria. namespace=\"%s\" emrClusterDefinitionName=\"%s\" " +
                    "emrClusterName=\"%s\" emrVpcPricingState=%s", emrClusterAlternateKeyDto.getNamespace(),
                emrClusterAlternateKeyDto.getEmrClusterDefinitionName(), emrClusterAlternateKeyDto.getEmrClusterName(),
                jsonHelper.objectToJson(emrVpcPricingState)));
            throw new ObjectNotFoundException(String.format(
                "There were no subnets which satisfied your best price search criteria. If you explicitly opted to use spot EC2 instances, please confirm " +
                    "that your instance types support spot pricing. Otherwise, try setting the max price or the on-demand threshold to a higher value.%n%s",
                emrVpcPricingStateFormatter.format(emrVpcPricingState)));
        }

        // Find the best prices from the result list
        EmrClusterPriceDto bestEmrClusterPrice = getEmrClusterPriceWithLowestCoreInstancePrice(emrClusterPrices);

        if (bestEmrClusterPrice != null)
        {
            // Find the best subnet among the best AZ's
            Subnet bestEmrClusterSubnet = getBestSubnetForAvailabilityZone(bestEmrClusterPrice.getAvailabilityZone(), subnets);

            // Update the definition with the new calculated values
            updateInstanceDefinitionsWithBestPrice(emrClusterDefinition, bestEmrClusterSubnet, bestEmrClusterPrice);
        }
    }

}