| /** |
| * Licensed to the Apache Software Foundation (ASF) under one |
| * or more contributor license agreements. See the NOTICE file |
| * distributed with this work for additional information |
| * regarding copyright ownership. The ASF licenses this file |
| * to you under the Apache License, Version 2.0 (the |
| * "License"); you may not use this file except in compliance |
| * with the License. You may obtain a copy of the License at |
| * |
| * http://www.apache.org/licenses/LICENSE-2.0 |
| * |
| * Unless required by applicable law or agreed to in writing, software |
| * distributed under the License is distributed on an "AS IS" BASIS, |
| * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. |
| * See the License for the specific language governing permissions and |
| * limitations under the License. |
| */ |
| |
| package org.apache.oozie.workflow.lite; |
| |
| import org.apache.commons.codec.binary.Base64; |
| import org.apache.commons.lang3.StringUtils; |
| import org.apache.hadoop.conf.Configuration; |
| import org.apache.hadoop.io.Writable; |
| import org.apache.oozie.ErrorCode; |
| import org.apache.oozie.action.ActionExecutor; |
| import org.apache.oozie.action.hadoop.FsActionExecutor; |
| import org.apache.oozie.action.oozie.SubWorkflowActionExecutor; |
| import org.apache.oozie.service.ActionService; |
| import org.apache.oozie.service.ConfigurationService; |
| import org.apache.oozie.service.SchemaService; |
| import org.apache.oozie.service.Services; |
| import org.apache.oozie.util.ELUtils; |
| import org.apache.oozie.util.IOUtils; |
| import org.apache.oozie.util.ParameterVerifier; |
| import org.apache.oozie.util.ParameterVerifierException; |
| import org.apache.oozie.util.WritableUtils; |
| import org.apache.oozie.util.XConfiguration; |
| import org.apache.oozie.util.XmlUtils; |
| import org.apache.oozie.workflow.WorkflowException; |
| import org.jdom.Element; |
| import org.jdom.JDOMException; |
| import org.jdom.Namespace; |
| import org.xml.sax.SAXException; |
| |
| import javax.xml.transform.stream.StreamSource; |
| import javax.xml.validation.Schema; |
| import javax.xml.validation.Validator; |
| import java.io.ByteArrayInputStream; |
| import java.io.ByteArrayOutputStream; |
| import java.io.DataInput; |
| import java.io.DataInputStream; |
| import java.io.DataOutput; |
| import java.io.DataOutputStream; |
| import java.io.IOException; |
| import java.io.Reader; |
| import java.io.StringReader; |
| import java.io.StringWriter; |
| import java.util.ArrayList; |
| import java.util.List; |
| import java.util.zip.Deflater; |
| import java.util.zip.DeflaterOutputStream; |
| import java.util.zip.Inflater; |
| import java.util.zip.InflaterInputStream; |
| |
| /** |
| * Class to parse and validate workflow xml |
| */ |
| public class LiteWorkflowAppParser { |
| |
| private static final String LAUNCHER_E = "launcher"; |
| private static final String DECISION_E = "decision"; |
| private static final String ACTION_E = "action"; |
| private static final String END_E = "end"; |
| private static final String START_E = "start"; |
| private static final String JOIN_E = "join"; |
| private static final String FORK_E = "fork"; |
| private static final Object KILL_E = "kill"; |
| |
| private static final String SLA_INFO = "info"; |
| private static final String CREDENTIALS = "credentials"; |
| private static final String GLOBAL = "global"; |
| private static final String PARAMETERS = "parameters"; |
| |
| private static final String NAME_A = "name"; |
| private static final String CRED_A = "cred"; |
| private static final String USER_RETRY_MAX_A = "retry-max"; |
| private static final String USER_RETRY_INTERVAL_A = "retry-interval"; |
| private static final String TO_A = "to"; |
| private static final String USER_RETRY_POLICY_A = "retry-policy"; |
| |
| private static final String FORK_PATH_E = "path"; |
| private static final String FORK_START_A = "start"; |
| |
| private static final String ACTION_OK_E = "ok"; |
| private static final String ACTION_ERROR_E = "error"; |
| |
| private static final String DECISION_SWITCH_E = "switch"; |
| private static final String DECISION_CASE_E = "case"; |
| private static final String DECISION_DEFAULT_E = "default"; |
| |
| private static final String SUBWORKFLOW_E = "sub-workflow"; |
| |
| private static final String KILL_MESSAGE_E = "message"; |
| public static final String VALIDATE_FORK_JOIN = "oozie.validate.ForkJoin"; |
| public static final String WF_VALIDATE_FORK_JOIN = "oozie.wf.validate.ForkJoin"; |
| |
| public static final String DEFAULT_NAME_NODE = "oozie.actions.default.name-node"; |
| public static final String DEFAULT_JOB_TRACKER = "oozie.actions.default.job-tracker"; |
| public static final String DEFAULT_RESOURCE_MANAGER = "oozie.actions.default.resource-manager"; |
| public static final String OOZIE_GLOBAL = "oozie.wf.globalconf"; |
| |
| private static final String JOB_TRACKER = "job-tracker"; |
| private static final String RESOURCE_MANAGER = "resource-manager"; |
| |
| private static final String NAME_NODE = "name-node"; |
| private static final String JOB_XML = "job-xml"; |
| private static final String CONFIGURATION = "configuration"; |
| |
| private Schema schema; |
| private Class<? extends ControlNodeHandler> controlNodeHandler; |
| private Class<? extends DecisionNodeHandler> decisionHandlerClass; |
| private Class<? extends ActionNodeHandler> actionHandlerClass; |
| |
| private String defaultNameNode; |
| private String defaultResourceManager; |
| private String defaultJobTracker; |
| private boolean isResourceManagerTagUsed; |
| |
| public LiteWorkflowAppParser(Schema schema, |
| Class<? extends ControlNodeHandler> controlNodeHandler, |
| Class<? extends DecisionNodeHandler> decisionHandlerClass, |
| Class<? extends ActionNodeHandler> actionHandlerClass) throws WorkflowException { |
| this.schema = schema; |
| this.controlNodeHandler = controlNodeHandler; |
| this.decisionHandlerClass = decisionHandlerClass; |
| this.actionHandlerClass = actionHandlerClass; |
| |
| defaultNameNode = getPropertyFromConfig(DEFAULT_NAME_NODE); |
| defaultResourceManager = getPropertyFromConfig(DEFAULT_RESOURCE_MANAGER); |
| defaultJobTracker = getPropertyFromConfig(DEFAULT_JOB_TRACKER); |
| } |
| |
| private String getPropertyFromConfig(final String configPropertyKey) { |
| String property = ConfigurationService.get(configPropertyKey); |
| if (property != null) { |
| property = property.trim(); |
| if (property.isEmpty()) { |
| property = null; |
| } |
| } |
| |
| return property; |
| } |
| |
| public LiteWorkflowApp validateAndParse(Reader reader, Configuration jobConf) throws WorkflowException { |
| return validateAndParse(reader, jobConf, null); |
| } |
| |
| /** |
| * Parse and validate xml to {@link LiteWorkflowApp} |
| * |
| * @param reader configuration reader |
| * @param jobConf job configuration |
| * @param configDefault default configuration |
| * @return LiteWorkflowApp |
| * @throws WorkflowException if workflow related issue occurs |
| */ |
| public LiteWorkflowApp validateAndParse(Reader reader, Configuration jobConf, Configuration configDefault) |
| throws WorkflowException { |
| try { |
| StringWriter writer = new StringWriter(); |
| IOUtils.copyCharStream(reader, writer); |
| String strDef = writer.toString(); |
| |
| if (schema != null) { |
| Validator validator = SchemaService.getValidator(schema); |
| validator.validate(new StreamSource(new StringReader(strDef))); |
| } |
| |
| Element wfDefElement = XmlUtils.parseXml(strDef); |
| ParameterVerifier.verifyParameters(jobConf, wfDefElement); |
| LiteWorkflowApp app = parse(strDef, wfDefElement, configDefault, jobConf); |
| |
| |
| boolean validateForkJoin = false; |
| |
| if (jobConf.getBoolean(WF_VALIDATE_FORK_JOIN, true) |
| && ConfigurationService.getBoolean(VALIDATE_FORK_JOIN)) { |
| validateForkJoin = true; |
| } |
| |
| LiteWorkflowValidator validator = new LiteWorkflowValidator(); |
| validator.validateWorkflow(app, validateForkJoin); |
| |
| return app; |
| } |
| catch (ParameterVerifierException ex) { |
| throw new WorkflowException(ex); |
| } |
| catch (JDOMException ex) { |
| throw new WorkflowException(ErrorCode.E0700, ex.getMessage(), ex); |
| } |
| catch (SAXException ex) { |
| throw new WorkflowException(ErrorCode.E0701, ex.getMessage(), ex); |
| } |
| catch (IOException ex) { |
| throw new WorkflowException(ErrorCode.E0702, ex.getMessage(), ex); |
| } |
| } |
| |
| /** |
| * Parse xml to {@link LiteWorkflowApp} |
| * |
| * @param strDef definition |
| * @param root root xml element |
| * @param configDefault default configuration |
| * @param jobConf job configuration |
| * @return LiteWorkflowApp |
| * @throws WorkflowException if workflow related issue occurs |
| */ |
| @SuppressWarnings({"unchecked"}) |
| private LiteWorkflowApp parse(String strDef, Element root, Configuration configDefault, Configuration jobConf) |
| throws WorkflowException { |
| Namespace ns = root.getNamespace(); |
| |
| LiteWorkflowApp def = null; |
| GlobalSectionData gData = jobConf.get(OOZIE_GLOBAL) == null ? |
| null : getGlobalFromString(jobConf.get(OOZIE_GLOBAL)); |
| boolean serializedGlobalConf = false; |
| for (Element eNode : (List<Element>) root.getChildren()) { |
| if (eNode.getName().equals(START_E)) { |
| def = new LiteWorkflowApp(root.getAttributeValue(NAME_A), strDef, |
| new StartNodeDef(controlNodeHandler, eNode.getAttributeValue(TO_A))); |
| } else if (eNode.getName().equals(END_E)) { |
| def.addNode(new EndNodeDef(eNode.getAttributeValue(NAME_A), controlNodeHandler)); |
| } else if (eNode.getName().equals(KILL_E)) { |
| def.addNode(new KillNodeDef(eNode.getAttributeValue(NAME_A), |
| eNode.getChildText(KILL_MESSAGE_E, ns), controlNodeHandler)); |
| } else if (eNode.getName().equals(FORK_E)) { |
| List<String> paths = new ArrayList<String>(); |
| for (Element tran : (List<Element>) eNode.getChildren(FORK_PATH_E, ns)) { |
| paths.add(tran.getAttributeValue(FORK_START_A)); |
| } |
| def.addNode(new ForkNodeDef(eNode.getAttributeValue(NAME_A), controlNodeHandler, paths)); |
| } else if (eNode.getName().equals(JOIN_E)) { |
| def.addNode(new JoinNodeDef(eNode.getAttributeValue(NAME_A), controlNodeHandler, eNode.getAttributeValue(TO_A))); |
| } else if (eNode.getName().equals(DECISION_E)) { |
| Element eSwitch = eNode.getChild(DECISION_SWITCH_E, ns); |
| List<String> transitions = new ArrayList<String>(); |
| for (Element e : (List<Element>) eSwitch.getChildren(DECISION_CASE_E, ns)) { |
| transitions.add(e.getAttributeValue(TO_A)); |
| } |
| transitions.add(eSwitch.getChild(DECISION_DEFAULT_E, ns).getAttributeValue(TO_A)); |
| |
| String switchStatement = XmlUtils.prettyPrint(eSwitch).toString(); |
| def.addNode(new DecisionNodeDef(eNode.getAttributeValue(NAME_A), switchStatement, decisionHandlerClass, |
| transitions)); |
| } else if (ACTION_E.equals(eNode.getName())) { |
| String[] transitions = new String[2]; |
| Element eActionConf = null; |
| for (Element elem : (List<Element>) eNode.getChildren()) { |
| if (ACTION_OK_E.equals(elem.getName())) { |
| transitions[0] = elem.getAttributeValue(TO_A); |
| } else if (ACTION_ERROR_E.equals(elem.getName())) { |
| transitions[1] = elem.getAttributeValue(TO_A); |
| } else if (SLA_INFO.equals(elem.getName()) || CREDENTIALS.equals(elem.getName())) { |
| continue; |
| } else { |
| if (!serializedGlobalConf && elem.getName().equals(SubWorkflowActionExecutor.ACTION_TYPE) && |
| elem.getChild(("propagate-configuration"), ns) != null && gData != null) { |
| serializedGlobalConf = true; |
| jobConf.set(OOZIE_GLOBAL, getGlobalString(gData)); |
| } |
| eActionConf = elem; |
| if (SUBWORKFLOW_E.equals(elem.getName())) { |
| handleDefaultsAndGlobal(gData, null, elem, ns); |
| } |
| else { |
| handleDefaultsAndGlobal(gData, configDefault, elem, ns); |
| } |
| } |
| } |
| |
| String credStr = eNode.getAttributeValue(CRED_A); |
| String userRetryMaxStr = eNode.getAttributeValue(USER_RETRY_MAX_A); |
| String userRetryIntervalStr = eNode.getAttributeValue(USER_RETRY_INTERVAL_A); |
| String userRetryPolicyStr = eNode.getAttributeValue(USER_RETRY_POLICY_A); |
| try { |
| if (!StringUtils.isEmpty(userRetryMaxStr)) { |
| userRetryMaxStr = ELUtils.resolveAppName(userRetryMaxStr, jobConf); |
| } |
| if (!StringUtils.isEmpty(userRetryIntervalStr)) { |
| userRetryIntervalStr = ELUtils.resolveAppName(userRetryIntervalStr, jobConf); |
| } |
| if (!StringUtils.isEmpty(userRetryPolicyStr)) { |
| userRetryPolicyStr = ELUtils.resolveAppName(userRetryPolicyStr, jobConf); |
| } |
| } |
| catch (Exception e) { |
| throw new WorkflowException(ErrorCode.E0703, e.getMessage()); |
| } |
| |
| String actionConf = XmlUtils.prettyPrint(eActionConf).toString(); |
| def.addNode(new ActionNodeDef(eNode.getAttributeValue(NAME_A), actionConf, actionHandlerClass, |
| transitions[0], transitions[1], credStr, userRetryMaxStr, userRetryIntervalStr, |
| userRetryPolicyStr)); |
| } else if (SLA_INFO.equals(eNode.getName()) || CREDENTIALS.equals(eNode.getName())) { |
| // No operation is required |
| } else if (eNode.getName().equals(GLOBAL)) { |
| if(jobConf.get(OOZIE_GLOBAL) != null) { |
| gData = getGlobalFromString(jobConf.get(OOZIE_GLOBAL)); |
| handleDefaultsAndGlobal(gData, null, eNode, ns); |
| } |
| |
| gData = parseGlobalSection(ns, eNode); |
| |
| } else if (eNode.getName().equals(PARAMETERS)) { |
| // No operation is required |
| } else { |
| throw new WorkflowException(ErrorCode.E0703, eNode.getName()); |
| } |
| } |
| return def; |
| } |
| |
| /** |
| * Read the GlobalSectionData from Base64 string. |
| * @param globalStr string value of global section |
| * @return GlobalSectionData |
| * @throws WorkflowException if workflow related issue occurs |
| */ |
| private GlobalSectionData getGlobalFromString(String globalStr) throws WorkflowException { |
| GlobalSectionData globalSectionData = new GlobalSectionData(); |
| try { |
| byte[] data = Base64.decodeBase64(globalStr); |
| Inflater inflater = new Inflater(); |
| DataInputStream ois = new DataInputStream(new InflaterInputStream(new ByteArrayInputStream(data), inflater)); |
| globalSectionData.readFields(ois); |
| ois.close(); |
| } catch (Exception ex) { |
| throw new WorkflowException(ErrorCode.E0700, "Error while processing global section conf"); |
| } |
| return globalSectionData; |
| } |
| |
| |
| /** |
| * Write the GlobalSectionData to a Base64 string. |
| * @param globalSectionData string value of global section |
| * @return String |
| * @throws WorkflowException if workflow related issue occurs |
| */ |
| private String getGlobalString(GlobalSectionData globalSectionData) throws WorkflowException { |
| ByteArrayOutputStream baos = new ByteArrayOutputStream(); |
| DataOutputStream oos = null; |
| try { |
| Deflater def = new Deflater(); |
| oos = new DataOutputStream(new DeflaterOutputStream(baos, def)); |
| globalSectionData.write(oos); |
| oos.close(); |
| } catch (IOException e) { |
| throw new WorkflowException(ErrorCode.E0700, "Error while processing global section conf"); |
| } |
| return Base64.encodeBase64String(baos.toByteArray()); |
| } |
| |
| private void addChildElement(Element parent, Namespace ns, String childName, String childValue) { |
| Element child = new Element(childName, ns); |
| child.setText(childValue); |
| parent.addContent(child); |
| } |
| |
| private class GlobalSectionData implements Writable { |
| String jobTracker; |
| String nameNode; |
| List<String> jobXmls; |
| Configuration conf; |
| |
| public GlobalSectionData() { |
| } |
| |
| public GlobalSectionData(String jobTracker, String nameNode, List<String> jobXmls, Configuration conf) { |
| this.jobTracker = jobTracker; |
| this.nameNode = nameNode; |
| this.jobXmls = jobXmls; |
| this.conf = conf; |
| } |
| |
| @Override |
| public void write(DataOutput dataOutput) throws IOException { |
| WritableUtils.writeStr(dataOutput, jobTracker); |
| WritableUtils.writeStr(dataOutput, nameNode); |
| |
| if(jobXmls != null && !jobXmls.isEmpty()) { |
| dataOutput.writeInt(jobXmls.size()); |
| for (String content : jobXmls) { |
| WritableUtils.writeStr(dataOutput, content); |
| } |
| } else { |
| dataOutput.writeInt(0); |
| } |
| if(conf != null) { |
| WritableUtils.writeStr(dataOutput, XmlUtils.prettyPrint(conf).toString()); |
| } else { |
| WritableUtils.writeStr(dataOutput, null); |
| } |
| } |
| |
| @Override |
| public void readFields(DataInput dataInput) throws IOException { |
| jobTracker = WritableUtils.readStr(dataInput); |
| nameNode = WritableUtils.readStr(dataInput); |
| int length = dataInput.readInt(); |
| if (length > 0) { |
| jobXmls = new ArrayList<String>(); |
| for (int i = 0; i < length; i++) { |
| jobXmls.add(WritableUtils.readStr(dataInput)); |
| } |
| } |
| String confString = WritableUtils.readStr(dataInput); |
| if(confString != null) { |
| conf = new XConfiguration(new StringReader(confString)); |
| } |
| } |
| } |
| |
| private GlobalSectionData parseGlobalSection(Namespace ns, Element global) throws WorkflowException { |
| GlobalSectionData gData = null; |
| if (global != null) { |
| String globalJobTracker = null; |
| Element globalJobTrackerElement = getResourceManager(ns, global); |
| isResourceManagerTagUsed = globalJobTrackerElement != null |
| && globalJobTrackerElement.getName().equals(RESOURCE_MANAGER); |
| if (globalJobTrackerElement != null) { |
| globalJobTracker = globalJobTrackerElement.getValue(); |
| } |
| |
| |
| String globalNameNode = null; |
| Element globalNameNodeElement = global.getChild(NAME_NODE, ns); |
| if (globalNameNodeElement != null) { |
| globalNameNode = globalNameNodeElement.getValue(); |
| } |
| |
| List<String> globalJobXmls = null; |
| @SuppressWarnings("unchecked") |
| List<Element> globalJobXmlElements = global.getChildren(JOB_XML, ns); |
| if (!globalJobXmlElements.isEmpty()) { |
| globalJobXmls = new ArrayList<String>(globalJobXmlElements.size()); |
| for(Element jobXmlElement: globalJobXmlElements) { |
| globalJobXmls.add(jobXmlElement.getText()); |
| } |
| } |
| |
| Configuration globalConf = new XConfiguration(); |
| Element globalConfigurationElement = global.getChild(CONFIGURATION, ns); |
| if (globalConfigurationElement != null) { |
| try { |
| globalConf = new XConfiguration(new StringReader(XmlUtils.prettyPrint(globalConfigurationElement).toString())); |
| } catch (IOException ioe) { |
| throw new WorkflowException(ErrorCode.E0700, "Error while processing global section conf"); |
| } |
| } |
| |
| Element globalLauncherElement = global.getChild(LAUNCHER_E, ns); |
| if (globalLauncherElement != null) { |
| LauncherConfigHandler launcherConfigHandler = new LauncherConfigHandler(globalConf, globalLauncherElement, ns); |
| launcherConfigHandler.processSettings(); |
| } |
| gData = new GlobalSectionData(globalJobTracker, globalNameNode, globalJobXmls, globalConf); |
| } |
| return gData; |
| } |
| |
| private Element getResourceManager(final Namespace ns, final Element element) { |
| final Element resourceManager = element.getChild(RESOURCE_MANAGER, ns); |
| if (resourceManager != null) { |
| return resourceManager; |
| } |
| return element.getChild(JOB_TRACKER, ns); |
| } |
| |
| private void handleDefaultsAndGlobal(GlobalSectionData gData, Configuration configDefault, Element actionElement, Namespace ns) |
| throws WorkflowException { |
| |
| ActionExecutor ae = Services.get().get(ActionService.class).getExecutor(actionElement.getName()); |
| if (ae == null && !GLOBAL.equals(actionElement.getName())) { |
| throw new WorkflowException(ErrorCode.E0723, actionElement.getName(), ActionService.class.getName()); |
| } |
| |
| Namespace actionNs = actionElement.getNamespace(); |
| |
| // If this is the global section or ActionExecutor.requiresNameNodeJobTracker() returns true, we parse the action's |
| // <name-node> and <job-tracker> fields. If those aren't defined, we take them from the <global> section. If those |
| // aren't defined, we take them from the oozie-site defaults. If those aren't defined, we throw a WorkflowException. |
| // However, for the SubWorkflow and FS Actions, as well as the <global> section, we don't throw the WorkflowException. |
| // Also, we only parse the NN (not the JT) for the FS Action. |
| if (SubWorkflowActionExecutor.ACTION_TYPE.equals(actionElement.getName()) || |
| FsActionExecutor.ACTION_TYPE.equals(actionElement.getName()) || |
| GLOBAL.equals(actionElement.getName()) || ae.requiresNameNodeJobTracker()) { |
| if (actionElement.getChild(NAME_NODE, actionNs) == null) { |
| if (gData != null && gData.nameNode != null) { |
| addChildElement(actionElement, actionNs, NAME_NODE, gData.nameNode); |
| } else if (defaultNameNode != null) { |
| addChildElement(actionElement, actionNs, NAME_NODE, defaultNameNode); |
| } else if (!(SubWorkflowActionExecutor.ACTION_TYPE.equals(actionElement.getName()) || |
| FsActionExecutor.ACTION_TYPE.equals(actionElement.getName()) || |
| GLOBAL.equals(actionElement.getName()))) { |
| throw new WorkflowException(ErrorCode.E0701, "No " + NAME_NODE + " defined"); |
| } |
| } |
| if (getResourceManager(actionNs, actionElement) == null && |
| !FsActionExecutor.ACTION_TYPE.equals(actionElement.getName())) { |
| if (gData != null && gData.jobTracker != null) { |
| addResourceManagerOrJobTracker(actionElement, actionNs, gData.jobTracker, isResourceManagerTagUsed); |
| } else if (defaultResourceManager != null) { |
| addResourceManagerOrJobTracker(actionElement, actionNs, defaultResourceManager, true); |
| } else if (defaultJobTracker != null) { |
| addResourceManagerOrJobTracker(actionElement, actionNs, defaultJobTracker, false); |
| } else if (!(SubWorkflowActionExecutor.ACTION_TYPE.equals(actionElement.getName()) || |
| GLOBAL.equals(actionElement.getName()))) { |
| throw new WorkflowException(ErrorCode.E0701, "No " + JOB_TRACKER + " or " + RESOURCE_MANAGER + " defined"); |
| } |
| } |
| } |
| |
| // If this is the global section or ActionExecutor.supportsConfigurationJobXML() returns true, we parse the action's |
| // <configuration> and <job-xml> fields. We also merge this with those from the <global> section, if given. If none are |
| // defined, empty values are placed. Exceptions are thrown if there's an error parsing, but not if they're not given. |
| if (GLOBAL.equals(actionElement.getName()) || ae.supportsConfigurationJobXML()) { |
| @SuppressWarnings("unchecked") |
| List<Element> actionJobXmls = actionElement.getChildren(JOB_XML, actionNs); |
| if (gData != null && gData.jobXmls != null) { |
| int i = 0; |
| for(String gJobXml : gData.jobXmls) { |
| boolean alreadyExists = false; |
| for (Element actionXml : actionJobXmls) { |
| if (gJobXml.equals(actionXml.getText())) { |
| alreadyExists = true; |
| break; |
| } |
| } |
| if (!alreadyExists) { |
| Element ejobXml = new Element(JOB_XML, actionNs); |
| ejobXml.setText(gJobXml); |
| actionElement.addContent(i, ejobXml); |
| i++; |
| } |
| } |
| } |
| |
| try { |
| XConfiguration actionConf = new XConfiguration(); |
| if (configDefault != null) |
| XConfiguration.copy(configDefault, actionConf); |
| if (gData != null && gData.conf != null) { |
| XConfiguration.copy(gData.conf, actionConf); |
| } |
| |
| Element launcherConfiguration = actionElement.getChild(LAUNCHER_E, actionNs); |
| if (launcherConfiguration != null) { |
| LauncherConfigHandler launcherConfigHandler = new LauncherConfigHandler( |
| actionConf, launcherConfiguration, actionNs); |
| launcherConfigHandler.processSettings(); |
| } |
| |
| Element actionConfiguration = actionElement.getChild(CONFIGURATION, actionNs); |
| if (actionConfiguration != null) { |
| //copy and override |
| XConfiguration.copy(new XConfiguration(new StringReader(XmlUtils.prettyPrint(actionConfiguration).toString())), |
| actionConf); |
| } |
| |
| int position = actionElement.indexOf(actionConfiguration); |
| actionElement.removeContent(actionConfiguration); //replace with enhanced one |
| Element eConfXml = XmlUtils.parseXml(actionConf.toXmlString(false)); |
| eConfXml.detach(); |
| eConfXml.setNamespace(actionNs); |
| if (position > 0) { |
| actionElement.addContent(position, eConfXml); |
| } |
| else { |
| actionElement.addContent(eConfXml); |
| } |
| } |
| catch (IOException e) { |
| throw new WorkflowException(ErrorCode.E0700, "Error while processing action conf"); |
| } |
| catch (JDOMException e) { |
| throw new WorkflowException(ErrorCode.E0700, "Error while processing action conf"); |
| } |
| } |
| } |
| |
| private void addResourceManagerOrJobTracker(final Element actionElement, final Namespace actionNs, final String jobTracker, |
| boolean isResourceManagerUsed) { |
| if (isResourceManagerUsed) { |
| addChildElement(actionElement, actionNs, RESOURCE_MANAGER, jobTracker); |
| } else { |
| addChildElement(actionElement, actionNs, JOB_TRACKER, jobTracker); |
| } |
| } |
| |
| private void addResourceManager(final Element actionElement, final Namespace actionNs, final String jobTracker) { |
| addChildElement(actionElement, actionNs, RESOURCE_MANAGER, jobTracker); |
| } |
| } |
