@VisibleForTesting
  public Set<NodeAttribute> parseAttributes(String config)
      throws IOException {
    if (Strings.isNullOrEmpty(config)) {
      return ImmutableSet.of();
    }
    Set<NodeAttribute> attributeSet = new HashSet<>();
    // Configuration value should be in one line, format:
    // "ATTRIBUTE_NAME,ATTRIBUTE_TYPE,ATTRIBUTE_VALUE",
    // multiple node-attributes are delimited by ":".
    // Each attribute str should not container any space.
    String[] attributeStrs = config.split(NODE_ATTRIBUTES_DELIMITER);
    for (String attributeStr : attributeStrs) {
      String[] fields = attributeStr.split(NODE_ATTRIBUTE_DELIMITER);
      if (fields.length != 3) {
        throw new IOException("Invalid value for "
            + YarnConfiguration.NM_PROVIDER_CONFIGURED_NODE_ATTRIBUTES
            + "=" + config);
      }

      // We don't allow user config to overwrite our dist prefix,
      // so disallow any prefix set in the configuration.
      if (fields[0].contains("/")) {
        throw new IOException("Node attribute set in "
            + YarnConfiguration.NM_PROVIDER_CONFIGURED_NODE_ATTRIBUTES
            + " should not contain any prefix.");
      }

      // Make sure attribute type is valid.
      if (!EnumUtils.isValidEnum(NodeAttributeType.class, fields[1])) {
        throw new IOException("Invalid node attribute type: "
            + fields[1] + ", valid values are "
            + Arrays.asList(NodeAttributeType.values()));
      }

      // Automatically setup prefix for collected attributes
      NodeAttribute na = NodeAttribute.newInstance(
          NodeAttribute.PREFIX_DISTRIBUTED,
          fields[0],
          NodeAttributeType.valueOf(fields[1]),
          fields[2]);

      // Since a NodeAttribute is identical with another one as long as
      // their prefix and name are same, to avoid attributes getting
      // overwritten by ambiguous attribute, make sure it fails in such
      // case.
      if (!attributeSet.add(na)) {
        throw new IOException("Ambiguous node attribute is found: "
            + na.toString() + ", a same attribute already exists");
      }
    }

    // Before updating the attributes to the provider,
    // verify if they are valid
    try {
      NodeLabelUtil.validateNodeAttributes(attributeSet);
    } catch (IOException e) {
      throw new IOException("Node attributes set by configuration property: "
          + YarnConfiguration.NM_PROVIDER_CONFIGURED_NODE_ATTRIBUTES
          + " is not valid. Detail message: " + e.getMessage());
    }
    return attributeSet;
  }
@Test
  public void testParseConfiguration() throws IOException {
    // ATTRIBUTE_NAME,ATTRIBUTE_TYPE,ATTRIBUTE_VALUE
    String attributesStr = "hostname,STRING,host1234:uptime,STRING,321543";
    Set<NodeAttribute> attributes = nodeAttributesProvider
        .parseAttributes(attributesStr);
    Assert.assertEquals(2, attributes.size());
    Iterator<NodeAttribute> ait = attributes.iterator();

    while(ait.hasNext()) {
      NodeAttribute attr = ait.next();
      NodeAttributeKey at = attr.getAttributeKey();
      if (at.getAttributeName().equals("hostname")) {
        Assert.assertEquals("hostname", at.getAttributeName());
        Assert.assertEquals(NodeAttribute.PREFIX_DISTRIBUTED,
            at.getAttributePrefix());
        Assert.assertEquals(NodeAttributeType.STRING,
            attr.getAttributeType());
        Assert.assertEquals("host1234", attr.getAttributeValue());
      } else if (at.getAttributeName().equals("uptime")) {
        Assert.assertEquals("uptime", at.getAttributeName());
        Assert.assertEquals(NodeAttribute.PREFIX_DISTRIBUTED,
            at.getAttributePrefix());
        Assert.assertEquals(NodeAttributeType.STRING,
            attr.getAttributeType());
        Assert.assertEquals("321543", attr.getAttributeValue());
      } else {
        Assert.fail("Unexpected attribute");
      }
    }
    // Missing type
    attributesStr = "hostname,host1234";
    try {
      nodeAttributesProvider.parseAttributes(attributesStr);
      Assert.fail("Expecting a parsing failure");
    } catch (IOException e) {
      Assert.assertNotNull(e);
      Assert.assertTrue(e.getMessage().contains("Invalid value"));
    }

    // Extra prefix
    attributesStr = "prefix/hostname,STRING,host1234";
    try {
      nodeAttributesProvider.parseAttributes(attributesStr);
      Assert.fail("Expecting a parsing failure");
    } catch (IOException e) {
      Assert.assertNotNull(e);
      Assert.assertTrue(e.getMessage()
          .contains("should not contain any prefix."));
    }

    // Invalid type
    attributesStr = "hostname,T,host1234";
    try {
      nodeAttributesProvider.parseAttributes(attributesStr);
      Assert.fail("Expecting a parsing failure");
    } catch (IOException e) {
      e.printStackTrace();
      Assert.assertNotNull(e);
      Assert.assertTrue(e.getMessage()
          .contains("Invalid node attribute type"));
    }
  }