OozieScheduling { public static void writeCoordinator(Schedule schedule, SchedulableJobManager manager, OutputStream output) throws IOException { XmlStreamWriter streamWriter = WriterFactory.newXmlWriter(output); PrettyPrintXMLWriter writer = new PrettyPrintXMLWriter(streamWriter); String jobName = schedule.getJobClass().getCanonicalName(); writer.startElement(COORDINATOR_ELEMENT); writer.addAttribute("name", jobName); writer.addAttribute("xmlns", OOZIE_COORD_NS); writer.addAttribute("frequency", schedule.getFrequency()); writer.addAttribute("start", formatter.print(schedule.getStartTime())); writer.addAttribute("end", "3000-01-01T00:00Z"); writer.addAttribute("timezone", "UTC"); writeCoordinatorDatasets(writer, schedule, manager); writer.startElement("action"); writer.startElement("workflow"); element(writer, "app-path", "${kiteAppRoot}/" + workflowPath(schedule)); writer.startElement("configuration"); property(writer, "kiteAppRoot", "${kiteAppRoot}"); property(writer, COORD_NOMINAL_TIME, "${coord:nominalTime()}"); for (String inputName: manager.getJobParameters().getInputNames()) { property(writer, "coord_" + toIdentifier(inputName), "${coord:dataIn('datain_" + toIdentifier(inputName) + "')}"); } for (String outputName: manager.getJobParameters().getOutputNames()) { property(writer, "coord_" + toIdentifier(outputName), "${coord:dataOut('dataout_" + toIdentifier(outputName) + "')}"); } writer.endElement(); writer.endElement(); writer.endElement(); writer.endElement(); streamWriter.flush(); }  static String workflowPath(Schedule schedule); static String coordPath(Schedule schedule); static Instant getNominalTime(Configuration conf); static void element(XMLWriter writer, String name, String attributeName,
                              String attributeValue); static void element(XMLWriter writer, String name, String text); static void writeWorkFlow(Schedule schedule,
                                   AppContext context,
                                   OutputStream output); static void writeCoordinator(Schedule schedule,
                                      SchedulableJobManager manager,
                                      OutputStream output); static void property(XMLWriter writer, String name, String value); static void writeBundle(Class appClass,
                                 AppContext context,
                                 Path appPath,
                                 List<Schedule> schedules,
                                 OutputStream output); static Map<String,View> loadViews(SchedulableJobManager manager, Configuration conf); static Map<String,String> getJobSettings(Schedule schedule, Configuration conf); static void writeJobConfiguration(XMLWriter writer, Schedule schedule, Configuration conf);  }
@Test public void testWriteCoordinator() throws Exception { ByteArrayOutputStream output = new ByteArrayOutputStream(); SchedulableJobManager manager = JobManagers.createSchedulable(testSchedule, new AppContext(new Configuration())); OozieScheduling.writeCoordinator(testSchedule, manager, output); Document coord = XMLUtil.toDom(output); XPath xpath = XMLUtil.getXPath(); assertEquals(testSchedule.getFrequency(), xpath.evaluate("coord:coordinator-app/@frequency", coord)); assertEquals(xpath.evaluate("coord:coordinator-app/coord:datasets/" + "coord:dataset[@name = \"ds_source_users\"]/coord:uri-template", coord), ScheduledInputOutputApp.INPUT_URI_PATTERN); assertEquals(xpath.evaluate("coord:coordinator-app/coord:datasets/" + "coord:dataset[@name = \"ds_target_users\"]/coord:uri-template", coord), ScheduledInputOutputApp.OUTPUT_URI_PATTERN); assertEquals("${coord:current(0)}", xpath.evaluate("coord:coordinator-app/coord:input-events/coord:data-in/coord:instance", coord)); assertEquals("${coord:current(0)}", xpath.evaluate("coord:coordinator-app/coord:output-events/coord:data-out/coord:instance", coord)); assertEquals("${kiteAppRoot}/oozie/workflows/scheduled-input-output", xpath.evaluate("coord:coordinator-app/coord:action/coord:workflow/coord:app-path", coord)); assertEquals("${coord:nominalTime()}", xpath.evaluate("coord:coordinator-app/coord:action/coord:workflow/coord:configuration/" + "coord:property/coord:value[../coord:name/text() = \"coordNominalTime\"]", coord)); assertEquals("${coord:dataIn('datain_source_users')}", xpath.evaluate("coord:coordinator-app/coord:action/coord:workflow/coord:configuration/" + "coord:property/coord:value[../coord:name/text() = \"coord_source_users\"]", coord)); assertEquals("${coord:dataOut('dataout_target_users')}", xpath.evaluate("coord:coordinator-app/coord:action/coord:workflow/coord:configuration/" + "coord:property/coord:value[../coord:name/text() = \"coord_target_users\"]", coord)); }