class EmrClusterDefinitionHelper {

    public void validateEmrClusterDefinitionConfiguration(EmrClusterDefinition emrClusterDefinition) throws IllegalArgumentException
    {
        Assert.notNull(emrClusterDefinition, "An EMR cluster definition configuration must be specified.");

        Assert.isTrue(StringUtils.isNotBlank(emrClusterDefinition.getSubnetId()), "Subnet ID must be specified");
        for (String token : emrClusterDefinition.getSubnetId().split(","))
        {
            Assert.isTrue(StringUtils.isNotBlank(token), "No blank is allowed in the list of subnet IDs");
        }

        Assert.isTrue(!emrHelper.isInstanceDefinitionsEmpty(emrClusterDefinition.getInstanceDefinitions()) ||
            CollectionUtils.isNotEmpty(emrClusterDefinition.getInstanceFleets()), "Instance group definitions or instance fleets must be specified.");

        if (!emrHelper.isInstanceDefinitionsEmpty(emrClusterDefinition.getInstanceDefinitions()))
        {
            // Check master instances.
            Assert.notNull(emrClusterDefinition.getInstanceDefinitions().getMasterInstances(), "Master instances must be specified.");
            validateMasterInstanceDefinition(emrClusterDefinition.getInstanceDefinitions().getMasterInstances());

            // Check core instances.
            if (emrClusterDefinition.getInstanceDefinitions().getCoreInstances() != null)
            {
                validateInstanceDefinition("core", emrClusterDefinition.getInstanceDefinitions().getCoreInstances(), 0);
                // If instance count is <= 0, remove the entire core instance definition since it is redundant.
                if (emrClusterDefinition.getInstanceDefinitions().getCoreInstances().getInstanceCount() <= 0)
                {
                    emrClusterDefinition.getInstanceDefinitions().setCoreInstances(null);
                }
            }

            // Check task instances
            if (emrClusterDefinition.getInstanceDefinitions().getTaskInstances() != null)
            {
                validateInstanceDefinition("task", emrClusterDefinition.getInstanceDefinitions().getTaskInstances(), 1);
            }

            // Check that total number of instances does not exceed the max allowed.
            int maxEmrInstanceCount = configurationHelper.getProperty(ConfigurationValue.MAX_EMR_INSTANCES_COUNT, Integer.class);
            if (maxEmrInstanceCount > 0)
            {
                int instancesRequested = emrClusterDefinition.getInstanceDefinitions().getMasterInstances().getInstanceCount();
                if (emrClusterDefinition.getInstanceDefinitions().getCoreInstances() != null)
                {
                    instancesRequested += emrClusterDefinition.getInstanceDefinitions().getCoreInstances().getInstanceCount();
                }
                if (emrClusterDefinition.getInstanceDefinitions().getTaskInstances() != null)
                {
                    instancesRequested += emrClusterDefinition.getInstanceDefinitions().getTaskInstances().getInstanceCount();
                }

                Assert.isTrue((maxEmrInstanceCount >= instancesRequested), "Total number of instances requested can not exceed : " + maxEmrInstanceCount);
            }
        }

        // Make sure instance fleet minimum ip available filter is non-negative
        if (emrClusterDefinition.getInstanceFleetMinimumIpAvailableFilter() != null)
        {
            Assert.isTrue(emrClusterDefinition.getInstanceFleetMinimumIpAvailableFilter() >= 0,
                    "Minimum number of available IPs required for the instance fleet must be non-negative");
        }

        // Validate node tags including checking for required tags and detecting any duplicate node tag names in case sensitive manner.
        Assert.notEmpty(emrClusterDefinition.getNodeTags(), "Node tags must be specified.");
        HashSet<String> nodeTagNameValidationSet = new HashSet<>();
        for (NodeTag nodeTag : emrClusterDefinition.getNodeTags())
        {
            Assert.hasText(nodeTag.getTagName(), "A node tag name must be specified.");
            Assert.hasText(nodeTag.getTagValue(), "A node tag value must be specified.");
            Assert.isTrue(!nodeTagNameValidationSet.contains(nodeTag.getTagName()), String.format("Duplicate node tag \"%s\" is found.", nodeTag.getTagName()));
            nodeTagNameValidationSet.add(nodeTag.getTagName());
        }

        // Validate the mandatory AWS tags are there
        for (String mandatoryTag : herdStringHelper.splitStringWithDefaultDelimiter(configurationHelper.getProperty(ConfigurationValue.MANDATORY_AWS_TAGS)))
        {
            Assert.isTrue(nodeTagNameValidationSet.contains(mandatoryTag), String.format("Mandatory AWS tag not specified: \"%s\"", mandatoryTag));
        }

        emrClusterDefinition.setAdditionalMasterSecurityGroups(
            assertNotBlankAndTrim(emrClusterDefinition.getAdditionalMasterSecurityGroups(), "additionalMasterSecurityGroup"));

        emrClusterDefinition
            .setAdditionalSlaveSecurityGroups(assertNotBlankAndTrim(emrClusterDefinition.getAdditionalSlaveSecurityGroups(), "additionalSlaveSecurityGroup"));

        // Fail if security configuration is specified for EMR version less than 4.8.0.
        if (StringUtils.isNotBlank(emrClusterDefinition.getSecurityConfiguration()))
        {
            final DefaultArtifactVersion securityConfigurationMinEmrVersion = new DefaultArtifactVersion("4.8.0");
            Assert.isTrue(StringUtils.isNotBlank(emrClusterDefinition.getReleaseLabel()) &&
                securityConfigurationMinEmrVersion.compareTo(new DefaultArtifactVersion(emrClusterDefinition.getReleaseLabel().replaceFirst("^(emr-)", ""))) <=
                    0, "EMR security configuration is not supported prior to EMR release 4.8.0.");
        }
    }

}