@Test
/**
 * Test that when we encounter missing metrics we skip the entire row and read
 * all of the valid rows.
 */
public void testMissingData() throws Exception {

    MacroBaseConf conf = new MacroBaseConf();
    conf.set(MacroBaseConf.CSV_INPUT_FILE, "src/test/resources/data/missingdata.csv");
    conf.set(MacroBaseConf.ATTRIBUTES, Lists.newArrayList("a1"));
    conf.set(MacroBaseConf.METRICS, Lists.newArrayList("m1","m2"));
    CSVIngester ingester = new CSVIngester(conf);
    List<Datum> data = ingester.getStream().drain();
    int count = 0;
    for(Datum d : data) {
        if (count == 3) {
            assertEquals(5, d.metrics().getEntry(0), 0.0);
        }
        count++;
    }
    assertEquals(4, count);
}