ReportGenerator implements Process { @Override public void run(PortBindings portBindings, Configuration conf, Map<String, String> parameters) throws Exception { Map<String, String> entriesToReport = collectEntriesToReport(parameters); List<ReportEntry> avroReport = convertToAvroReport(entriesToReport); FileSystem fs = FileSystem.get(conf); Path reportPath = portBindings.getOutput().get(REPORT_PORT_OUT_NAME); DataStore.create(avroReport, new FileSystemPath(fs, reportPath)); }  @Override Map<String, PortType> getInputPorts(); @Override Map<String, PortType> getOutputPorts(); @Override void run(PortBindings portBindings, Configuration conf, Map<String, String> parameters);  }
@Test public void run() throws Exception { Path outputDirPath = new Path(tempFolder.getRoot().getPath()); PortBindings portBindings = new PortBindings(ImmutableMap.of(), ImmutableMap.of("report", outputDirPath)); Configuration conf = new Configuration(false); Map<String, String> parameters = ImmutableMap.of( "report.group.firstParam", "11", "report.group.secondParam", "22", "notReportPrefixed.group.thirdParam", "3123"); reportGenerator.run(portBindings, conf, parameters); List<ReportEntry> actualReportEntries = AvroTestUtils.readLocalAvroDataStore(tempFolder.getRoot().getPath()); assertThat(actualReportEntries, containsInAnyOrder( new ReportEntry("group.firstParam", ReportEntryType.COUNTER, "11"), new ReportEntry("group.secondParam", ReportEntryType.COUNTER, "22"))); }