class RequiredConfigPropertiesValidator implements TopologyValidator {

  @Override
  public void validate(ClusterTopology topology) throws InvalidTopologyException {

    // collect required properties
    Map<String, Map<String, Collection<String>>> requiredPropertiesByService = getRequiredPropertiesByService(topology.getBlueprint());

    // find missing properties in the cluster configuration
    Map<String, Collection<String>> missingProperties = new TreeMap<>();
    Map<String, Map<String, String>> topologyConfiguration = new HashMap<>(topology.getConfiguration().getFullProperties(1));

    for (HostGroup hostGroup : topology.getBlueprint().getHostGroups().values()) {
      LOGGER.debug("Processing hostgroup configurations for hostgroup: {}", hostGroup.getName());

      // copy of all configurations available in the topology hgConfig -> topologyConfig -> bpConfig
      Map<String, Map<String, String>> operationalConfigurations = new HashMap<>(topologyConfiguration);

      for (Map.Entry<String, Map<String, String>> hostgroupConfigEntry : hostGroup.getConfiguration().getProperties().entrySet()) {
        if (operationalConfigurations.containsKey(hostgroupConfigEntry.getKey())) {
          operationalConfigurations.get(hostgroupConfigEntry.getKey()).putAll(hostgroupConfigEntry.getValue());
        } else {
          operationalConfigurations.put(hostgroupConfigEntry.getKey(), hostgroupConfigEntry.getValue());
        }
      }

      for (String hostGroupService : hostGroup.getServices()) {

        if (!requiredPropertiesByService.containsKey(hostGroupService)) {
          // there are no required properties for the service
          LOGGER.debug("There are no required properties found for hostgroup/service: [{}/{}]", hostGroup.getName(), hostGroupService);
          continue;
        }

        Map<String, Collection<String>> requiredPropertiesByType = requiredPropertiesByService.get(hostGroupService);

        for (String configType : requiredPropertiesByType.keySet()) {

          // We need a copy not to modify the original
          Collection<String> requiredPropertiesForType = new HashSet(
              requiredPropertiesByType.get(configType));

          if (!operationalConfigurations.containsKey(configType)) {
            // all required configuration is missing for the config type
            missingProperties = addTomissingProperties(missingProperties, hostGroup.getName(), requiredPropertiesForType);
            continue;
          }

          Collection<String> operationalConfigsForType = operationalConfigurations.get(configType).keySet();
          requiredPropertiesForType.removeAll(operationalConfigsForType);
          if (!requiredPropertiesForType.isEmpty()) {
            LOGGER.info("Found missing properties in hostgroup: {}, config type: {}, mising properties: {}", hostGroup.getName(),
              configType, requiredPropertiesForType);
            missingProperties = addTomissingProperties(missingProperties, hostGroup.getName(), requiredPropertiesForType);
          }
        }
      }

    }

    if (!missingProperties.isEmpty()) {
      throw new InvalidTopologyException("Missing required properties.  Specify a value for these " +
        "properties in the blueprint or cluster creation template configuration. " + missingProperties);
    }

  }

}