@VisibleForTesting
  public static void validateAndResolveService(Service service,
      SliderFileSystem fs, org.apache.hadoop.conf.Configuration conf) throws
      IOException {
    boolean dnsEnabled = conf.getBoolean(RegistryConstants.KEY_DNS_ENABLED,
        RegistryConstants.DEFAULT_DNS_ENABLED);
    if (dnsEnabled) {
      if (RegistryUtils.currentUser().length()
          > RegistryConstants.MAX_FQDN_LABEL_LENGTH) {
        throw new IllegalArgumentException(
            RestApiErrorMessages.ERROR_USER_NAME_INVALID);
      }
      userNamePattern.validate(RegistryUtils.currentUser());
    }

    if (StringUtils.isEmpty(service.getName())) {
      throw new IllegalArgumentException(
          RestApiErrorMessages.ERROR_APPLICATION_NAME_INVALID);
    }

    if (StringUtils.isEmpty(service.getVersion())) {
      throw new IllegalArgumentException(String.format(
          RestApiErrorMessages.ERROR_APPLICATION_VERSION_INVALID,
          service.getName()));
    }

    validateNameFormat(service.getName(), conf);

    // If the service has no components, throw error
    if (!hasComponent(service)) {
      throw new IllegalArgumentException(
          "No component specified for " + service.getName());
    }

    if (UserGroupInformation.isSecurityEnabled()) {
      validateKerberosPrincipal(service.getKerberosPrincipal());
    }

    // Validate the Docker client config.
    try {
      validateDockerClientConfiguration(service, conf);
    } catch (IOException e) {
      throw new IllegalArgumentException(e);
    }

    // Validate there are no component name collisions (collisions are not
    // currently supported) and add any components from external services
    Configuration globalConf = service.getConfiguration();
    Set<String> componentNames = new HashSet<>();
    List<Component> componentsToRemove = new ArrayList<>();
    List<Component> componentsToAdd = new ArrayList<>();
    for (Component comp : service.getComponents()) {
      int maxCompLength = RegistryConstants.MAX_FQDN_LABEL_LENGTH;
      maxCompLength = maxCompLength - Long.toString(Long.MAX_VALUE).length();
      if (dnsEnabled && comp.getName().length() > maxCompLength) {
        throw new IllegalArgumentException(String.format(RestApiErrorMessages
            .ERROR_COMPONENT_NAME_INVALID, maxCompLength, comp.getName()));
      }
      if (service.getName().equals(comp.getName())) {
        throw new IllegalArgumentException(String.format(RestApiErrorMessages
                .ERROR_COMPONENT_NAME_CONFLICTS_WITH_SERVICE_NAME,
            comp.getName(), service.getName()));
      }
      if (componentNames.contains(comp.getName())) {
        throw new IllegalArgumentException("Component name collision: " +
            comp.getName());
      }
      // If artifact is of type SERVICE (which cannot be filled from global),
      // read external service and add its components to this service
      if (comp.getArtifact() != null && comp.getArtifact().getType() ==
          Artifact.TypeEnum.SERVICE) {
        if (StringUtils.isEmpty(comp.getArtifact().getId())) {
          throw new IllegalArgumentException(
              RestApiErrorMessages.ERROR_ARTIFACT_ID_INVALID);
        }
        LOG.info("Marking {} for removal", comp.getName());
        componentsToRemove.add(comp);
        List<Component> externalComponents = getComponents(fs,
            comp.getArtifact().getId());
        for (Component c : externalComponents) {
          Component override = service.getComponent(c.getName());
          if (override != null && override.getArtifact() == null) {
            // allow properties from external components to be overridden /
            // augmented by properties in this component, except for artifact
            // which must be read from external component
            override.mergeFrom(c);
            LOG.info("Merging external component {} from external {}", c
                .getName(), comp.getName());
          } else {
            if (componentNames.contains(c.getName())) {
              throw new IllegalArgumentException("Component name collision: " +
                  c.getName());
            }
            componentNames.add(c.getName());
            componentsToAdd.add(c);
            LOG.info("Adding component {} from external {}", c.getName(),
                comp.getName());
          }
        }
      } else {
        // otherwise handle as a normal component
        componentNames.add(comp.getName());
        // configuration
        comp.getConfiguration().mergeFrom(globalConf);
      }
    }
    service.getComponents().removeAll(componentsToRemove);
    service.getComponents().addAll(componentsToAdd);

    // Validate components and let global values take effect if component level
    // values are not provided
    Artifact globalArtifact = service.getArtifact();
    Resource globalResource = service.getResource();
    for (Component comp : service.getComponents()) {
      // fill in global artifact unless it is type SERVICE
      if (comp.getArtifact() == null && service.getArtifact() != null
          && service.getArtifact().getType() != Artifact.TypeEnum
          .SERVICE) {
        comp.setArtifact(globalArtifact);
      }
      // fill in global resource
      if (comp.getResource() == null) {
        comp.setResource(globalResource);
      }
      // validate dependency existence
      if (comp.getDependencies() != null) {
        for (String dependency : comp.getDependencies()) {
          if (!componentNames.contains(dependency)) {
            throw new IllegalArgumentException(String.format(
                RestApiErrorMessages.ERROR_DEPENDENCY_INVALID, dependency,
                comp.getName()));
          }
        }
      }
      validateComponent(comp, fs.getFileSystem(), conf);
    }
    validatePlacementPolicy(service.getComponents(), componentNames);

    // validate dependency tree
    sortByDependencies(service.getComponents());

    // Service lifetime if not specified, is set to unlimited lifetime
    if (service.getLifetime() == null) {
      service.setLifetime(RestApiConstants.DEFAULT_UNLIMITED_LIFETIME);
    }
  }
