ReportMerger implements Process { @Override public void run(PortBindings portBindings, Configuration conf, Map<String, String> parameters) throws Exception { FileSystem fs = FileSystem.get(conf); List<ReportEntry> allReportEntries = readAllPartialReports(fs, portBindings.getInput().get(PARTIAL_REPORTS_PORT_IN_NAME)); writeActionData(allReportEntries); JsonObject jsonReport = buildJsonReport(allReportEntries); writeJsonReport(jsonReport, fs, portBindings.getOutput().get(REPORT_PORT_OUT_NAME)); }  @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 { List<ReportEntry> partialReport1 = Lists.newArrayList( new ReportEntry("param1.paramA.I", ReportEntryType.COUNTER, "3"), new ReportEntry("param1.paramA.III", ReportEntryType.COUNTER, "6"), new ReportEntry("param1.paramB", ReportEntryType.COUNTER, "4"), new ReportEntry("param1.paramX", ReportEntryType.COUNTER, "33")); List<ReportEntry> partialReport2 = Lists.newArrayList( new ReportEntry("param2", ReportEntryType.COUNTER, "12"), new ReportEntry("param1.paramA.II", ReportEntryType.COUNTER, "2"), new ReportEntry("param1.paramC.II", ReportEntryType.COUNTER, "5"), new ReportEntry("param1.paramA.duration", ReportEntryType.DURATION, "1000"), new ReportEntry("param1.paramX.Z", ReportEntryType.COUNTER, "1333")); AvroTestUtils.createLocalAvroDataStore(partialReport1, inputPartialReportsBasePath + "/report1"); AvroTestUtils.createLocalAvroDataStore(partialReport2, inputPartialReportsBasePath + "/report2"); PortBindings portBindings = new PortBindings( ImmutableMap.of("partial_reports", new Path(inputPartialReportsBasePath)), ImmutableMap.of("report", new Path(outputReportPath))); Configuration conf = new Configuration(false); reportMerger.run(portBindings, conf, Collections.emptyMap()); JsonObject actualReportJson = readJson(outputReportPath); JsonObject expectedJsonReport = readJsonFromClasspath("/eu/dnetlib/iis/common/report/report_merged.json"); assertEquals(expectedJsonReport, actualReportJson); Properties actionData = getStoredProperties(); assertNotNull(actionData); assertEquals(8, actionData.size()); assertEquals("3", actionData.getProperty("param1.paramA.I")); assertEquals("6", actionData.getProperty("param1.paramA.III")); assertEquals("4", actionData.getProperty("param1.paramB")); assertEquals("33", actionData.getProperty("param1.paramX")); assertEquals("12", actionData.getProperty("param2")); assertEquals("2", actionData.getProperty("param1.paramA.II")); assertEquals("5", actionData.getProperty("param1.paramC.II")); assertEquals("1333", actionData.getProperty("param1.paramX.Z")); }