@Test
public void testNodeRegistrationWithAttributes() throws Exception {

  writeToHostsFile("host2");
  Configuration conf = new Configuration();
  conf.set(YarnConfiguration.RM_NODES_INCLUDE_FILE_PATH,
      hostFile.getAbsolutePath());
  conf.setClass(YarnConfiguration.FS_NODE_ATTRIBUTE_STORE_IMPL_CLASS,
      FileSystemNodeAttributeStore.class, NodeAttributeStore.class);
  File tempDir = File.createTempFile("nattr", ".tmp");
  tempDir.delete();
  tempDir.mkdirs();
  tempDir.deleteOnExit();
  conf.set(YarnConfiguration.FS_NODE_ATTRIBUTE_STORE_ROOT_DIR,
      tempDir.getAbsolutePath());
  rm = new MockRM(conf);
  rm.start();
  ResourceTrackerService resourceTrackerService =
      rm.getResourceTrackerService();
  RegisterNodeManagerRequest registerReq =
      Records.newRecord(RegisterNodeManagerRequest.class);
  NodeId nodeId = NodeId.newInstance("host2", 1234);
  Resource capability = BuilderUtils.newResource(1024, 1);
  NodeAttribute nodeAttribute1 = NodeAttribute
      .newInstance(NodeAttribute.PREFIX_DISTRIBUTED, "Attr1",
          NodeAttributeType.STRING, "V1");
  NodeAttribute nodeAttribute2 = NodeAttribute
      .newInstance(NodeAttribute.PREFIX_DISTRIBUTED, "Attr2",
          NodeAttributeType.STRING, "V2");
  registerReq.setResource(capability);
  registerReq.setNodeId(nodeId);
  registerReq.setHttpPort(1234);
  registerReq.setNMVersion(YarnVersionInfo.getVersion());
  registerReq.setNodeAttributes(toSet(nodeAttribute1, nodeAttribute2));
  RegisterNodeManagerResponse response =
      resourceTrackerService.registerNodeManager(registerReq);
  Assert.assertEquals("Action should be normal on valid Node Attributes",
      NodeAction.NORMAL, response.getNodeAction());
  Assert.assertTrue(NodeLabelUtil.isNodeAttributesEquals(
      rm.getRMContext().getNodeAttributesManager()
          .getAttributesForNode(nodeId.getHost()).keySet(),
      registerReq.getNodeAttributes()));
  Assert.assertTrue("Valid Node Attributes were not accepted by RM",
      response.getAreNodeAttributesAcceptedByRM());
  if (rm != null) {
    rm.stop();
  }
}