@Test
public void testParquetReadingWithParquetToPojoMapping() throws Exception
{

  FileContext.getLocalFSFileContext().delete(new Path(new File(testMeta.dir).getAbsolutePath()), true);
  List<EventRecord> data = Lists.newArrayList();
  data.add(new EventRecord(1, "cust1", 12321L, true, 12.22f, 12.23));
  data.add(new EventRecord(2, "cust2", 12322L, true, 22.22f, 22.23));
  data.add(new EventRecord(3, "cust3", 12323L, true, 32.22f, 32.23));
  writeParquetFile(PARQUET_SCHEMA, new File(testMeta.dir, "data.parquet"), data);
  parquetFilePOJOReader.output.setSink(outputSink);
  parquetFilePOJOReader.setDirectory(testMeta.dir);
  parquetFilePOJOReader.setParquetSchema(PARQUET_SCHEMA);
  parquetFilePOJOReader.setParquetToPOJOFieldsMapping(
      "event_id:event_id_v2:INTEGER,org_id:org_id_v2:STRING,long_id:long_id_v2:LONG,css_file_loaded:css_file_loaded_v2:BOOLEAN,float_val:float_val_v2:FLOAT,double_val:double_val_v2:DOUBLE");
  parquetFilePOJOReader.setup(testMeta.context);
  testMeta.portContext.getAttributes().put(Context.PortContext.TUPLE_CLASS, EventRecordV2.class);
  parquetFilePOJOReader.output.setup(testMeta.portContext);
  for (long wid = 0; wid < 2; wid++) {
    parquetFilePOJOReader.beginWindow(0);
    parquetFilePOJOReader.emitTuples();
    parquetFilePOJOReader.endWindow();
  }
  Assert.assertEquals("number tuples", 3, outputSink.collectedTuples.size());
  parquetFilePOJOReader.teardown();
}