OozieTimeReportGenerator implements Process { @Override public void run(PortBindings portBindings, Configuration conf, Map<String, String> parameters) throws Exception { List<WorkflowAction> actions = fetchWorkflowActions(parameters.get(OOZIE_SERVICE_LOC_PARAM), parameters.get(WORKFLOW_JOB_ID_PARAM)); Map<String, List<String>> reportKeysToActionNames = mapReportKeysToActionNames(parameters); List<ReportEntry> reportEntries = Lists.newArrayList(); for (Map.Entry<String, List<String>> reportKeyToActionNamesEntry : reportKeysToActionNames.entrySet()) { long totalDuration = 0L; for (String actionName : reportKeyToActionNamesEntry.getValue()) { totalDuration += fetchActionDuration(actions, actionName); } if (totalDuration > 0) { reportEntries.add(ReportEntryFactory.createDurationReportEntry(reportKeyToActionNamesEntry.getKey(), totalDuration)); } } FileSystem fs = FileSystem.get(conf); Path reportPath = portBindings.getOutput().get(REPORT_PORT_OUT_NAME); DataStore.create(reportEntries, 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 { PortBindings portBindings = createPortBindings(); Configuration conf = new Configuration(false); Map<String, String> parameters = ImmutableMap.of( "oozieServiceLoc", oozieUrl, "jobId", jobId, "report.group.first", "action_1,action_3,nonexisting_action_1", "report.group.second", "action_2", "report.group.nonexisting", "nonexisting_action_1, nonexisting_action_2"); when(oozieClientFactory.createOozieClient(oozieUrl)).thenReturn(oozieClient); oozieTimeReportGenerator.run(portBindings, conf, parameters); List<ReportEntry> actualReport = AvroTestUtils.readLocalAvroDataStore(tempFolder.getRoot().getPath()); assertThat(actualReport, containsInAnyOrder( new ReportEntry("group.first", ReportEntryType.DURATION, "3826000"), new ReportEntry("group.second", ReportEntryType.DURATION, "63000"))); }