FileSystemNodeLabelsStore extends NodeLabelsStore { @Override public void init(Configuration conf) throws Exception { fsWorkingPath = new Path(conf.get(YarnConfiguration.FS_NODE_LABELS_STORE_ROOT_DIR, getDefaultFSNodeLabelsRootDir())); setFileSystem(conf); if (!fs.exists(fsWorkingPath)) { fs.mkdirs(fsWorkingPath); } } FileSystemNodeLabelsStore(CommonNodeLabelsManager mgr); @Override void init(Configuration conf); @Override void close(); @Override void updateNodeToLabelsMappings(
      Map<NodeId, Set<String>> nodeToLabels); @Override void storeNewClusterNodeLabels(Set<String> labels); @Override void removeClusterNodeLabels(Collection<String> labels); @Override void recover();  }
@SuppressWarnings({ "unchecked", "rawtypes" }) @Test(timeout = 10000) public void testRecoverWithMirror() throws Exception { mgr.addToCluserNodeLabels(toSet("p1", "p2", "p3")); mgr.addToCluserNodeLabels(toSet("p4")); mgr.addToCluserNodeLabels(toSet("p5", "p6")); mgr.replaceLabelsOnNode((Map) ImmutableMap.of(toNodeId("n1"), toSet("p1"), toNodeId("n2"), toSet("p2"))); mgr.replaceLabelsOnNode((Map) ImmutableMap.of(toNodeId("n3"), toSet("p3"), toNodeId("n4"), toSet("p4"), toNodeId("n5"), toSet("p5"), toNodeId("n6"), toSet("p6"), toNodeId("n7"), toSet("p6"))); mgr.removeFromClusterNodeLabels(toSet("p1")); mgr.removeFromClusterNodeLabels(Arrays.asList("p3", "p5")); mgr.stop(); mgr = new MockNodeLabelManager(); mgr.init(conf); mgr.start(); Assert.assertEquals(3, mgr.getClusterNodeLabels().size()); Assert.assertTrue(mgr.getClusterNodeLabels().containsAll( Arrays.asList("p2", "p4", "p6"))); assertMapContains(mgr.getNodeLabels(), ImmutableMap.of(toNodeId("n2"), toSet("p2"), toNodeId("n4"), toSet("p4"), toNodeId("n6"), toSet("p6"), toNodeId("n7"), toSet("p6"))); assertLabelsToNodesEquals(mgr.getLabelsToNodes(), ImmutableMap.of( "p6", toSet(toNodeId("n6"), toNodeId("n7")), "p4", toSet(toNodeId("n4")), "p2", toSet(toNodeId("n2")))); mgr.stop(); mgr = new MockNodeLabelManager(); mgr.init(conf); mgr.start(); Assert.assertEquals(3, mgr.getClusterNodeLabels().size()); Assert.assertTrue(mgr.getClusterNodeLabels().containsAll( Arrays.asList("p2", "p4", "p6"))); assertMapContains(mgr.getNodeLabels(), ImmutableMap.of(toNodeId("n2"), toSet("p2"), toNodeId("n4"), toSet("p4"), toNodeId("n6"), toSet("p6"), toNodeId("n7"), toSet("p6"))); assertLabelsToNodesEquals(mgr.getLabelsToNodes(), ImmutableMap.of( "p6", toSet(toNodeId("n6"), toNodeId("n7")), "p4", toSet(toNodeId("n4")), "p2", toSet(toNodeId("n2")))); mgr.stop(); }
@SuppressWarnings({ "unchecked", "rawtypes" }) @Test(timeout = 10000) public void testEditlogRecover() throws Exception { mgr.addToCluserNodeLabels(toSet("p1", "p2", "p3")); mgr.addToCluserNodeLabels(toSet("p4")); mgr.addToCluserNodeLabels(toSet("p5", "p6")); mgr.replaceLabelsOnNode(ImmutableMap.of(toNodeId("n1"), toSet("p1"), toNodeId("n2"), toSet("p2"))); mgr.replaceLabelsOnNode((Map) ImmutableMap.of(toNodeId("n3"), toSet("p3"), toNodeId("n4"), toSet("p4"), toNodeId("n5"), toSet("p5"), toNodeId("n6"), toSet("p6"), toNodeId("n7"), toSet("p6"))); mgr.removeFromClusterNodeLabels(toSet("p1")); mgr.removeFromClusterNodeLabels(Arrays.asList("p3", "p5")); mgr.stop(); mgr = new MockNodeLabelManager(); mgr.init(conf); mgr.start(); Assert.assertEquals(3, mgr.getClusterNodeLabels().size()); Assert.assertTrue(mgr.getClusterNodeLabels().containsAll( Arrays.asList("p2", "p4", "p6"))); assertMapContains(mgr.getNodeLabels(), ImmutableMap.of(toNodeId("n2"), toSet("p2"), toNodeId("n4"), toSet("p4"), toNodeId("n6"), toSet("p6"), toNodeId("n7"), toSet("p6"))); assertLabelsToNodesEquals(mgr.getLabelsToNodes(), ImmutableMap.of( "p6", toSet(toNodeId("n6"), toNodeId("n7")), "p4", toSet(toNodeId("n4")), "p2", toSet(toNodeId("n2")))); mgr.stop(); }
@SuppressWarnings({ "unchecked", "rawtypes" }) @Test public void testSerilizationAfterRecovery() throws Exception { mgr.addToCluserNodeLabels(toSet("p1", "p2", "p3")); mgr.addToCluserNodeLabels(toSet("p4")); mgr.addToCluserNodeLabels(toSet("p5", "p6")); mgr.replaceLabelsOnNode(ImmutableMap.of(toNodeId("n1"), toSet("p1"), toNodeId("n2"), toSet("p2"))); mgr.replaceLabelsOnNode((Map) ImmutableMap.of(toNodeId("n3"), toSet("p3"), toNodeId("n4"), toSet("p4"), toNodeId("n5"), toSet("p5"), toNodeId("n6"), toSet("p6"), toNodeId("n7"), toSet("p6"))); mgr.removeFromClusterNodeLabels(toSet("p1")); mgr.removeFromClusterNodeLabels(Arrays.asList("p3", "p5")); mgr.stop(); mgr = new MockNodeLabelManager(); mgr.init(conf); mgr.start(); Assert.assertEquals(3, mgr.getClusterNodeLabels().size()); Assert.assertTrue(mgr.getClusterNodeLabels().containsAll( Arrays.asList("p2", "p4", "p6"))); assertMapContains(mgr.getNodeLabels(), ImmutableMap.of(toNodeId("n2"), toSet("p2"), toNodeId("n4"), toSet("p4"), toNodeId("n6"), toSet("p6"), toNodeId("n7"), toSet("p6"))); assertLabelsToNodesEquals(mgr.getLabelsToNodes(), ImmutableMap.of( "p6", toSet(toNodeId("n6"), toNodeId("n7")), "p4", toSet(toNodeId("n4")), "p2", toSet(toNodeId("n2")))); mgr = new MockNodeLabelManager(); mgr.init(conf); mgr.start(); mgr.addToCluserNodeLabels(toSet("p7", "p8")); mgr.stop(); mgr = new MockNodeLabelManager(); mgr.init(conf); mgr.start(); mgr.addToCluserNodeLabels(toSet("p9")); mgr.stop(); mgr = new MockNodeLabelManager(); mgr.init(conf); mgr.start(); Assert.assertEquals(6, mgr.getClusterNodeLabels().size()); Assert.assertTrue(mgr.getClusterNodeLabels().containsAll( Arrays.asList("p2", "p4", "p6", "p7", "p8", "p9"))); mgr.stop(); }
@SuppressWarnings({ "unchecked", "rawtypes" }) @Test(timeout = 10000) public void testRecoverWithMirror() throws Exception { mgr.addToCluserNodeLabels(toSet("p1", "p2", "p3")); mgr.addToCluserNodeLabels(toSet("p4")); mgr.addToCluserNodeLabels(toSet("p5", "p6")); mgr.replaceLabelsOnNode((Map) ImmutableMap.of(toNodeId("n1"), toSet("p1"), toNodeId("n2"), toSet("p2"))); mgr.replaceLabelsOnNode((Map) ImmutableMap.of(toNodeId("n3"), toSet("p3"), toNodeId("n4"), toSet("p4"), toNodeId("n5"), toSet("p5"), toNodeId("n6"), toSet("p6"), toNodeId("n7"), toSet("p6"))); mgr.removeFromClusterNodeLabels(toSet("p1")); mgr.removeFromClusterNodeLabels(Arrays.asList("p3", "p5")); mgr.stop(); mgr = new MockNodeLabelManager(); mgr.init(conf); Assert.assertEquals(3, mgr.getClusterNodeLabels().size()); Assert.assertTrue(mgr.getClusterNodeLabels().containsAll( Arrays.asList("p2", "p4", "p6"))); assertMapContains(mgr.getNodeLabels(), ImmutableMap.of(toNodeId("n2"), toSet("p2"), toNodeId("n4"), toSet("p4"), toNodeId("n6"), toSet("p6"), toNodeId("n7"), toSet("p6"))); mgr.stop(); mgr = new MockNodeLabelManager(); mgr.init(conf); Assert.assertEquals(3, mgr.getClusterNodeLabels().size()); Assert.assertTrue(mgr.getClusterNodeLabels().containsAll( Arrays.asList("p2", "p4", "p6"))); assertMapContains(mgr.getNodeLabels(), ImmutableMap.of(toNodeId("n2"), toSet("p2"), toNodeId("n4"), toSet("p4"), toNodeId("n6"), toSet("p6"), toNodeId("n7"), toSet("p6"))); mgr.stop(); }
@SuppressWarnings({ "unchecked", "rawtypes" }) @Test(timeout = 10000) public void testEditlogRecover() throws Exception { mgr.addToCluserNodeLabels(toSet("p1", "p2", "p3")); mgr.addToCluserNodeLabels(toSet("p4")); mgr.addToCluserNodeLabels(toSet("p5", "p6")); mgr.replaceLabelsOnNode(ImmutableMap.of(toNodeId("n1"), toSet("p1"), toNodeId("n2"), toSet("p2"))); mgr.replaceLabelsOnNode((Map) ImmutableMap.of(toNodeId("n3"), toSet("p3"), toNodeId("n4"), toSet("p4"), toNodeId("n5"), toSet("p5"), toNodeId("n6"), toSet("p6"), toNodeId("n7"), toSet("p6"))); mgr.removeFromClusterNodeLabels(toSet("p1")); mgr.removeFromClusterNodeLabels(Arrays.asList("p3", "p5")); mgr.stop(); mgr = new MockNodeLabelManager(); mgr.init(conf); Assert.assertEquals(3, mgr.getClusterNodeLabels().size()); Assert.assertTrue(mgr.getClusterNodeLabels().containsAll( Arrays.asList("p2", "p4", "p6"))); assertMapContains(mgr.getNodeLabels(), ImmutableMap.of(toNodeId("n2"), toSet("p2"), toNodeId("n4"), toSet("p4"), toNodeId("n6"), toSet("p6"), toNodeId("n7"), toSet("p6"))); mgr.stop(); }
@SuppressWarnings({ "unchecked", "rawtypes" }) @Test public void testSerilizationAfterRecovery() throws Exception { mgr.addToCluserNodeLabels(toSet("p1", "p2", "p3")); mgr.addToCluserNodeLabels(toSet("p4")); mgr.addToCluserNodeLabels(toSet("p5", "p6")); mgr.replaceLabelsOnNode(ImmutableMap.of(toNodeId("n1"), toSet("p1"), toNodeId("n2"), toSet("p2"))); mgr.replaceLabelsOnNode((Map) ImmutableMap.of(toNodeId("n3"), toSet("p3"), toNodeId("n4"), toSet("p4"), toNodeId("n5"), toSet("p5"), toNodeId("n6"), toSet("p6"), toNodeId("n7"), toSet("p6"))); mgr.removeFromClusterNodeLabels(toSet("p1")); mgr.removeFromClusterNodeLabels(Arrays.asList("p3", "p5")); mgr.stop(); mgr = new MockNodeLabelManager(); mgr.init(conf); mgr.start(); Assert.assertEquals(3, mgr.getClusterNodeLabels().size()); Assert.assertTrue(mgr.getClusterNodeLabels().containsAll( Arrays.asList("p2", "p4", "p6"))); assertMapContains(mgr.getNodeLabels(), ImmutableMap.of(toNodeId("n2"), toSet("p2"), toNodeId("n4"), toSet("p4"), toNodeId("n6"), toSet("p6"), toNodeId("n7"), toSet("p6"))); mgr = new MockNodeLabelManager(); mgr.init(conf); mgr.start(); mgr.addToCluserNodeLabels(toSet("p7", "p8")); mgr.stop(); mgr = new MockNodeLabelManager(); mgr.init(conf); mgr.start(); mgr.addToCluserNodeLabels(toSet("p9")); mgr.stop(); mgr = new MockNodeLabelManager(); mgr.init(conf); mgr.start(); Assert.assertEquals(6, mgr.getClusterNodeLabels().size()); Assert.assertTrue(mgr.getClusterNodeLabels().containsAll( Arrays.asList("p2", "p4", "p6", "p7", "p8", "p9"))); mgr.stop(); }