@Test
public void testConvertToStream() throws Exception {

    Map<String, Map<String, Record>> byTime = new TreeMap<>();
    Record record1 = new Record();
    record1.attributes = new TreeMap<>();
    record1.attributes.put("attr1", "foo");
    record1.attributes.put("attr2", "bar");
    record1.values = new HashMap<>();
    record1.values.put("custom.googleapis.com/test", 77.0);
    Record record2 = new Record();
    record2.attributes = new TreeMap<>();
    record2.attributes.put("attr1", "foo");
    record2.attributes.put("attr2", "baz");
    record2.values = new HashMap<>();
    record2.values.put("custom.googleapis.com/test", 88.0);
    // This record is incomplete and should be skipped.
    Record record3 = new Record();
    record3.attributes = new TreeMap<>();
    record3.attributes.put("attr1", "foo");
    record3.attributes.put("attr2", "bar");
    record3.values = new HashMap<>();
    Record record4 = new Record();
    record4.attributes = new TreeMap<>();
    record4.attributes.put("attr1", "foo");
    record4.attributes.put("attr2", "bar");
    record4.values = new HashMap<>();
    record4.values.put("custom.googleapis.com/test", 99.0);
    Map<String, Record> recordMap1 = new HashMap<>();
    recordMap1.put("attr-map-key1", record1);
    recordMap1.put("attr-map-key2", record2);
    Map<String, Record> recordMap2 = new HashMap<>();
    recordMap2.put("attr-map-key1", record3);
    Map<String, Record> recordMap3 = new HashMap<>();
    recordMap3.put("attr-map-key1", record4);
    byTime.put("2016-08-08T00:00:00Z", recordMap1);
    byTime.put("2016-08-08T00:00:01Z", recordMap2);
    byTime.put("2016-08-08T00:00:02Z", recordMap3);
    GoogleMonitoringIngester ingester = new GoogleMonitoringIngester(conf);
    MBStream<Datum> stream = ingester.convertToStream(byTime);
    List<Datum> data = stream.drain();
    assertEquals(3, data.size(), 0.0);
    assertEquals(77, data.get(0).metrics().getEntry(0), 0.0);
    assertEquals(0, data.get(0).attributes().get(0), 0.0);
    assertEquals(1, data.get(0).attributes().get(1), 0.0);
    assertEquals(88, data.get(1).metrics().getEntry(0), 0.0);
    assertEquals(0, data.get(1).attributes().get(0), 0.0);
    assertEquals(2, data.get(1).attributes().get(1), 0.0);
    assertEquals(99, data.get(2).metrics().getEntry(0), 0.0);
    assertEquals(0, data.get(2).attributes().get(0), 0.0);
    assertEquals(1, data.get(2).attributes().get(1), 0.0);
}