@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 testDisableFetchNodeAttributes() 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 -1 to disable refresh.
    conf.setLong(
        YarnConfiguration.NM_NODE_ATTRIBUTES_PROVIDER_FETCH_INTERVAL_MS, -1);
    ConfigurationNodeAttributesProvider spyProvider =
        Mockito.spy(nodeAttributesProvider);
    Mockito.when(spyProvider.parseAttributes(Mockito.any()))
        .thenReturn(expectedAttributes1);
    spyProvider.init(conf);
    spyProvider.start();
    Assert.assertEquals(expectedAttributes1,
        spyProvider.getDescriptors());

    // The configuration added another attribute,
    // as we disabled the fetch interval, this value cannot be
    // updated to the provider.
    Set<NodeAttribute> expectedAttributes2 = new HashSet<>();
    expectedAttributes2.add(NodeAttribute
        .newInstance("test.io", "os",
            NodeAttributeType.STRING, "windows"));
    Mockito.when(spyProvider.parseAttributes(Mockito.anyString()))
        .thenReturn(expectedAttributes2);

    // Wait a few seconds until we get the value update, expecting a failure.
    try {
      GenericTestUtils.waitFor(() -> {
        Set<NodeAttribute> attributes = spyProvider.getDescriptors();
        return "os".equalsIgnoreCase(attributes
            .iterator().next().getAttributeKey().getAttributeName());
      }, 500, 1000);
    } catch (Exception e) {
      // Make sure we get the timeout exception.
      Assert.assertTrue(e instanceof TimeoutException);
      return;
    }

    Assert.fail("Expecting a failure in previous check!");
  }