@Test(timeout = 90000)
  public void testResourceValidation() throws Exception {
    assertEquals(RegistryConstants.MAX_FQDN_LABEL_LENGTH + 1, LEN_64_STR
        .length());

    SliderFileSystem sfs = ServiceTestUtils.initMockFs();

    Service app = new Service();

    // no name
    try {
      ServiceApiUtil.validateAndResolveService(app, sfs, CONF_DNS_ENABLED);
      Assert.fail(EXCEPTION_PREFIX + "service with no name");
    } catch (IllegalArgumentException e) {
      assertEquals(ERROR_APPLICATION_NAME_INVALID, e.getMessage());
    }

    app.setName("test");
    // no version
    try {
      ServiceApiUtil.validateAndResolveService(app, sfs, CONF_DNS_ENABLED);
      Assert.fail(EXCEPTION_PREFIX + " service with no version");
    } catch (IllegalArgumentException e) {
      assertEquals(String.format(ERROR_APPLICATION_VERSION_INVALID,
          app.getName()), e.getMessage());
    }

    app.setVersion("v1");
    // bad format name
    String[] badNames = {"4finance", "Finance", "finance@home", LEN_64_STR};
    for (String badName : badNames) {
      app.setName(badName);
      try {
        ServiceApiUtil.validateAndResolveService(app, sfs, CONF_DNS_ENABLED);
        Assert.fail(EXCEPTION_PREFIX + "service with bad name " + badName);
      } catch (IllegalArgumentException e) {

      }
    }

    // launch command not specified
    app.setName(LEN_64_STR);
    Component comp = new Component().name("comp1");
    app.addComponent(comp);
    try {
      ServiceApiUtil.validateAndResolveService(app, sfs, CONF_DEFAULT_DNS);
      Assert.fail(EXCEPTION_PREFIX + "service with no launch command");
    } catch (IllegalArgumentException e) {
      assertEquals(RestApiErrorMessages.ERROR_ABSENT_LAUNCH_COMMAND,
          e.getMessage());
    }

    // launch command not specified
    app.setName(LEN_64_STR.substring(0, RegistryConstants
        .MAX_FQDN_LABEL_LENGTH));
    try {
      ServiceApiUtil.validateAndResolveService(app, sfs, CONF_DNS_ENABLED);
      Assert.fail(EXCEPTION_PREFIX + "service with no launch command");
    } catch (IllegalArgumentException e) {
      assertEquals(RestApiErrorMessages.ERROR_ABSENT_LAUNCH_COMMAND,
          e.getMessage());
    }

    // memory not specified
    comp.setLaunchCommand("sleep 1");
    Resource res = new Resource();
    app.setResource(res);
    try {
      ServiceApiUtil.validateAndResolveService(app, sfs, CONF_DNS_ENABLED);
      Assert.fail(EXCEPTION_PREFIX + "service with no memory");
    } catch (IllegalArgumentException e) {
      assertEquals(String.format(
          RestApiErrorMessages.ERROR_RESOURCE_MEMORY_FOR_COMP_INVALID,
          comp.getName()), e.getMessage());
    }

    // invalid no of cpus
    res.setMemory("100mb");
    res.setCpus(-2);
    try {
      ServiceApiUtil.validateAndResolveService(app, sfs, CONF_DNS_ENABLED);
      Assert.fail(
          EXCEPTION_PREFIX + "service with invalid no of cpus");
    } catch (IllegalArgumentException e) {
      assertEquals(String.format(
          RestApiErrorMessages.ERROR_RESOURCE_CPUS_FOR_COMP_INVALID_RANGE,
          comp.getName()), e.getMessage());
    }

    // number of containers not specified
    res.setCpus(2);
    try {
      ServiceApiUtil.validateAndResolveService(app, sfs, CONF_DNS_ENABLED);
      Assert.fail(EXCEPTION_PREFIX + "service with no container count");
    } catch (IllegalArgumentException e) {
      Assert.assertTrue(e.getMessage()
          .contains(ERROR_CONTAINERS_COUNT_INVALID));
    }

    // specifying profile along with cpus/memory raises exception
    res.setProfile("hbase_finance_large");
    try {
      ServiceApiUtil.validateAndResolveService(app, sfs, CONF_DNS_ENABLED);
      Assert.fail(EXCEPTION_PREFIX
          + "service with resource profile along with cpus/memory");
    } catch (IllegalArgumentException e) {
      assertEquals(String.format(RestApiErrorMessages
              .ERROR_RESOURCE_PROFILE_MULTIPLE_VALUES_FOR_COMP_NOT_SUPPORTED,
          comp.getName()),
          e.getMessage());
    }

    // currently resource profile alone is not supported.
    // TODO: remove the next test once resource profile alone is supported.
    res.setCpus(null);
    res.setMemory(null);
    try {
      ServiceApiUtil.validateAndResolveService(app, sfs, CONF_DNS_ENABLED);
      Assert.fail(EXCEPTION_PREFIX + "service with resource profile only");
    } catch (IllegalArgumentException e) {
      assertEquals(ERROR_RESOURCE_PROFILE_NOT_SUPPORTED_YET,
          e.getMessage());
    }

    // unset profile here and add cpus/memory back
    res.setProfile(null);
    res.setCpus(2);
    res.setMemory("2gb");

    // null number of containers
    try {
      ServiceApiUtil.validateAndResolveService(app, sfs, CONF_DNS_ENABLED);
      Assert.fail(EXCEPTION_PREFIX + "null number of containers");
    } catch (IllegalArgumentException e) {
      Assert.assertTrue(e.getMessage()
          .startsWith(ERROR_CONTAINERS_COUNT_INVALID));
    }
  }