@Test public void testParseStaleDatanodeListMultipleDatanodes() throws Exception { String json = "{" + "\"1.2.3.4:1\": {\"numBlocks\": 0}, " + "\"1.2.3.4:2\": {\"numBlocks\": 15}, " + "\"1.2.3.4:3\": {\"numBlocks\": 5}, " + "\"1.2.3.4:4\": {\"numBlocks\": 10} " + "}"; Set<String> out = DynoInfraUtils.parseStaleDataNodeList(json, 10, LOG); assertEquals(2, out.size()); assertTrue(out.contains("1.2.3.4:1")); assertTrue(out.contains("1.2.3.4:3")); }
static Set<String> parseStaleDataNodeList(String liveNodeJsonString, final int blockThreshold, final Log log) throws IOException { final Set<String> dataNodesToReport = new HashSet<>(); JsonFactory fac = new JsonFactory(); JsonParser parser = fac.createJsonParser(IOUtils.toInputStream(liveNodeJsonString, StandardCharsets.UTF_8.name())); int objectDepth = 0; String currentNodeAddr = null; for (JsonToken tok = parser.nextToken(); tok != null; tok = parser.nextToken()) { if (tok == JsonToken.START_OBJECT) { objectDepth++; } else if (tok == JsonToken.END_OBJECT) { objectDepth--; } else if (tok == JsonToken.FIELD_NAME) { if (objectDepth == 1) { currentNodeAddr = parser.getCurrentName(); } else if (objectDepth == 2) { if (parser.getCurrentName().equals("numBlocks")) { JsonToken valueToken = parser.nextToken(); if (valueToken != JsonToken.VALUE_NUMBER_INT || currentNodeAddr == null) { throw new IOException(String.format("Malformed LiveNodes JSON; got token = %s; currentNodeAddr = %s: %s", valueToken, currentNodeAddr, liveNodeJsonString)); } int numBlocks = parser.getIntValue(); if (numBlocks < blockThreshold) { log.debug(String.format("Queueing Datanode <%s> for block report; numBlocks = %d", currentNodeAddr, numBlocks)); dataNodesToReport.add(currentNodeAddr); } else { log.debug(String.format("Not queueing Datanode <%s> for block report; numBlocks = %d", currentNodeAddr, numBlocks)); } } } } } return dataNodesToReport; }
DynoInfraUtils { static Set<String> parseStaleDataNodeList(String liveNodeJsonString, final int blockThreshold, final Log log) throws IOException { final Set<String> dataNodesToReport = new HashSet<>(); JsonFactory fac = new JsonFactory(); JsonParser parser = fac.createJsonParser(IOUtils.toInputStream(liveNodeJsonString, StandardCharsets.UTF_8.name())); int objectDepth = 0; String currentNodeAddr = null; for (JsonToken tok = parser.nextToken(); tok != null; tok = parser.nextToken()) { if (tok == JsonToken.START_OBJECT) { objectDepth++; } else if (tok == JsonToken.END_OBJECT) { objectDepth--; } else if (tok == JsonToken.FIELD_NAME) { if (objectDepth == 1) { currentNodeAddr = parser.getCurrentName(); } else if (objectDepth == 2) { if (parser.getCurrentName().equals("numBlocks")) { JsonToken valueToken = parser.nextToken(); if (valueToken != JsonToken.VALUE_NUMBER_INT || currentNodeAddr == null) { throw new IOException(String.format("Malformed LiveNodes JSON; got token = %s; currentNodeAddr = %s: %s", valueToken, currentNodeAddr, liveNodeJsonString)); } int numBlocks = parser.getIntValue(); if (numBlocks < blockThreshold) { log.debug(String.format("Queueing Datanode <%s> for block report; numBlocks = %d", currentNodeAddr, numBlocks)); dataNodesToReport.add(currentNodeAddr); } else { log.debug(String.format("Not queueing Datanode <%s> for block report; numBlocks = %d", currentNodeAddr, numBlocks)); } } } } } return dataNodesToReport; } }
DynoInfraUtils { static Set<String> parseStaleDataNodeList(String liveNodeJsonString, final int blockThreshold, final Log log) throws IOException { final Set<String> dataNodesToReport = new HashSet<>(); JsonFactory fac = new JsonFactory(); JsonParser parser = fac.createJsonParser(IOUtils.toInputStream(liveNodeJsonString, StandardCharsets.UTF_8.name())); int objectDepth = 0; String currentNodeAddr = null; for (JsonToken tok = parser.nextToken(); tok != null; tok = parser.nextToken()) { if (tok == JsonToken.START_OBJECT) { objectDepth++; } else if (tok == JsonToken.END_OBJECT) { objectDepth--; } else if (tok == JsonToken.FIELD_NAME) { if (objectDepth == 1) { currentNodeAddr = parser.getCurrentName(); } else if (objectDepth == 2) { if (parser.getCurrentName().equals("numBlocks")) { JsonToken valueToken = parser.nextToken(); if (valueToken != JsonToken.VALUE_NUMBER_INT || currentNodeAddr == null) { throw new IOException(String.format("Malformed LiveNodes JSON; got token = %s; currentNodeAddr = %s: %s", valueToken, currentNodeAddr, liveNodeJsonString)); } int numBlocks = parser.getIntValue(); if (numBlocks < blockThreshold) { log.debug(String.format("Queueing Datanode <%s> for block report; numBlocks = %d", currentNodeAddr, numBlocks)); dataNodesToReport.add(currentNodeAddr); } else { log.debug(String.format("Not queueing Datanode <%s> for block report; numBlocks = %d", currentNodeAddr, numBlocks)); } } } } } return dataNodesToReport; }  }
DynoInfraUtils { static Set<String> parseStaleDataNodeList(String liveNodeJsonString, final int blockThreshold, final Log log) throws IOException { final Set<String> dataNodesToReport = new HashSet<>(); JsonFactory fac = new JsonFactory(); JsonParser parser = fac.createJsonParser(IOUtils.toInputStream(liveNodeJsonString, StandardCharsets.UTF_8.name())); int objectDepth = 0; String currentNodeAddr = null; for (JsonToken tok = parser.nextToken(); tok != null; tok = parser.nextToken()) { if (tok == JsonToken.START_OBJECT) { objectDepth++; } else if (tok == JsonToken.END_OBJECT) { objectDepth--; } else if (tok == JsonToken.FIELD_NAME) { if (objectDepth == 1) { currentNodeAddr = parser.getCurrentName(); } else if (objectDepth == 2) { if (parser.getCurrentName().equals("numBlocks")) { JsonToken valueToken = parser.nextToken(); if (valueToken != JsonToken.VALUE_NUMBER_INT || currentNodeAddr == null) { throw new IOException(String.format("Malformed LiveNodes JSON; got token = %s; currentNodeAddr = %s: %s", valueToken, currentNodeAddr, liveNodeJsonString)); } int numBlocks = parser.getIntValue(); if (numBlocks < blockThreshold) { log.debug(String.format("Queueing Datanode <%s> for block report; numBlocks = %d", currentNodeAddr, numBlocks)); dataNodesToReport.add(currentNodeAddr); } else { log.debug(String.format("Not queueing Datanode <%s> for block report; numBlocks = %d", currentNodeAddr, numBlocks)); } } } } } return dataNodesToReport; }  static File fetchHadoopTarball(File destinationDir, String version, Configuration conf, Log log); }
DynoInfraUtils { static Set<String> parseStaleDataNodeList(String liveNodeJsonString, final int blockThreshold, final Log log) throws IOException { final Set<String> dataNodesToReport = new HashSet<>(); JsonFactory fac = new JsonFactory(); JsonParser parser = fac.createJsonParser(IOUtils.toInputStream(liveNodeJsonString, StandardCharsets.UTF_8.name())); int objectDepth = 0; String currentNodeAddr = null; for (JsonToken tok = parser.nextToken(); tok != null; tok = parser.nextToken()) { if (tok == JsonToken.START_OBJECT) { objectDepth++; } else if (tok == JsonToken.END_OBJECT) { objectDepth--; } else if (tok == JsonToken.FIELD_NAME) { if (objectDepth == 1) { currentNodeAddr = parser.getCurrentName(); } else if (objectDepth == 2) { if (parser.getCurrentName().equals("numBlocks")) { JsonToken valueToken = parser.nextToken(); if (valueToken != JsonToken.VALUE_NUMBER_INT || currentNodeAddr == null) { throw new IOException(String.format("Malformed LiveNodes JSON; got token = %s; currentNodeAddr = %s: %s", valueToken, currentNodeAddr, liveNodeJsonString)); } int numBlocks = parser.getIntValue(); if (numBlocks < blockThreshold) { log.debug(String.format("Queueing Datanode <%s> for block report; numBlocks = %d", currentNodeAddr, numBlocks)); dataNodesToReport.add(currentNodeAddr); } else { log.debug(String.format("Not queueing Datanode <%s> for block report; numBlocks = %d", currentNodeAddr, numBlocks)); } } } } } return dataNodesToReport; }  static File fetchHadoopTarball(File destinationDir, String version, Configuration conf, Log log); static final String DYNO_CONF_PREFIX; static final String DYNO_INFRA_PREFIX; static final String APACHE_DOWNLOAD_MIRROR_KEY; static final String APACHE_DOWNLOAD_MIRROR_DEFAULT; static final String HADOOP_TAR_FILENAME_FORMAT; static final String DATANODE_LIVE_MIN_FRACTION_KEY; static final float DATANODE_LIVE_MIN_FRACTION_DEFAULT; static final String MISSING_BLOCKS_MAX_FRACTION_KEY; static final float MISSING_BLOCKS_MAX_FRACTION_DEFAULT; static final String UNDERREPLICATED_BLOCKS_MAX_FRACTION_KEY; static final float UNDERREPLICATED_BLOCKS_MAX_FRACTION_DEFAULT; static final String NAMENODE_STARTUP_PROGRESS_JMX_QUERY; static final String FSNAMESYSTEM_JMX_QUERY; static final String FSNAMESYSTEM_STATE_JMX_QUERY; static final String NAMENODE_INFO_JMX_QUERY; static final String JMX_MISSING_BLOCKS; static final String JMX_UNDER_REPLICATED_BLOCKS; static final String JMX_BLOCKS_TOTAL; static final String JMX_LIVE_NODE_COUNT; static final String JMX_LIVE_NODES_LIST; }