OOZIE-1941 Bundle coordinator name can't be parameterized
Project: http://git-wip-us.apache.org/repos/asf/oozie/repo Commit: http://git-wip-us.apache.org/repos/asf/oozie/commit/1507cc2b Tree: http://git-wip-us.apache.org/repos/asf/oozie/tree/1507cc2b Diff: http://git-wip-us.apache.org/repos/asf/oozie/diff/1507cc2b Branch: refs/remotes/trunk Commit: 1507cc2b97add380009aebb6c69207b708a996d2 Parents: ec41c73 Author: Purshotam Shah <[email protected]> Authored: Thu Sep 4 10:50:55 2014 -0700 Committer: Purshotam Shah <[email protected]> Committed: Thu Sep 4 10:50:55 2014 -0700 ---------------------------------------------------------------------- client/src/main/resources/oozie-bundle-0.2.xsd | 2 +- .../main/java/org/apache/oozie/ErrorCode.java | 2 + .../command/bundle/BundleStartXCommand.java | 9 +++ .../command/bundle/BundleSubmitXCommand.java | 11 ++- .../apache/oozie/service/RecoveryService.java | 15 +++- .../bundle/TestBundleSubmitXCommand.java | 84 +++++++++++++++++++- .../org/apache/oozie/test/XDataTestCase.java | 6 +- core/src/test/resources/bundle-submit-job.xml | 6 +- release-log.txt | 1 + 9 files changed, 123 insertions(+), 13 deletions(-) ---------------------------------------------------------------------- http://git-wip-us.apache.org/repos/asf/oozie/blob/1507cc2b/client/src/main/resources/oozie-bundle-0.2.xsd ---------------------------------------------------------------------- diff --git a/client/src/main/resources/oozie-bundle-0.2.xsd b/client/src/main/resources/oozie-bundle-0.2.xsd index 7da8197..b4e37e5 100644 --- a/client/src/main/resources/oozie-bundle-0.2.xsd +++ b/client/src/main/resources/oozie-bundle-0.2.xsd @@ -56,7 +56,7 @@ <xs:element name="app-path" type="xs:string" minOccurs="1" maxOccurs="1"/> <xs:element name="configuration" type="bundle:CONFIGURATION" minOccurs="0" maxOccurs="1"/> </xs:sequence> - <xs:attribute name="name" type="bundle:IDENTIFIER" use="required"/> + <xs:attribute name="name" type="xs:string" use="required"/> <xs:attribute name="critical" type="xs:string" use="optional"/> </xs:complexType> <xs:complexType name="CONFIGURATION"> http://git-wip-us.apache.org/repos/asf/oozie/blob/1507cc2b/core/src/main/java/org/apache/oozie/ErrorCode.java ---------------------------------------------------------------------- diff --git a/core/src/main/java/org/apache/oozie/ErrorCode.java b/core/src/main/java/org/apache/oozie/ErrorCode.java index 6de6db4..f9f88f4 100644 --- a/core/src/main/java/org/apache/oozie/ErrorCode.java +++ b/core/src/main/java/org/apache/oozie/ErrorCode.java @@ -234,6 +234,8 @@ public enum ErrorCode { E1318(XLog.STD, "No coord jobs for the bundle=[{0}], fail the bundle"), E1319(XLog.STD, "Invalid bundle coord job namespace, [{0}]"), E1320(XLog.STD, "Bundle Job change error, [{0}]"), + E1321(XLog.STD, "Error evaluating coord name, [{0}]"), + E1400(XLog.STD, "doAs (proxyuser) failure"), http://git-wip-us.apache.org/repos/asf/oozie/blob/1507cc2b/core/src/main/java/org/apache/oozie/command/bundle/BundleStartXCommand.java ---------------------------------------------------------------------- diff --git a/core/src/main/java/org/apache/oozie/command/bundle/BundleStartXCommand.java b/core/src/main/java/org/apache/oozie/command/bundle/BundleStartXCommand.java index 3523186..e026efb 100644 --- a/core/src/main/java/org/apache/oozie/command/bundle/BundleStartXCommand.java +++ b/core/src/main/java/org/apache/oozie/command/bundle/BundleStartXCommand.java @@ -44,6 +44,7 @@ import org.apache.oozie.executor.jpa.BundleJobQueryExecutor; import org.apache.oozie.executor.jpa.BundleJobQueryExecutor.BundleJobQuery; import org.apache.oozie.executor.jpa.JPAExecutorException; import org.apache.oozie.executor.jpa.BatchQueryExecutor.UpdateEntry; +import org.apache.oozie.util.ELUtils; import org.apache.oozie.util.JobUtils; import org.apache.oozie.util.LogUtils; import org.apache.oozie.util.ParamChecker; @@ -242,12 +243,20 @@ public class BundleStartXCommand extends StartTransitionXCommand { List<Element> coordElems = bAppXml.getChildren("coordinator", bAppXml.getNamespace()); for (Element coordElem : coordElems) { Attribute name = coordElem.getAttribute("name"); + Configuration coordConf = mergeConfig(coordElem); coordConf.set(OozieClient.BUNDLE_ID, jobId); if (OozieJobInfo.isJobInfoEnabled()) { coordConf.set(OozieJobInfo.BUNDLE_NAME, bundleJob.getAppName()); } + String coordName=name.getValue(); + try { + coordName = ELUtils.resolveAppName(coordName, coordConf); + } + catch (Exception e) { + throw new CommandException(ErrorCode.E1321, e.getMessage(), e); + } queue(new CoordSubmitXCommand(coordConf, bundleJob.getId(), name.getValue())); } http://git-wip-us.apache.org/repos/asf/oozie/blob/1507cc2b/core/src/main/java/org/apache/oozie/command/bundle/BundleSubmitXCommand.java ---------------------------------------------------------------------- diff --git a/core/src/main/java/org/apache/oozie/command/bundle/BundleSubmitXCommand.java b/core/src/main/java/org/apache/oozie/command/bundle/BundleSubmitXCommand.java index d479086..de78ab7 100644 --- a/core/src/main/java/org/apache/oozie/command/bundle/BundleSubmitXCommand.java +++ b/core/src/main/java/org/apache/oozie/command/bundle/BundleSubmitXCommand.java @@ -472,10 +472,17 @@ public class BundleSubmitXCommand extends SubmitTransitionXCommand { for (Element elem : coordElems) { Attribute name = elem.getAttribute("name"); if (name != null) { - if (set.contains(name.getValue())) { + String coordName = name.getValue(); + try { + coordName = ELUtils.resolveAppName(name.getValue(), conf); + } + catch (Exception e) { + throw new CommandException(ErrorCode.E1321, e.getMessage(), e); + } + if (set.contains(coordName)) { throw new CommandException(ErrorCode.E1304, name); } - set.add(name.getValue()); + set.add(coordName); } else { throw new CommandException(ErrorCode.E1305); http://git-wip-us.apache.org/repos/asf/oozie/blob/1507cc2b/core/src/main/java/org/apache/oozie/service/RecoveryService.java ---------------------------------------------------------------------- diff --git a/core/src/main/java/org/apache/oozie/service/RecoveryService.java b/core/src/main/java/org/apache/oozie/service/RecoveryService.java index 65510a5..c47024d 100644 --- a/core/src/main/java/org/apache/oozie/service/RecoveryService.java +++ b/core/src/main/java/org/apache/oozie/service/RecoveryService.java @@ -59,6 +59,7 @@ import org.apache.oozie.executor.jpa.CoordJobQueryExecutor.CoordJobQuery; import org.apache.oozie.executor.jpa.JPAExecutorException; import org.apache.oozie.executor.jpa.WorkflowActionQueryExecutor; import org.apache.oozie.executor.jpa.WorkflowActionQueryExecutor.WorkflowActionQuery; +import org.apache.oozie.util.ELUtils; import org.apache.oozie.util.JobUtils; import org.apache.oozie.util.XCallable; import org.apache.oozie.util.XConfiguration; @@ -195,11 +196,19 @@ public class RecoveryService implements Service { List<Element> coordElems = bAppXml.getChildren("coordinator", bAppXml.getNamespace()); for (Element coordElem : coordElems) { Attribute name = coordElem.getAttribute("name"); - if (name.getValue().equals(baction.getCoordName())) { - Configuration coordConf = mergeConfig(coordElem, bundleJob); + String coordName=name.getValue(); + Configuration coordConf = mergeConfig(coordElem, bundleJob); + try { + coordName = ELUtils.resolveAppName(coordName, coordConf); + } + catch (Exception e) { + log.error("Error evaluating coord name " + e.getMessage(), e); + continue; + } + if (coordName.equals(baction.getCoordName())) { coordConf.set(OozieClient.BUNDLE_ID, baction.getBundleId()); queueCallable(new CoordSubmitXCommand(coordConf, - bundleJob.getId(), name.getValue())); + bundleJob.getId(), coordName)); } } } http://git-wip-us.apache.org/repos/asf/oozie/blob/1507cc2b/core/src/test/java/org/apache/oozie/command/bundle/TestBundleSubmitXCommand.java ---------------------------------------------------------------------- diff --git a/core/src/test/java/org/apache/oozie/command/bundle/TestBundleSubmitXCommand.java b/core/src/test/java/org/apache/oozie/command/bundle/TestBundleSubmitXCommand.java index 26c11d3..f793601 100644 --- a/core/src/test/java/org/apache/oozie/command/bundle/TestBundleSubmitXCommand.java +++ b/core/src/test/java/org/apache/oozie/command/bundle/TestBundleSubmitXCommand.java @@ -20,22 +20,28 @@ package org.apache.oozie.command.bundle; import java.io.IOException; import java.io.StringReader; +import java.io.UnsupportedEncodingException; import java.util.Date; +import java.util.List; +import java.util.regex.Matcher; import org.apache.hadoop.conf.Configuration; import org.apache.hadoop.fs.Path; +import org.apache.oozie.BundleActionBean; import org.apache.oozie.BundleJobBean; import org.apache.oozie.ErrorCode; import org.apache.oozie.client.Job; import org.apache.oozie.client.OozieClient; import org.apache.oozie.command.CommandException; +import org.apache.oozie.executor.jpa.BundleActionQueryExecutor; +import org.apache.oozie.executor.jpa.BundleActionQueryExecutor.BundleActionQuery; import org.apache.oozie.service.JPAService; import org.apache.oozie.service.Services; import org.apache.oozie.test.XDataTestCase; import org.apache.oozie.util.XConfiguration; public class TestBundleSubmitXCommand extends XDataTestCase { - + private Services services; @Override @@ -76,7 +82,7 @@ public class TestBundleSubmitXCommand extends XDataTestCase { jobConf.set(OozieClient.BUNDLE_APP_PATH, appPath.toString()); BundleSubmitXCommand command = new BundleSubmitXCommand(true, jobConf); - BundleJobBean bundleBean = (BundleJobBean)command.getJob(); + BundleJobBean bundleBean = (BundleJobBean) command.getJob(); bundleBean.setStartTime(new Date()); bundleBean.setEndTime(new Date()); command.call(); @@ -86,7 +92,79 @@ public class TestBundleSubmitXCommand extends XDataTestCase { // bundle-submit-job.xml contains the Apache license but this result should not contain the comment block assertTrue("submit result should not contain <!-- ", !result.contains("<!--")); assertTrue("submit result should not contain --> ", !result.contains("-->")); - + } + public void testCoordJobNameParameterization() throws Exception { + final XConfiguration jobConf = setUpBundle(); + jobConf.set("coordName1", "coord1"); + jobConf.set("coordName2", "coord2"); + + BundleSubmitXCommand command = new BundleSubmitXCommand(jobConf); + final BundleJobBean bundleBean = (BundleJobBean) command.getJob(); + bundleBean.setStartTime(new Date()); + bundleBean.setEndTime(new Date()); + final String jobId = command.call(); + sleep(2000); + new BundleStartXCommand(jobId).call(); + waitFor(200000, new Predicate() { + public boolean evaluate() throws Exception { + List<BundleActionBean> actions = BundleActionQueryExecutor.getInstance().getList( + BundleActionQuery.GET_BUNDLE_ACTIONS_STATUS_UNIGNORED_FOR_BUNDLE, jobId); + return actions.get(0).getStatus().equals(Job.Status.RUNNING); + } + }); + + final List<BundleActionBean> actions = BundleActionQueryExecutor.getInstance().getList( + BundleActionQuery.GET_BUNDLE_ACTIONS_STATUS_UNIGNORED_FOR_BUNDLE, jobId); + assertEquals(actions.get(0).getCoordName(), "coord1"); + assertEquals(actions.get(1).getCoordName(), "coord2"); + } + + public void testDuplicateCoordName() throws Exception { + final XConfiguration jobConf = setUpBundle(); + // setting same coordname. + jobConf.set("coordName1", "coord"); + jobConf.set("coordName2", "coord"); + + BundleSubmitXCommand command = new BundleSubmitXCommand(true, jobConf); + BundleJobBean bundleBean = (BundleJobBean) command.getJob(); + bundleBean.setStartTime(new Date()); + bundleBean.setEndTime(new Date()); + try { + command.call(); + } + catch (CommandException e) { + assertTrue(e.getMessage().contains("Bundle Job submission Error")); + assertEquals(e.getErrorCode(), ErrorCode.E1310); + } + } + + private XConfiguration setUpBundle() throws UnsupportedEncodingException, IOException { + XConfiguration jobConf = new XConfiguration(); + + final Path coordPath1 = new Path(getFsTestCaseDir(), "coord1"); + final Path coordPath2 = new Path(getFsTestCaseDir(), "coord2"); + writeCoordXml(coordPath1, "coord-job-bundle.xml"); + writeCoordXml(coordPath2, "coord-job-bundle.xml"); + + Path bundleAppPath = new Path(getFsTestCaseDir(), "bundle"); + String bundleAppXml = getBundleXml("bundle-submit-job.xml"); + assertNotNull(bundleAppXml); + assertTrue(bundleAppXml.length() > 0); + + bundleAppXml = bundleAppXml.replaceAll("#app_path1", + Matcher.quoteReplacement(new Path(coordPath1.toString(), "coordinator.xml").toString())); + bundleAppXml = bundleAppXml.replaceAll("#app_path2", + Matcher.quoteReplacement(new Path(coordPath2.toString(), "coordinator.xml").toString())); + + writeToFile(bundleAppXml, bundleAppPath, "bundle.xml"); + final Path appPath = new Path(bundleAppPath, "bundle.xml"); + jobConf.set(OozieClient.BUNDLE_APP_PATH, appPath.toString()); + jobConf.set("appName", "test"); + + jobConf.set(OozieClient.USER_NAME, getTestUser()); + return jobConf; + + } } http://git-wip-us.apache.org/repos/asf/oozie/blob/1507cc2b/core/src/test/java/org/apache/oozie/test/XDataTestCase.java ---------------------------------------------------------------------- diff --git a/core/src/test/java/org/apache/oozie/test/XDataTestCase.java b/core/src/test/java/org/apache/oozie/test/XDataTestCase.java index ad85cd7..c149d16 100644 --- a/core/src/test/java/org/apache/oozie/test/XDataTestCase.java +++ b/core/src/test/java/org/apache/oozie/test/XDataTestCase.java @@ -1315,6 +1315,8 @@ public abstract class XDataTestCase extends XHCatTestCase { conf.set("jobTracker", getJobTrackerUri()); conf.set("nameNode", getNameNodeUri()); conf.set("appName", "bundle-app-name"); + conf.set("coordName1", "coord1"); + conf.set("coordName2", "coord2"); BundleJobBean bundle = new BundleJobBean(); bundle.setId(jobID); @@ -1381,6 +1383,8 @@ public abstract class XDataTestCase extends XHCatTestCase { conf.set(OozieClient.USER_NAME, getTestUser()); conf.set("jobTracker", getJobTrackerUri()); conf.set("nameNode", getNameNodeUri()); + conf.set("coordName1", "coord1"); + conf.set("coordName2", "coord2"); BundleJobBean bundle = new BundleJobBean(); bundle.setId(Services.get().get(UUIDService.class).generateId(ApplicationType.BUNDLE)); @@ -1400,7 +1404,7 @@ public abstract class XDataTestCase extends XHCatTestCase { return bundle; } - private String getBundleXml(String resourceXmlName) { + protected String getBundleXml(String resourceXmlName) { try { Reader reader = IOUtils.getResourceAsReader(resourceXmlName, -1); String appXml = IOUtils.getReaderAsString(reader, -1); http://git-wip-us.apache.org/repos/asf/oozie/blob/1507cc2b/core/src/test/resources/bundle-submit-job.xml ---------------------------------------------------------------------- diff --git a/core/src/test/resources/bundle-submit-job.xml b/core/src/test/resources/bundle-submit-job.xml index fd5d256..a9599ef 100644 --- a/core/src/test/resources/bundle-submit-job.xml +++ b/core/src/test/resources/bundle-submit-job.xml @@ -15,11 +15,11 @@ See the License for the specific language governing permissions and limitations under the License. --> -<bundle-app name='${appName}' xmlns:xsi='http://www.w3.org/2001/XMLSchema-instance' xmlns='uri:oozie:bundle:0.1'> +<bundle-app name='${appName}' xmlns:xsi='http://www.w3.org/2001/XMLSchema-instance' xmlns='uri:oozie:bundle:0.2'> <controls> <kick-off-time>2099-02-01T00:00Z</kick-off-time> </controls> - <coordinator name='coord1' critical='true'> + <coordinator name='${coordName1}' critical='true'> <app-path>#app_path1</app-path> <configuration> <property> @@ -36,7 +36,7 @@ </property> </configuration> </coordinator> - <coordinator name='coord2' critical='false'> + <coordinator name='${coordName2}' critical='false'> <app-path>#app_path2</app-path> <configuration> <property> http://git-wip-us.apache.org/repos/asf/oozie/blob/1507cc2b/release-log.txt ---------------------------------------------------------------------- diff --git a/release-log.txt b/release-log.txt index 890d2c5..aeae768 100644 --- a/release-log.txt +++ b/release-log.txt @@ -1,5 +1,6 @@ -- Oozie 4.2.0 release (trunk - unreleased) +OOZIE-1941 Bundle coordinator name can't be parameterized (puru) OOZIE-1966 Fix Headers in java code (shwethags via rkanter) OOZIE-1428 The delay time for requeue should be configurable (sree2k via rkanter) OOZIE-1830 Change hadoop-1 profile to use 1.2.1 (seoeun25 via rkanter)
