@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(timeout=30000L)
  public void testNodeAttributesFetchInterval()
      throws IOException, InterruptedException {
    Set<NodeAttribute> expectedAttributes1 = new HashSet<>();
    expectedAttributes1.add(NodeAttribute
        .newInstance("test.io", "host",
            NodeAttributeType.STRING, "host1"));

    Configuration conf = new Configuration();
    // Set fetch interval to 1s for testing
    conf.setLong(
        YarnConfiguration.NM_NODE_ATTRIBUTES_PROVIDER_FETCH_INTERVAL_MS, 1000);
    ConfigurationNodeAttributesProvider spyProvider =
        Mockito.spy(nodeAttributesProvider);
    Mockito.when(spyProvider.parseAttributes(Mockito.any()))
        .thenReturn(expectedAttributes1);

    spyProvider.init(conf);
    spyProvider.start();

    // Verify init value is honored.
    Assert.assertEquals(expectedAttributes1, spyProvider.getDescriptors());

    // Configuration provider provides a different set of attributes.
    Set<NodeAttribute> expectedAttributes2 = new HashSet<>();
    expectedAttributes2.add(NodeAttribute
        .newInstance("test.io", "os",
            NodeAttributeType.STRING, "windows"));
    Mockito.when(spyProvider.parseAttributes(Mockito.any()))
        .thenReturn(expectedAttributes2);

    // Since we set fetch interval to 1s, it needs to wait for 1s until
    // the updated attributes is updated to the provider. So we are expecting
    // to see some old values for a short window.
    ArrayList<String> keysMet = new ArrayList<>();
    int numOfOldValue = 0;
    int numOfNewValue = 0;
    // Run 5 times in 500ms interval
    int times=5;
    while(times>0) {
      Set<NodeAttribute> current = spyProvider.getDescriptors();
      Assert.assertEquals(1, current.size());
      String attributeName =
          current.iterator().next().getAttributeKey().getAttributeName();
      if ("host".equals(attributeName)){
        numOfOldValue++;
      } else if ("os".equals(attributeName)) {
        numOfNewValue++;
      }
      Thread.sleep(500);
      times--;
    }
    // We should either see the old value or the new value.
    Assert.assertEquals(5, numOfNewValue + numOfOldValue);
    // Both values should be more than 0.
    Assert.assertTrue(numOfOldValue > 0);
    Assert.assertTrue(numOfNewValue > 0);
  }