You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@tez.apache.org by hi...@apache.org on 2014/04/17 22:40:26 UTC
git commit: TEZ-737. DAG name should be unique within a Tez Session.
(Mohammad Kamrul Islam via hitesh)
Repository: incubator-tez
Updated Branches:
refs/heads/master 6b1871970 -> d775daaee
TEZ-737. DAG name should be unique within a Tez Session. (Mohammad Kamrul Islam via hitesh)
Project: http://git-wip-us.apache.org/repos/asf/incubator-tez/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-tez/commit/d775daae
Tree: http://git-wip-us.apache.org/repos/asf/incubator-tez/tree/d775daae
Diff: http://git-wip-us.apache.org/repos/asf/incubator-tez/diff/d775daae
Branch: refs/heads/master
Commit: d775daaee0743730df70b2a5e9bb6e30d03c4171
Parents: 6b18719
Author: Hitesh Shah <hi...@apache.org>
Authored: Thu Apr 17 13:37:00 2014 -0700
Committer: Hitesh Shah <hi...@apache.org>
Committed: Thu Apr 17 13:37:00 2014 -0700
----------------------------------------------------------------------
.../java/org/apache/tez/client/TezClient.java | 3 +-
.../java/org/apache/tez/client/TezSession.java | 3 +-
.../apache/tez/dag/api/DuplicateDAGName.java | 41 ++++++++++++
.../org/apache/tez/dag/app/DAGAppMaster.java | 11 ++++
.../apache/tez/mapreduce/TestMRRJobsDAGApi.java | 9 ++-
.../java/org/apache/tez/test/TestTezJobs.java | 65 +++++++++++++++++++-
6 files changed, 124 insertions(+), 8 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-tez/blob/d775daae/tez-api/src/main/java/org/apache/tez/client/TezClient.java
----------------------------------------------------------------------
diff --git a/tez-api/src/main/java/org/apache/tez/client/TezClient.java b/tez-api/src/main/java/org/apache/tez/client/TezClient.java
index f4c6011..dff6d4b 100644
--- a/tez-api/src/main/java/org/apache/tez/client/TezClient.java
+++ b/tez-api/src/main/java/org/apache/tez/client/TezClient.java
@@ -107,7 +107,8 @@ public class TezClient {
ApplicationSubmissionContext appContext = TezClientUtils.createApplicationSubmissionContext(
conf, appId, dag, dag.getName(), amConfig, tezJarResources, credentials);
LOG.info("Submitting DAG to YARN"
- + ", applicationId=" + appId);
+ + ", applicationId=" + appId
+ + ", dagName=" + dag.getName());
yarnClient.submitApplication(appContext);
} catch (YarnException e) {
throw new TezException(e);
http://git-wip-us.apache.org/repos/asf/incubator-tez/blob/d775daae/tez-api/src/main/java/org/apache/tez/client/TezSession.java
----------------------------------------------------------------------
diff --git a/tez-api/src/main/java/org/apache/tez/client/TezSession.java b/tez-api/src/main/java/org/apache/tez/client/TezSession.java
index 222e77c..5950c2e 100644
--- a/tez-api/src/main/java/org/apache/tez/client/TezSession.java
+++ b/tez-api/src/main/java/org/apache/tez/client/TezSession.java
@@ -190,7 +190,8 @@ public class TezSession {
String dagId;
LOG.info("Submitting dag to TezSession"
+ ", sessionName=" + sessionName
- + ", applicationId=" + applicationId);
+ + ", applicationId=" + applicationId
+ + ", dagName=" + dag.getName());
if (additionalAmResources != null && !additionalAmResources.isEmpty()) {
for (LocalResource lr : additionalAmResources.values()) {
http://git-wip-us.apache.org/repos/asf/incubator-tez/blob/d775daae/tez-api/src/main/java/org/apache/tez/dag/api/DuplicateDAGName.java
----------------------------------------------------------------------
diff --git a/tez-api/src/main/java/org/apache/tez/dag/api/DuplicateDAGName.java b/tez-api/src/main/java/org/apache/tez/dag/api/DuplicateDAGName.java
new file mode 100644
index 0000000..52a835d
--- /dev/null
+++ b/tez-api/src/main/java/org/apache/tez/dag/api/DuplicateDAGName.java
@@ -0,0 +1,41 @@
+/**
+ * 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.tez.dag.api;
+
+/**
+ * This exception is thrown when there are duplicate DAG names in a session.
+ *
+ */
+public class DuplicateDAGName extends TezException {
+
+ private static final long serialVersionUID = -6405001382308895028L;
+
+ public DuplicateDAGName(Throwable cause) {
+ super(cause);
+ }
+
+ public DuplicateDAGName(String message) {
+ super(message);
+ }
+
+ public DuplicateDAGName(String message, Throwable cause) {
+ super(message, cause);
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/incubator-tez/blob/d775daae/tez-dag/src/main/java/org/apache/tez/dag/app/DAGAppMaster.java
----------------------------------------------------------------------
diff --git a/tez-dag/src/main/java/org/apache/tez/dag/app/DAGAppMaster.java b/tez-dag/src/main/java/org/apache/tez/dag/app/DAGAppMaster.java
index 94857b4..68b019e 100644
--- a/tez-dag/src/main/java/org/apache/tez/dag/app/DAGAppMaster.java
+++ b/tez-dag/src/main/java/org/apache/tez/dag/app/DAGAppMaster.java
@@ -34,6 +34,7 @@ import java.util.Arrays;
import java.util.Collections;
import java.util.EnumSet;
import java.util.HashMap;
+import java.util.HashSet;
import java.util.Iterator;
import java.util.LinkedHashMap;
import java.util.List;
@@ -93,6 +94,7 @@ import org.apache.tez.common.impl.LogUtils;
import org.apache.tez.common.security.JobTokenIdentifier;
import org.apache.tez.common.security.TokenCache;
import org.apache.tez.dag.api.DagTypeConverters;
+import org.apache.tez.dag.api.DuplicateDAGName;
import org.apache.tez.dag.api.TezConfiguration;
import org.apache.tez.dag.api.TezConstants;
import org.apache.tez.dag.api.TezException;
@@ -236,6 +238,10 @@ public class DAGAppMaster extends AbstractService {
private Path recoveryDataDir;
private Path currentRecoveryDataDir;
private FileSystem recoveryFS;
+ /**
+ * set of already executed dag names.
+ */
+ Set<String> dagNames = new HashSet<String>();
protected boolean isLastAMRetry = false;
@@ -1856,6 +1862,11 @@ public class DAGAppMaster extends AbstractService {
long submitTime = this.clock.getTime();
this.state = DAGAppMasterState.RUNNING;
this.appName = dagPlan.getName();
+ if (dagNames.contains(dagPlan.getName())) {
+ throw new DuplicateDAGName("Duplicate dag name '" + dagPlan.getName() + "'");
+ }
+ dagNames.add(dagPlan.getName());
+
// /////////////////// Create the job itself.
DAG newDAG = createDAG(dagPlan);
_updateLoggers(newDAG, "");
http://git-wip-us.apache.org/repos/asf/incubator-tez/blob/d775daae/tez-tests/src/test/java/org/apache/tez/mapreduce/TestMRRJobsDAGApi.java
----------------------------------------------------------------------
diff --git a/tez-tests/src/test/java/org/apache/tez/mapreduce/TestMRRJobsDAGApi.java b/tez-tests/src/test/java/org/apache/tez/mapreduce/TestMRRJobsDAGApi.java
index 394df72..1fc362c 100644
--- a/tez-tests/src/test/java/org/apache/tez/mapreduce/TestMRRJobsDAGApi.java
+++ b/tez-tests/src/test/java/org/apache/tez/mapreduce/TestMRRJobsDAGApi.java
@@ -25,8 +25,6 @@ import java.io.OutputStream;
import java.io.OutputStreamWriter;
import java.net.URI;
import java.net.URISyntaxException;
-import java.rmi.RemoteException;
-import java.security.CodeSource;
import java.util.Collections;
import java.util.HashMap;
import java.util.List;
@@ -124,6 +122,7 @@ public class TestMRRJobsDAGApi {
private static Configuration conf = new Configuration();
private static FileSystem remoteFs;
+ private Random random = new Random();
private static String TEST_ROOT_DIR = "target" + Path.SEPARATOR
+ TestMRRJobsDAGApi.class.getName() + "-tmpDir";
@@ -491,7 +490,7 @@ public class TestMRRJobsDAGApi {
byte[] stage1InputPayload = MRHelpers.createMRInputPayload(stage1Payload, null);
byte[] stage3Payload = MRHelpers.createUserPayloadFromConf(stage3Conf);
- DAG dag = new DAG("testMRRSleepJobDagSubmit");
+ DAG dag = new DAG("testMRRSleepJobDagSubmit-" + random.nextInt(1000));
int stage1NumTasks = genSplitsInAM ? -1 : inputSplitInfo.getNumTasks();
Class<? extends TezRootInputInitializer> inputInitializerClazz =
genSplitsInAM ? (initializerClass == null ? MRInputAMSplitGenerator.class : initializerClass)
@@ -631,8 +630,8 @@ public class TestMRRJobsDAGApi {
}
if(dagViaRPC) {
- LOG.info("Submitting dag to tez session with appId="
- + tezSession.getApplicationId());
+ LOG.info("Submitting dag to tez session with appId=" + tezSession.getApplicationId()
+ + " and Dag Name=" + dag.getName());
dagClient = tezSession.submitDAG(dag, additionalLocalResources);
Assert.assertEquals(TezSessionStatus.RUNNING,
tezSession.getSessionStatus());
http://git-wip-us.apache.org/repos/asf/incubator-tez/blob/d775daae/tez-tests/src/test/java/org/apache/tez/test/TestTezJobs.java
----------------------------------------------------------------------
diff --git a/tez-tests/src/test/java/org/apache/tez/test/TestTezJobs.java b/tez-tests/src/test/java/org/apache/tez/test/TestTezJobs.java
index dd2fd75..5b72463 100644
--- a/tez-tests/src/test/java/org/apache/tez/test/TestTezJobs.java
+++ b/tez-tests/src/test/java/org/apache/tez/test/TestTezJobs.java
@@ -20,6 +20,8 @@ package org.apache.tez.test;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertNotNull;
+import static org.junit.Assert.assertTrue;
+import static org.junit.Assert.fail;
import java.io.IOException;
import java.util.HashMap;
@@ -31,13 +33,17 @@ import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.hdfs.MiniDFSCluster;
+import org.apache.hadoop.yarn.api.records.ApplicationId;
import org.apache.hadoop.yarn.api.records.LocalResource;
import org.apache.hadoop.yarn.api.records.Resource;
import org.apache.tez.client.AMConfiguration;
import org.apache.tez.client.TezClient;
+import org.apache.tez.client.TezSession;
+import org.apache.tez.client.TezSessionConfiguration;
import org.apache.tez.common.counters.FileSystemCounter;
import org.apache.tez.common.counters.TaskCounter;
import org.apache.tez.dag.api.DAG;
+import org.apache.tez.dag.api.DuplicateDAGName;
import org.apache.tez.dag.api.ProcessorDescriptor;
import org.apache.tez.dag.api.TezConfiguration;
import org.apache.tez.dag.api.TezException;
@@ -67,6 +73,7 @@ public class TestTezJobs {
private static Configuration conf = new Configuration();
private static FileSystem remoteFs;
+ private Random random = new Random();
private static String TEST_ROOT_DIR = "target" + Path.SEPARATOR + TestTezJobs.class.getName()
+ "-tmpDir";
@@ -116,7 +123,7 @@ public class TestTezJobs {
dag.addVertex(vertex);
TezConfiguration tezConf = new TezConfiguration(mrrTezCluster.getConfig());
- Path remoteStagingDir = remoteFs.makeQualified(new Path("/tmp", String.valueOf(new Random()
+ Path remoteStagingDir = remoteFs.makeQualified(new Path("/tmp", String.valueOf(random
.nextInt(100000))));
remoteFs.mkdirs(remoteStagingDir);
tezConf.set(TezConfiguration.TEZ_AM_STAGING_DIR, remoteStagingDir.toString());
@@ -142,4 +149,60 @@ public class TestTezJobs {
assertNotNull(dagStatus.getDAGCounters().findCounter(TaskCounter.GC_TIME_MILLIS));
ExampleDriver.printDAGStatus(dagClient, new String[] { "SleepVertex" }, true, true);
}
+
+ @Test(timeout = 100000)
+ public void testMultipleDAGsWithDuplicateName() throws TezException, IOException,
+ InterruptedException {
+ TezSession tezSession = null;
+ try {
+ TezConfiguration tezConf = new TezConfiguration(mrrTezCluster.getConfig());
+ Path remoteStagingDir = remoteFs.makeQualified(new Path("/tmp", String.valueOf(random
+ .nextInt(100000))));
+ remoteFs.mkdirs(remoteStagingDir);
+ tezConf.set(TezConfiguration.TEZ_AM_STAGING_DIR, remoteStagingDir.toString());
+ AMConfiguration amConf = new AMConfiguration(new HashMap<String, String>(),
+ new HashMap<String, LocalResource>(), tezConf, null);
+ TezSessionConfiguration sessionConfig = new TezSessionConfiguration(amConf, tezConf);
+
+ TezClient tezClient = new TezClient(tezConf);
+ ApplicationId appId = tezClient.createApplication();
+ tezSession = new TezSession("OrderedWordCountSession", appId, sessionConfig);
+ tezSession.start();
+
+ SleepProcessorConfig spConf = new SleepProcessorConfig(1);
+ for (int dagIndex = 1; dagIndex <= 2; dagIndex++) {
+ DAG dag = new DAG("TezSleepProcessor");
+ Vertex vertex = new Vertex("SleepVertex", new ProcessorDescriptor(
+ SleepProcessor.class.getName()).setUserPayload(spConf.toUserPayload()), 1,
+ Resource.newInstance(1024, 1));
+ dag.addVertex(vertex);
+
+ DAGClient dagClient = null;
+ try {
+ dagClient = tezSession.submitDAG(dag);
+ if (dagIndex > 1) {
+ fail("Should fail due to duplicate dag name for dagIndex: " + dagIndex);
+ }
+ } catch (TezException tex) {
+ if (dagIndex > 1) {
+ assertTrue(tex.getMessage().contains("Duplicate dag name "));
+ continue;
+ }
+ fail("DuplicateDAGName exception thrown for 1st DAG submission");
+ }
+ DAGStatus dagStatus = dagClient.getDAGStatus(null);
+ while (!dagStatus.isCompleted()) {
+ LOG.debug("Waiting for job to complete. Sleeping for 500ms." + " Current state: "
+ + dagStatus.getState());
+ Thread.sleep(500l);
+ dagStatus = dagClient.getDAGStatus(null);
+ }
+ }
+ } finally {
+ if (tezSession != null) {
+ tezSession.stop();
+ }
+ }
+ }
+
}