WorkflowPrinter { public Workflow execute(List<ProcessInfo> processInfos, String workflowName) { CommonNodeMaintainer nodeMaintainer = new CommonNodeMaintainer(); NodeCollection nc = nodeMaintainer.getNc(); Set<String> printedNodeNames = new HashSet<String>(); if (workflowName == null || workflowName.trim().isEmpty()) { LOGGER.error(EMPTYERROR); throw new MetadataException(EMPTYERROR); } final String prefixXml = "<workflow-app name=\"" + workflowName + "\" xmlns=\"uri:oozie:workflow:0.4\">\n"; String pid = processInfos.get(0).getProcessId().toString(); StringBuilder credentials = new StringBuilder(); credentials.append(isSecurityEnabled(pid, "security")); final String postfixXml = "\n</workflow-app>"; StringBuilder workflowXML = new StringBuilder(); StringBuilder stepXML = new StringBuilder(); LOGGER.info("Starting workflow generation for " + workflowName); Map<Integer, ProcessInfo> tempProcessInfos = new HashMap<Integer, ProcessInfo>(); for (ProcessInfo processInfo : processInfos) { tempProcessInfos.put(processInfo.getProcessId(), processInfo); } for (ProcessInfo processInfo : processInfos) { LOGGER.debug("processing " + processInfo); NodeCollection nodeCollection = uniqNodeCollectionTreeMap.get(processInfo.getProcessId()); if (nodeCollection == null) { nodeCollection = new NodeCollection(processInfo); } String[] children = processInfo.getNextProcessIds().split(","); for (String child : children) { LOGGER.debug("Analyzing child: " + child); Integer childKey = new Integer(child); NodeCollection childNodeCollection = uniqNodeCollectionTreeMap.get(childKey); if (childNodeCollection == null) { childNodeCollection = new NodeCollection(tempProcessInfos.get(childKey)); } childNodeCollection.addParent(nodeCollection, nodeMaintainer); nodeCollection.addChild(childNodeCollection, nodeMaintainer); uniqNodeCollectionTreeMap.put(childKey, childNodeCollection); } if (nodeCollection.getHaltStepNode().getToNode().getId().equals(tempProcessInfos.get(nodeCollection.getId()).getParentProcessId())) { nodeCollection.getHaltStepNode().setToNode(nc.getHaltJobNode()); } if (tempProcessInfos.get(nodeCollection.getId()).getParentProcessId() == 0) { OozieNode firstNode = null; if (nodeCollection.getForkNode() != null) { LOGGER.info("First node in the workflow is a fork"); firstNode = nodeCollection.getForkNode(); } else { firstNode = nodeCollection.getChildren().get(0).getInitStepNode(); } LOGGER.debug("firstNode=" + firstNode.getName()); nc.getRecoveryDecisionNode().setToNode(firstNode); } uniqNodeCollectionTreeMap.put(processInfo.getProcessId(), nodeCollection); } LOGGER.debug("uniqNodeCollectionTreeMap=" + uniqNodeCollectionTreeMap); Map<String, OozieNode> oozieNodeMap = new HashMap<String, OozieNode>(); oozieNodeMap.put(nc.getStart().getName(), nc.getStart()); oozieNodeMap.put(nc.getInitJobNode().getName(), nc.getInitJobNode()); LOGGER.debug("nodeMaintainer.restartNodes=" + nodeMaintainer.getRestartNodes()); nc.getRecoveryDecisionNode().setPreviousHaltNodes(nodeMaintainer.getRestartNodes()); oozieNodeMap.put(nc.getRecoveryDecisionNode().getName(), nc.getRecoveryDecisionNode()); for (Integer id : uniqNodeCollectionTreeMap.keySet()) { if (tempProcessInfos.get(id).getParentProcessId() == 0) { uniqNodeCollectionTreeMap.get(id).setActionNode(null); uniqNodeCollectionTreeMap.get(id).setTermStepNode(null); uniqNodeCollectionTreeMap.get(id).setHaltStepNode(null); uniqNodeCollectionTreeMap.get(id).setInitStepNode(null); nc.setIdForProcessNodes(id); } LOGGER.debug("Adding collection for: " + id); oozieNodeMap.putAll(uniqNodeCollectionTreeMap.get(id).getOozieNodes()); stepXML.append(uniqNodeCollectionTreeMap.get(id).toXML(printedNodeNames)); } oozieNodeMap.put(nc.getHaltJobNode().getName(), nc.getHaltJobNode()); oozieNodeMap.put(nc.getTermJobNode().getName(), nc.getTermJobNode()); oozieNodeMap.put(nc.getHalt().getName(), nc.getHalt()); oozieNodeMap.put(nc.getKill().getName(), nc.getKill()); LOGGER.debug("oozieNodeMap size=" + oozieNodeMap.size()); workflowXML.append(prefixXml); workflowXML.append(credentials.toString()); workflowXML.append(nc.getStart().getXML()); workflowXML.append(nc.getInitJobNode().getXML()); workflowXML.append(nc.getRecoveryDecisionNode().getXML()); workflowXML.append(stepXML); workflowXML.append(nc.getHaltJobNode().getXML()); workflowXML.append(nc.getTermJobNode().getXML()); workflowXML.append(nc.getKill().getXML()); workflowXML.append(nc.getHalt().getXML()); workflowXML.append(postfixXml); LOGGER.debug("Complete !"); Workflow workflow = new Workflow(); workflow.setXml(workflowXML); workflow.setDot(DotUtil.getDot(oozieNodeMap.values())); return workflow; }  Workflow execute(List<ProcessInfo> processInfos, String workflowName); String isSecurityEnabled(String pid, String configGroup); String getCredProperties(String pid, String configGroup); Workflow execInfo(List<ProcessInfo> processInfos, String workflowName);  }
@Test @Ignore public void testExecute1() throws Exception { List<ProcessInfo> processInfos = getProcessBeans1(); String workflowXML = new WorkflowPrinter().execute(processInfos, "test-workflow").getXml().toString(); LOGGER.info("workflowxml is "+workflowXML); PrintWriter out = new PrintWriter("unitTest1.workflow.xml"); out.println(workflowXML); out.close(); OozieCLI oozieCLI = new OozieCLI(); oozieCLI.run(new String[]{"validate", "unitTest1.workflow.xml"}); }
@Test @Ignore public void testExecute2() throws Exception { List<ProcessInfo> processInfos = getProcessBeans2(); String workflowXML = new WorkflowPrinter().execute(processInfos, "test-workflow").getXml().toString(); PrintWriter out = new PrintWriter("unitTest2.workflow.xml"); out.println(workflowXML); out.close(); OozieCLI oozieCLI = new OozieCLI(); oozieCLI.run(new String[]{"validate", "unitTest2.workflow.xml"}); }