You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@tez.apache.org by bi...@apache.org on 2014/08/21 22:55:39 UTC
[1/4] git commit: TEZ-1469. AM/Session LRs are not shipped to
vertices in new API use-case (bikas)
Repository: tez
Updated Branches:
refs/heads/branch-0.5.0 65eb31893 -> d04529696
TEZ-1469. AM/Session LRs are not shipped to vertices in new API use-case (bikas)
Project: http://git-wip-us.apache.org/repos/asf/tez/repo
Commit: http://git-wip-us.apache.org/repos/asf/tez/commit/0a41e3a5
Tree: http://git-wip-us.apache.org/repos/asf/tez/tree/0a41e3a5
Diff: http://git-wip-us.apache.org/repos/asf/tez/diff/0a41e3a5
Branch: refs/heads/branch-0.5.0
Commit: 0a41e3a5fefc2bfa6047b9ab900d210273dbb8f4
Parents: 65eb318
Author: Bikas Saha <bi...@apache.org>
Authored: Wed Aug 20 18:46:10 2014 -0700
Committer: Bikas Saha <bi...@apache.org>
Committed: Thu Aug 21 13:46:38 2014 -0700
----------------------------------------------------------------------
CHANGES.txt | 1 +
.../org/apache/tez/client/AMConfiguration.java | 27 +++--
.../java/org/apache/tez/client/TezClient.java | 78 ++++++--------
.../org/apache/tez/client/TezClientUtils.java | 106 +++++++------------
.../org/apache/tez/common/TezCommonUtils.java | 19 +++-
.../main/java/org/apache/tez/dag/api/DAG.java | 26 ++++-
.../tez/dag/api/DataSourceDescriptor.java | 22 ++--
.../org/apache/tez/dag/api/TezConstants.java | 9 +-
.../java/org/apache/tez/dag/api/Vertex.java | 17 +--
.../org/apache/tez/client/TestTezClient.java | 22 ++--
.../apache/tez/common/TestTezCommonUtils.java | 4 +-
.../org/apache/tez/dag/api/TestDAGPlan.java | 18 ++--
.../org/apache/tez/dag/api/TestDAGVerify.java | 81 +++++++++++++-
.../java/org/apache/tez/dag/app/AppContext.java | 3 -
.../org/apache/tez/dag/app/DAGAppMaster.java | 14 +--
.../apache/tez/dag/app/dag/impl/VertexImpl.java | 1 -
.../apache/tez/mapreduce/client/YARNRunner.java | 2 +-
.../mapreduce/hadoop/TestMRInputHelpers.java | 10 +-
.../mapreduce/examples/FilterLinesByWord.java | 4 +-
.../examples/FilterLinesByWordOneToOne.java | 4 +-
.../tez/mapreduce/examples/MRRSleepJob.java | 6 +-
.../examples/TestOrderedWordCount.java | 8 +-
.../apache/tez/mapreduce/TestMRRJobsDAGApi.java | 2 +-
23 files changed, 274 insertions(+), 210 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/tez/blob/0a41e3a5/CHANGES.txt
----------------------------------------------------------------------
diff --git a/CHANGES.txt b/CHANGES.txt
index fa6110a..d5852b0 100644
--- a/CHANGES.txt
+++ b/CHANGES.txt
@@ -79,6 +79,7 @@ INCOMPATIBLE CHANGES
TEZ-1455. Replace deprecated junit.framework.Assert with org.junit.Assert
TEZ-1465. Update and document IntersectExample. Change name to JoinExample
TEZ-1449. Change user payloads to work with a byte buffer
+ TEZ-1472. AM/Session LRs are not shipped to vertices in new API use-case
ALL CHANGES
http://git-wip-us.apache.org/repos/asf/tez/blob/0a41e3a5/tez-api/src/main/java/org/apache/tez/client/AMConfiguration.java
----------------------------------------------------------------------
diff --git a/tez-api/src/main/java/org/apache/tez/client/AMConfiguration.java b/tez-api/src/main/java/org/apache/tez/client/AMConfiguration.java
index 9206e72..d8cc2bf 100644
--- a/tez-api/src/main/java/org/apache/tez/client/AMConfiguration.java
+++ b/tez-api/src/main/java/org/apache/tez/client/AMConfiguration.java
@@ -31,18 +31,18 @@ import com.google.common.collect.Maps;
@Private
class AMConfiguration {
- private Map<String, LocalResource> localResources;
+ private Map<String, LocalResource> amLocalResources = Maps.newHashMap();
private TezConfiguration tezConf;
private Credentials credentials;
private YarnConfiguration yarnConfig;
private Map<String, String> env;
+ private LocalResource binaryConfLRsrc;
AMConfiguration(TezConfiguration tezConf, Map<String, LocalResource> localResources,
Credentials credentials) {
- this.localResources = Maps.newHashMap();
this.tezConf = tezConf;
if (localResources != null) {
- addLocalResources(localResources);
+ addAMLocalResources(localResources);
}
if (credentials != null) {
setCredentials(credentials);
@@ -50,12 +50,12 @@ class AMConfiguration {
}
- void addLocalResources(Map<String, LocalResource> localResources) {
- this.localResources.putAll(localResources);
+ void addAMLocalResources(Map<String, LocalResource> localResources) {
+ this.amLocalResources.putAll(localResources);
}
- void clearLocalResources() {
- this.localResources.clear();
+ void clearAMLocalResources() {
+ this.amLocalResources.clear();
}
void setCredentials(Credentials credentials) {
@@ -74,8 +74,8 @@ class AMConfiguration {
return this.tezConf.get(TezConfiguration.TEZ_QUEUE_NAME);
}
- Map<String, LocalResource> getLocalResources() {
- return localResources;
+ Map<String, LocalResource> getAMLocalResources() {
+ return amLocalResources;
}
TezConfiguration getTezConfiguration() {
@@ -93,4 +93,13 @@ class AMConfiguration {
Map<String, String> getEnv() {
return env;
}
+
+ void setBinaryConfLR(LocalResource binaryConfLRsrc) {
+ this.binaryConfLRsrc = binaryConfLRsrc;
+ }
+
+ LocalResource getBinaryConfLR() {
+ return binaryConfLRsrc;
+ }
+
}
http://git-wip-us.apache.org/repos/asf/tez/blob/0a41e3a5/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 7250379..4b1f221 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
@@ -38,7 +38,6 @@ import org.apache.hadoop.yarn.api.records.LocalResource;
import org.apache.hadoop.yarn.api.records.LocalResourceType;
import org.apache.hadoop.yarn.conf.YarnConfiguration;
import org.apache.hadoop.yarn.exceptions.YarnException;
-import org.apache.tez.common.TezYARNUtils;
import org.apache.tez.common.security.JobTokenSecretManager;
import org.apache.tez.dag.api.DAG;
import org.apache.tez.dag.api.DAGSubmissionTimedOut;
@@ -48,7 +47,6 @@ import org.apache.tez.dag.api.SessionNotRunning;
import org.apache.tez.dag.api.TezConfiguration;
import org.apache.tez.dag.api.TezConstants;
import org.apache.tez.dag.api.TezException;
-import org.apache.tez.dag.api.Vertex;
import org.apache.tez.dag.api.client.DAGClient;
import org.apache.tez.dag.api.client.rpc.DAGClientAMProtocolBlockingPB;
import org.apache.tez.dag.api.client.rpc.DAGClientAMProtocolRPC.GetAMStatusRequestProto;
@@ -159,8 +157,8 @@ public class TezClient {
* as app master name is session mode
* @param tezConf
* Configuration for the framework
- * @param localResources
- * resources for the App Master
+ * @param localFiles
+ * local files for the App Master
* @param credentials
* Set security credentials to be used inside the app master, if
* needed. Tez App Master needs credentials to access the staging
@@ -174,9 +172,9 @@ public class TezClient {
* calling start()
*/
public static TezClient create(String name, TezConfiguration tezConf,
- @Nullable Map<String, LocalResource> localResources,
+ @Nullable Map<String, LocalResource> localFiles,
@Nullable Credentials credentials) {
- return new TezClient(name, tezConf, localResources, credentials);
+ return new TezClient(name, tezConf, localFiles, credentials);
}
/**
@@ -195,58 +193,59 @@ public class TezClient {
/**
* Create a new TezClient with AM session mode set explicitly. This overrides
* the setting from configuration.
- * Set the initial resources and security credentials for the App Master.
+ * Set the initial files and security credentials for the App Master.
* @param name
* Name of the client. Used for logging etc. This will also be used
* as app master name is session mode
* @param tezConf Configuration for the framework
* @param isSession The AM will run in session mode or not
- * @param localResources resources for the App Master
+ * @param localFiles local files for the App Master
* @param credentials credentials for the App Master
*/
public static TezClient create(String name, TezConfiguration tezConf, boolean isSession,
- @Nullable Map<String, LocalResource> localResources,
+ @Nullable Map<String, LocalResource> localFiles,
@Nullable Credentials credentials) {
- return new TezClient(name, tezConf, isSession, localResources, credentials);
+ return new TezClient(name, tezConf, isSession, localFiles, credentials);
}
/**
- * Add local resources for the DAG App Master. <br>
+ * Add local files for the DAG App Master. These may be files, archives,
+ * jars etc.<br>
* <p>
- * In non-session mode these will be added to the resources of the App Master
- * to be launched for the next DAG. Resources added via this method will
+ * In non-session mode these will be added to the files of the App Master
+ * to be launched for the next DAG. Files added via this method will
* accumulate and be used for every new App Master until
- * clearAppMasterLocalResource() is invoked. <br>
+ * {@link #clearAppMasterLocalFiles()} is invoked. <br>
* <p>
- * In session mode, the recommended usage is to add all resources before
- * calling start() so that all needed resources are available to the app
- * master before it starts. When called after start(), these local resources
+ * In session mode, the recommended usage is to add all files before
+ * calling start() so that all needed files are available to the app
+ * master before it starts. When called after start(), these local files
* will be re-localized to the running session DAG App Master and will be
* added to its classpath for execution of this DAG.
* <p>
- * Caveats for invoking this method after start() in session mode: Resources
+ * Caveats for invoking this method after start() in session mode: files
* accumulate across DAG submissions and are never removed from the classpath.
- * Only LocalResourceType.FILE is supported. All resources will be treated as
+ * Only LocalResourceType.FILE is supported. All files will be treated as
* private.
*
- * @param localResources
+ * @param localFiles
*/
- public synchronized void addAppMasterLocalResources(Map<String, LocalResource> localResources) {
- Preconditions.checkNotNull(localResources);
+ public synchronized void addAppMasterLocalFiles(Map<String, LocalResource> localFiles) {
+ Preconditions.checkNotNull(localFiles);
if (isSession && sessionStarted) {
- additionalLocalResources.putAll(localResources);
+ additionalLocalResources.putAll(localFiles);
}
- amConfig.addLocalResources(localResources);
+ amConfig.addAMLocalResources(localFiles);
}
/**
- * If the next DAG App Master needs different local resources, then use this
- * method to clear the local resources and then add the new local resources
- * using addAppMasterLocalResources(). This method is a no-op in session mode,
+ * If the next DAG App Master needs different local files, then use this
+ * method to clear the local files and then add the new local files
+ * using {@link #addAppMasterLocalFiles(Map)}. This method is a no-op in session mode,
* after start() is called.
*/
- public synchronized void clearAppMasterLocalResource() {
- amConfig.clearLocalResources();
+ public synchronized void clearAppMasterLocalFiles() {
+ amConfig.clearAMLocalResources();
}
/**
@@ -304,7 +303,7 @@ public class TezClient {
ApplicationSubmissionContext appContext =
TezClientUtils.createApplicationSubmissionContext(
- amConfig.getTezConfiguration(), sessionAppId,
+ sessionAppId,
null, clientName, amConfig,
tezJarResources, sessionCredentials);
@@ -365,19 +364,10 @@ public class TezClient {
+ lr.getType() + " is not supported, only " + LocalResourceType.FILE + " is supported");
}
}
-
- // Obtain DAG specific credentials.
- TezClientUtils.setupDAGCredentials(dag, sessionCredentials, amConfig.getTezConfiguration());
-
- // TODO TEZ-1229 - fix jar resources
- // setup env
- for (Vertex v : dag.getVertices()) {
- Map<String, String> taskEnv = v.getTaskEnvironment();
- TezYARNUtils.setupDefaultEnv(taskEnv, amConfig.getTezConfiguration(),
- TezConfiguration.TEZ_TASK_LAUNCH_ENV, TezConfiguration.TEZ_TASK_LAUNCH_ENV_DEFAULT,
- TezClientUtils.usingTezLibsFromArchive(getTezJarResources(sessionCredentials)));
- TezClientUtils.setDefaultLaunchCmdOpts(v, amConfig.getTezConfiguration());
- }
+
+ Map<String, LocalResource> tezJarResources = getTezJarResources(sessionCredentials);
+ TezClientUtils.updateDAGVertices(dag, amConfig, tezJarResources,
+ TezClientUtils.usingTezLibsFromArchive(tezJarResources), sessionCredentials);
DAGPlan dagPlan = dag.createDag(amConfig.getTezConfiguration());
SubmitDAGRequestProto.Builder requestBuilder = SubmitDAGRequestProto.newBuilder();
@@ -684,7 +674,7 @@ public class TezClient {
// Add credentials for tez-local resources.
Map<String, LocalResource> tezJarResources = getTezJarResources(credentials);
ApplicationSubmissionContext appContext = TezClientUtils
- .createApplicationSubmissionContext(amConfig.getTezConfiguration(),
+ .createApplicationSubmissionContext(
appId, dag, dag.getName(), amConfig, tezJarResources, credentials);
LOG.info("Submitting DAG to YARN"
+ ", applicationId=" + appId
http://git-wip-us.apache.org/repos/asf/tez/blob/0a41e3a5/tez-api/src/main/java/org/apache/tez/client/TezClientUtils.java
----------------------------------------------------------------------
diff --git a/tez-api/src/main/java/org/apache/tez/client/TezClientUtils.java b/tez-api/src/main/java/org/apache/tez/client/TezClientUtils.java
index 566a2d8..d931519 100644
--- a/tez-api/src/main/java/org/apache/tez/client/TezClientUtils.java
+++ b/tez-api/src/main/java/org/apache/tez/client/TezClientUtils.java
@@ -387,13 +387,13 @@ public class TezClientUtils {
* @throws YarnException
*/
static ApplicationSubmissionContext createApplicationSubmissionContext(
- TezConfiguration conf, ApplicationId appId, DAG dag, String amName,
+ ApplicationId appId, DAG dag, String amName,
AMConfiguration amConfig, Map<String, LocalResource> tezJarResources,
Credentials sessionCreds)
throws IOException, YarnException{
Preconditions.checkNotNull(sessionCreds);
-
+ TezConfiguration conf = amConfig.getTezConfiguration();
boolean tezLrsAsArchive = usingTezLibsFromArchive(tezJarResources);
FileSystem fs = TezClientUtils.ensureStagingDirExists(conf,
@@ -434,12 +434,6 @@ public class TezClientUtils {
amLaunchCredentials.writeTokenStorageToStream(dob);
securityTokens = ByteBuffer.wrap(dob.getData(), 0, dob.getLength());
- // Need to set credentials based on DAG and the URIs which have been set for the DAG.
-
- if (dag != null) {
- setupDAGCredentials(dag, sessionCreds, conf);
- }
-
// Setup the command to run the AM
List<String> vargs = new ArrayList<String>(8);
vargs.add(Environment.JAVA_HOME.$() + "/bin/java");
@@ -497,18 +491,17 @@ public class TezClientUtils {
}
}
- Map<String, LocalResource> localResources =
+ Map<String, LocalResource> amLocalResources =
new TreeMap<String, LocalResource>();
// Not fetching credentials for AMLocalResources. Expect this to be provided via AMCredentials.
- if (amConfig.getLocalResources() != null) {
- localResources.putAll(amConfig.getLocalResources());
+ if (amConfig.getAMLocalResources() != null) {
+ amLocalResources.putAll(amConfig.getAMLocalResources());
}
- localResources.putAll(tezJarResources);
+ amLocalResources.putAll(tezJarResources);
// emit conf as PB file
- Configuration finalTezConf = createFinalTezConfForApp(conf,
- amConfig.getTezConfiguration());
+ Configuration finalTezConf = createFinalTezConfForApp(amConfig.getTezConfiguration());
FSDataOutputStream amConfPBOutBinaryStream = null;
try {
@@ -535,27 +528,19 @@ public class TezClientUtils {
TezClientUtils.createLocalResource(fs,
binaryConfPath, LocalResourceType.FILE,
LocalResourceVisibility.APPLICATION);
- localResources.put(TezConstants.TEZ_PB_BINARY_CONF_NAME,
+ amConfig.setBinaryConfLR(binaryConfLRsrc);
+ amLocalResources.put(TezConstants.TEZ_PB_BINARY_CONF_NAME,
binaryConfLRsrc);
// Create Session Jars definition to be sent to AM as a local resource
- Path sessionJarsPath = TezCommonUtils.getTezSessionJarStagingPath(tezSysStagingPath);
+ Path sessionJarsPath = TezCommonUtils.getTezAMJarStagingPath(tezSysStagingPath);
FSDataOutputStream sessionJarsPBOutStream = null;
try {
- Map<String, LocalResource> sessionJars =
- new HashMap<String, LocalResource>(tezJarResources.size() + 1);
- sessionJars.putAll(tezJarResources);
- sessionJars.put(TezConstants.TEZ_PB_BINARY_CONF_NAME,
- binaryConfLRsrc);
- DAGProtos.PlanLocalResourcesProto proto =
- DagTypeConverters.convertFromLocalResources(sessionJars);
sessionJarsPBOutStream = TezCommonUtils.createFileForAM(fs, sessionJarsPath);
- proto.writeDelimitedTo(sessionJarsPBOutStream);
-
// Write out the initial list of resources which will be available in the AM
DAGProtos.PlanLocalResourcesProto amResourceProto;
- if (amConfig.getLocalResources() != null && !amConfig.getLocalResources().isEmpty()) {
- amResourceProto = DagTypeConverters.convertFromLocalResources(localResources);
+ if (amLocalResources != null && !amLocalResources.isEmpty()) {
+ amResourceProto = DagTypeConverters.convertFromLocalResources(amLocalResources);
} else {
amResourceProto = DAGProtos.PlanLocalResourcesProto.getDefaultInstance();
}
@@ -570,8 +555,8 @@ public class TezClientUtils {
TezClientUtils.createLocalResource(fs,
sessionJarsPath, LocalResourceType.FILE,
LocalResourceVisibility.APPLICATION);
- localResources.put(
- TezConstants.TEZ_SESSION_LOCAL_RESOURCES_PB_FILE_NAME,
+ amLocalResources.put(
+ TezConstants.TEZ_AM_LOCAL_RESOURCES_PB_FILE_NAME,
sessionJarsPBLRsrc);
String user = UserGroupInformation.getCurrentUser().getShortUserName();
@@ -579,21 +564,8 @@ public class TezClientUtils {
Map<ApplicationAccessType, String> acls = aclManager.toYARNACls();
if(dag != null) {
-
- for (Vertex v : dag.getVertices()) {
- if (tezJarResources != null) {
- v.getTaskLocalFiles().putAll(tezJarResources);
- }
- v.getTaskLocalFiles().put(TezConstants.TEZ_PB_BINARY_CONF_NAME,
- binaryConfLRsrc);
-
- Map<String, String> taskEnv = v.getTaskEnvironment();
- TezYARNUtils.setupDefaultEnv(taskEnv, conf,
- TezConfiguration.TEZ_TASK_LAUNCH_ENV,
- TezConfiguration.TEZ_TASK_LAUNCH_ENV_DEFAULT, tezLrsAsArchive);
-
- TezClientUtils.setDefaultLaunchCmdOpts(v, amConfig.getTezConfiguration());
- }
+
+ updateDAGVertices(dag, amConfig, tezJarResources, tezLrsAsArchive, sessionCreds);
// emit protobuf DAG file style
Path binaryPath = TezCommonUtils.getTezBinPlanStagingPath(tezSysStagingPath);
@@ -602,8 +574,6 @@ public class TezClientUtils {
+ tezSysStagingPath + " binaryConfPath :" + binaryConfPath + " sessionJarsPath :"
+ sessionJarsPath + " binaryPlanPath :" + binaryPath);
}
- amConfig.getTezConfiguration().set(TezConstants.TEZ_AM_PLAN_REMOTE_PATH,
- binaryPath.toUri().toString());
DAGPlan dagPB = dag.createDag(amConfig.getTezConfiguration());
@@ -619,14 +589,14 @@ public class TezClientUtils {
}
}
- localResources.put(TezConstants.TEZ_PB_PLAN_BINARY_NAME,
+ amLocalResources.put(TezConstants.TEZ_PB_PLAN_BINARY_NAME,
TezClientUtils.createLocalResource(fs,
binaryPath, LocalResourceType.FILE,
LocalResourceVisibility.APPLICATION));
if (Level.DEBUG.isGreaterOrEqual(Level.toLevel(amLogLevel))) {
Path textPath = localizeDagPlanAsText(dagPB, fs, amConfig, strAppId, tezSysStagingPath);
- localResources.put(TezConstants.TEZ_PB_PLAN_TEXT_NAME,
+ amLocalResources.put(TezConstants.TEZ_PB_PLAN_TEXT_NAME,
TezClientUtils.createLocalResource(fs,
textPath, LocalResourceType.FILE,
LocalResourceVisibility.APPLICATION));
@@ -635,7 +605,7 @@ public class TezClientUtils {
// Setup ContainerLaunchContext for AM container
ContainerLaunchContext amContainer =
- ContainerLaunchContext.newInstance(localResources, environment,
+ ContainerLaunchContext.newInstance(amLocalResources, environment,
vargsFinal, null, securityTokens, acls);
// Set up the ApplicationSubmissionContext
@@ -662,6 +632,25 @@ public class TezClientUtils {
}
+ static void updateDAGVertices(DAG dag, AMConfiguration amConfig,
+ Map<String, LocalResource> tezJarResources, boolean tezLrsAsArchive,
+ Credentials credentials) throws IOException {
+ setupDAGCredentials(dag, credentials, amConfig.getTezConfiguration());
+ for (Vertex v : dag.getVertices()) {
+ if (tezJarResources != null) {
+ v.getTaskLocalFiles().putAll(tezJarResources);
+ }
+ v.getTaskLocalFiles().put(TezConstants.TEZ_PB_BINARY_CONF_NAME,
+ amConfig.getBinaryConfLR());
+
+ Map<String, String> taskEnv = v.getTaskEnvironment();
+ TezYARNUtils.setupDefaultEnv(taskEnv, amConfig.getTezConfiguration(),
+ TezConfiguration.TEZ_TASK_LAUNCH_ENV,
+ TezConfiguration.TEZ_TASK_LAUNCH_ENV_DEFAULT, tezLrsAsArchive);
+
+ setDefaultLaunchCmdOpts(v, amConfig.getTezConfiguration());
+ }
+ }
static void maybeAddDefaultLoggingJavaOpts(String logLevel, List<String> vargs) {
if (vargs != null && !vargs.isEmpty()) {
@@ -714,29 +703,14 @@ public class TezClientUtils {
+ "," + TezConstants.TEZ_CONTAINER_LOGGER_NAME);
}
- static Configuration createFinalTezConfForApp(TezConfiguration tezConf,
- TezConfiguration amConf) {
+ static Configuration createFinalTezConfForApp(TezConfiguration amConf) {
Configuration conf = new Configuration(false);
conf.setQuietMode(true);
- assert tezConf != null;
assert amConf != null;
Entry<String, String> entry;
- Iterator<Entry<String, String>> iter = tezConf.iterator();
- while (iter.hasNext()) {
- entry = iter.next();
- // Copy all tez config parameters.
- if (entry.getKey().startsWith(TezConfiguration.TEZ_PREFIX)) {
- conf.set(entry.getKey(), entry.getValue());
- if (LOG.isDebugEnabled()) {
- LOG.debug("Adding tez dag am parameter from conf: " + entry.getKey()
- + ", with value: " + entry.getValue());
- }
- }
- }
-
- iter = amConf.iterator();
+ Iterator<Entry<String, String>> iter = amConf.iterator();
while (iter.hasNext()) {
entry = iter.next();
// Copy all tez config parameters.
http://git-wip-us.apache.org/repos/asf/tez/blob/0a41e3a5/tez-api/src/main/java/org/apache/tez/common/TezCommonUtils.java
----------------------------------------------------------------------
diff --git a/tez-api/src/main/java/org/apache/tez/common/TezCommonUtils.java b/tez-api/src/main/java/org/apache/tez/common/TezCommonUtils.java
index 15878c3..5148770 100644
--- a/tez-api/src/main/java/org/apache/tez/common/TezCommonUtils.java
+++ b/tez-api/src/main/java/org/apache/tez/common/TezCommonUtils.java
@@ -22,6 +22,7 @@ import java.io.IOException;
import java.util.ArrayList;
import java.util.Collection;
import java.util.List;
+import java.util.Map;
import java.util.StringTokenizer;
import java.util.zip.Deflater;
import java.util.zip.DeflaterOutputStream;
@@ -38,6 +39,7 @@ import org.apache.hadoop.fs.Path;
import org.apache.hadoop.fs.permission.FsPermission;
import org.apache.hadoop.security.Credentials;
import org.apache.hadoop.security.token.Token;
+import org.apache.hadoop.yarn.api.records.LocalResource;
import org.apache.tez.client.TezClient;
import org.apache.tez.dag.api.TezConfiguration;
import org.apache.tez.dag.api.TezConstants;
@@ -165,8 +167,8 @@ public class TezCommonUtils {
* @return path to store the session jars
*/
@Private
- public static Path getTezSessionJarStagingPath(Path tezSysStagingPath) {
- return new Path(tezSysStagingPath, TezConstants.TEZ_SESSION_LOCAL_RESOURCES_PB_FILE_NAME);
+ public static Path getTezAMJarStagingPath(Path tezSysStagingPath) {
+ return new Path(tezSysStagingPath, TezConstants.TEZ_AM_LOCAL_RESOURCES_PB_FILE_NAME);
}
/**
@@ -295,6 +297,19 @@ public class TezCommonUtils {
public static FSDataOutputStream createFileForAM(FileSystem fs, Path filePath) throws IOException {
return FileSystem.create(fs, filePath, new FsPermission(TEZ_AM_FILE_PERMISSION));
}
+
+ public static void addAdditionalLocalResources(Map<String, LocalResource> additionalLrs,
+ Map<String, LocalResource> originalLRs) {
+ if (additionalLrs != null && !additionalLrs.isEmpty()) {
+ for (Map.Entry<String, LocalResource> lr : additionalLrs.entrySet()) {
+ if (originalLRs.containsKey(lr.getKey())) {
+ throw new TezUncheckedException("Attempting to add duplicate resource: " + lr.getKey());
+ } else {
+ originalLRs.put(lr.getKey(), lr.getValue());
+ }
+ }
+ }
+ }
@Private
public static ByteString compressByteArrayToByteString(byte[] inBytes) throws IOException {
http://git-wip-us.apache.org/repos/asf/tez/blob/0a41e3a5/tez-api/src/main/java/org/apache/tez/dag/api/DAG.java
----------------------------------------------------------------------
diff --git a/tez-api/src/main/java/org/apache/tez/dag/api/DAG.java b/tez-api/src/main/java/org/apache/tez/dag/api/DAG.java
index 5ad724c..fe73c47 100644
--- a/tez-api/src/main/java/org/apache/tez/dag/api/DAG.java
+++ b/tez-api/src/main/java/org/apache/tez/dag/api/DAG.java
@@ -63,6 +63,7 @@ import org.apache.tez.dag.api.records.DAGProtos.VertexPlan;
import com.google.common.annotations.VisibleForTesting;
import com.google.common.base.Preconditions;
import com.google.common.collect.Lists;
+import com.google.common.collect.Maps;
import com.google.common.collect.Sets;
/**
@@ -85,13 +86,28 @@ public class DAG {
Set<VertexGroup> vertexGroups = Sets.newHashSet();
Set<GroupInputEdge> groupInputEdges = Sets.newHashSet();
private DAGAccessControls dagAccessControls;
-
+ Map<String, LocalResource> commonTaskLocalFiles = Maps.newHashMap();
+
private Stack<String> topologicalVertexStack = new Stack<String>();
public DAG(String name) {
this.name = name;
}
+ /**
+ * Set the files etc that must be provided to the tasks of this DAG
+ * @param localFiles
+ * files that must be available locally for each task. These files
+ * may be regular files, archives etc. as specified by the value
+ * elements of the map.
+ * @return {@link DAG}
+ */
+ public DAG addTaskLocalFiles(Map<String, LocalResource> localFiles) {
+ Preconditions.checkNotNull(localFiles);
+ TezCommonUtils.addAdditionalLocalResources(localFiles, commonTaskLocalFiles);
+ return this;
+ }
+
public synchronized DAG addVertex(Vertex vertex) {
if (vertices.containsKey(vertex.getName())) {
throw new IllegalStateException(
@@ -597,7 +613,7 @@ public class DAG {
if (dataSource.getCredentials() != null) {
credentials.addAll(dataSource.getCredentials());
}
- vertex.addAdditionalLocalResources(dataSource.getAdditionalLocalResources());
+ vertex.addTaskLocalFiles(dataSource.getAdditionalLocalFiles());
}
if (dataSources.size() == 1) {
DataSourceDescriptor dataSource = dataSources.get(0);
@@ -614,6 +630,9 @@ public class DAG {
}
}
+ // add common task files for this DAG
+ vertex.addTaskLocalFiles(commonTaskLocalFiles);
+
VertexPlan.Builder vertexBuilder = VertexPlan.newBuilder();
vertexBuilder.setName(vertex.getName());
vertexBuilder.setType(PlanVertexType.NORMAL); // vertex type is implicitly NORMAL until TEZ-46.
@@ -752,7 +771,8 @@ public class DAG {
confProtoBuilder.addConfKeyValues(kvp);
}
}
- dagBuilder.setDagKeyValues(confProtoBuilder);
+ dagBuilder.setDagKeyValues(confProtoBuilder); // This does not seem to be used anywhere
+ // should this replace BINARY_PB_CONF???
if (credentials != null) {
dagBuilder.setCredentialsBinary(DagTypeConverters.convertCredentialsToProto(credentials));
http://git-wip-us.apache.org/repos/asf/tez/blob/0a41e3a5/tez-api/src/main/java/org/apache/tez/dag/api/DataSourceDescriptor.java
----------------------------------------------------------------------
diff --git a/tez-api/src/main/java/org/apache/tez/dag/api/DataSourceDescriptor.java b/tez-api/src/main/java/org/apache/tez/dag/api/DataSourceDescriptor.java
index fc9cf23..f8e6072 100644
--- a/tez-api/src/main/java/org/apache/tez/dag/api/DataSourceDescriptor.java
+++ b/tez-api/src/main/java/org/apache/tez/dag/api/DataSourceDescriptor.java
@@ -42,7 +42,7 @@ public class DataSourceDescriptor {
private final Credentials credentials;
private final int numShards;
private final VertexLocationHint locationHint;
- private final Map<String, LocalResource> additionalLocalResources;
+ private final Map<String, LocalResource> additionalLocalFiles;
private DataSourceDescriptor(InputDescriptor inputDescriptor,
@Nullable InputInitializerDescriptor initializerDescriptor,
@@ -55,13 +55,13 @@ public class DataSourceDescriptor {
int numShards,
@Nullable Credentials credentials,
@Nullable VertexLocationHint locationHint,
- @Nullable Map<String, LocalResource> additionalLocalResources) {
+ @Nullable Map<String, LocalResource> additionalLocalFiles) {
this.inputDescriptor = inputDescriptor;
this.initializerDescriptor = initializerDescriptor;
this.numShards = numShards;
this.credentials = credentials;
this.locationHint = locationHint;
- this.additionalLocalResources = additionalLocalResources;
+ this.additionalLocalFiles = additionalLocalFiles;
}
/**
@@ -104,19 +104,19 @@ public class DataSourceDescriptor {
* @param numShards Number of shards of data
* @param credentials Credentials needed to access the data
* @param locationHint Location hints for the vertex tasks
- * @param additionalLocalResources additional local resources required by this Input. An attempt
- * will be made to add these resources to the Vertex as Private
+ * @param additionalLocalFiles additional local files required by this Input. An attempt
+ * will be made to add these files to the Vertex as Private
* resources. If a name conflict occurs, a {@link
- * org.apache.tez.dag.api.TezException} will be thrown
+ * org.apache.tez.dag.api.TezUncheckedException} will be thrown
*/
public static DataSourceDescriptor create(InputDescriptor inputDescriptor,
@Nullable InputInitializerDescriptor initializerDescriptor,
int numShards,
@Nullable Credentials credentials,
@Nullable VertexLocationHint locationHint,
- @Nullable Map<String, LocalResource> additionalLocalResources) {
+ @Nullable Map<String, LocalResource> additionalLocalFiles) {
return new DataSourceDescriptor(inputDescriptor, initializerDescriptor, numShards, credentials,
- locationHint, additionalLocalResources);
+ locationHint, additionalLocalFiles);
}
public InputDescriptor getInputDescriptor() {
@@ -160,12 +160,12 @@ public class DataSourceDescriptor {
}
/**
- * Get the list of additional local resources which were specified during creation.
+ * Get the list of additional local files which were specified during creation.
* @return
*/
@InterfaceAudience.Private
- public @Nullable Map<String, LocalResource> getAdditionalLocalResources() {
- return additionalLocalResources;
+ public @Nullable Map<String, LocalResource> getAdditionalLocalFiles() {
+ return additionalLocalFiles;
}
http://git-wip-us.apache.org/repos/asf/tez/blob/0a41e3a5/tez-api/src/main/java/org/apache/tez/dag/api/TezConstants.java
----------------------------------------------------------------------
diff --git a/tez-api/src/main/java/org/apache/tez/dag/api/TezConstants.java b/tez-api/src/main/java/org/apache/tez/dag/api/TezConstants.java
index ba634af..8af7e84 100644
--- a/tez-api/src/main/java/org/apache/tez/dag/api/TezConstants.java
+++ b/tez-api/src/main/java/org/apache/tez/dag/api/TezConstants.java
@@ -63,7 +63,7 @@ public class TezConstants {
public static final String TEZ_CONTAINER_ERR_FILE_NAME = "stderr";
public static final String TEZ_CONTAINER_OUT_FILE_NAME = "stdout";
- public static final String TEZ_SESSION_LOCAL_RESOURCES_PB_FILE_NAME =
+ public static final String TEZ_AM_LOCAL_RESOURCES_PB_FILE_NAME =
TezConfiguration.TEZ_SESSION_PREFIX + "local-resources.pb";
public static final String TEZ_APPLICATION_TYPE = "TEZ";
@@ -83,13 +83,6 @@ public class TezConstants {
// Configuration keys used internally and not set by the users
- /**
- * The complete path to the serialized dag plan file
- * <code>TEZ_AM_PLAN_PB_BINARY</code>. Used to make the plan available to
- * individual tasks if needed. This will be inside the staging dir
- */
- public static final String TEZ_AM_PLAN_REMOTE_PATH = "dag-am-plan.remote.path";
-
// These are session specific DAG ACL's. Currently here because these can only be specified
// via code in the API.
/**
http://git-wip-us.apache.org/repos/asf/tez/blob/0a41e3a5/tez-api/src/main/java/org/apache/tez/dag/api/Vertex.java
----------------------------------------------------------------------
diff --git a/tez-api/src/main/java/org/apache/tez/dag/api/Vertex.java b/tez-api/src/main/java/org/apache/tez/dag/api/Vertex.java
index 5e15e16..6397c82 100644
--- a/tez-api/src/main/java/org/apache/tez/dag/api/Vertex.java
+++ b/tez-api/src/main/java/org/apache/tez/dag/api/Vertex.java
@@ -28,6 +28,7 @@ import org.apache.hadoop.classification.InterfaceAudience;
import org.apache.hadoop.classification.InterfaceAudience.Public;
import org.apache.hadoop.yarn.api.records.LocalResource;
import org.apache.hadoop.yarn.api.records.Resource;
+import org.apache.tez.common.TezCommonUtils;
import org.apache.tez.dag.api.VertexGroup.GroupInfo;
import org.apache.tez.dag.api.VertexLocationHint.TaskLocationHint;
import org.apache.tez.runtime.api.LogicalIOProcessor;
@@ -247,9 +248,9 @@ public class Vertex {
* elements of the map.
* @return this Vertex
*/
- public Vertex setTaskLocalFiles(Map<String, LocalResource> localFiles) {
+ public Vertex addTaskLocalFiles(Map<String, LocalResource> localFiles) {
if (localFiles != null) {
- this.taskLocalResources.putAll(localFiles);
+ TezCommonUtils.addAdditionalLocalResources(localFiles, taskLocalResources);
}
return this;
}
@@ -428,18 +429,6 @@ public class Vertex {
return Collections.unmodifiableList(outputVertices);
}
- void addAdditionalLocalResources(Map<String, LocalResource> additionalLrs) {
- if (additionalLrs != null && !additionalLrs.isEmpty()) {
- for (Map.Entry<String, LocalResource> lr : additionalLrs.entrySet()) {
- if (taskLocalResources.containsKey(lr.getKey())) {
- throw new TezUncheckedException("Attempting to add duplicate resource: " + lr.getKey());
- } else {
- taskLocalResources.put(lr.getKey(), lr.getValue());
- }
- }
- }
- }
-
/**
* Set the cpu/memory etc resources used by tasks of this vertex
* @param resource {@link Resource} for the tasks of this vertex
http://git-wip-us.apache.org/repos/asf/tez/blob/0a41e3a5/tez-api/src/test/java/org/apache/tez/client/TestTezClient.java
----------------------------------------------------------------------
diff --git a/tez-api/src/test/java/org/apache/tez/client/TestTezClient.java b/tez-api/src/test/java/org/apache/tez/client/TestTezClient.java
index ddddb16..e05b004 100644
--- a/tez-api/src/test/java/org/apache/tez/client/TestTezClient.java
+++ b/tez-api/src/test/java/org/apache/tez/client/TestTezClient.java
@@ -19,6 +19,7 @@
package org.apache.tez.client;
import java.io.IOException;
+import java.util.Collections;
import java.util.Map;
import java.util.concurrent.atomic.AtomicReference;
@@ -134,7 +135,7 @@ public class TestTezClient {
ApplicationSubmissionContext context = captor.getValue();
Assert.assertEquals(3, context.getAMContainerSpec().getLocalResources().size());
Assert.assertTrue(context.getAMContainerSpec().getLocalResources().containsKey(
- TezConstants.TEZ_SESSION_LOCAL_RESOURCES_PB_FILE_NAME));
+ TezConstants.TEZ_AM_LOCAL_RESOURCES_PB_FILE_NAME));
Assert.assertTrue(context.getAMContainerSpec().getLocalResources().containsKey(
TezConstants.TEZ_PB_BINARY_CONF_NAME));
Assert.assertTrue(context.getAMContainerSpec().getLocalResources().containsKey(
@@ -143,10 +144,19 @@ public class TestTezClient {
verify(client.mockYarnClient, times(0)).submitApplication(captor.capture());
}
- DAG dag = new DAG("DAG").addVertex(
- Vertex.create("Vertex", ProcessorDescriptor.create("P"), 1, Resource.newInstance(1, 1)));
+ String mockLR1Name = "LR1";
+ Map<String, LocalResource> lrDAG = Collections.singletonMap(mockLR1Name, LocalResource
+ .newInstance(URL.newInstance("file:///", "localhost", 0, "test"), LocalResourceType.FILE,
+ LocalResourceVisibility.PUBLIC, 1, 1));
+ Vertex vertex = Vertex.create("Vertex", ProcessorDescriptor.create("P"), 1,
+ Resource.newInstance(1, 1));
+ DAG dag = new DAG("DAG").addVertex(vertex).addTaskLocalFiles(lrDAG);
DAGClient dagClient = client.submitDAG(dag);
+ // verify that both DAG and TezClient localResources are added to the vertex
+ Map<String, LocalResource> vertexLR = vertex.getTaskLocalFiles();
+ Assert.assertTrue(vertexLR.containsKey(mockLR1Name));
+
Assert.assertTrue(dagClient.getExecutionContext().contains(client.mockAppId.toString()));
if (isSession) {
@@ -157,7 +167,7 @@ public class TestTezClient {
ApplicationSubmissionContext context = captor.getValue();
Assert.assertEquals(4, context.getAMContainerSpec().getLocalResources().size());
Assert.assertTrue(context.getAMContainerSpec().getLocalResources().containsKey(
- TezConstants.TEZ_SESSION_LOCAL_RESOURCES_PB_FILE_NAME));
+ TezConstants.TEZ_AM_LOCAL_RESOURCES_PB_FILE_NAME));
Assert.assertTrue(context.getAMContainerSpec().getLocalResources().containsKey(
TezConstants.TEZ_PB_BINARY_CONF_NAME));
Assert.assertTrue(context.getAMContainerSpec().getLocalResources().containsKey(
@@ -171,7 +181,7 @@ public class TestTezClient {
lrs.clear();
lrs.put(lrName2, LocalResource.newInstance(URL.newInstance("file:///", "localhost", 0, "test"),
LocalResourceType.FILE, LocalResourceVisibility.PUBLIC, 1, 1));
- client.addAppMasterLocalResources(lrs);
+ client.addAppMasterLocalFiles(lrs);
ApplicationId appId2 = ApplicationId.newInstance(0, 2);
when(client.mockYarnClient.createApplication().getNewApplicationResponse().getApplicationId())
@@ -199,7 +209,7 @@ public class TestTezClient {
ApplicationSubmissionContext context = captor.getValue();
Assert.assertEquals(5, context.getAMContainerSpec().getLocalResources().size());
Assert.assertTrue(context.getAMContainerSpec().getLocalResources().containsKey(
- TezConstants.TEZ_SESSION_LOCAL_RESOURCES_PB_FILE_NAME));
+ TezConstants.TEZ_AM_LOCAL_RESOURCES_PB_FILE_NAME));
Assert.assertTrue(context.getAMContainerSpec().getLocalResources().containsKey(
TezConstants.TEZ_PB_BINARY_CONF_NAME));
Assert.assertTrue(context.getAMContainerSpec().getLocalResources().containsKey(
http://git-wip-us.apache.org/repos/asf/tez/blob/0a41e3a5/tez-api/src/test/java/org/apache/tez/common/TestTezCommonUtils.java
----------------------------------------------------------------------
diff --git a/tez-api/src/test/java/org/apache/tez/common/TestTezCommonUtils.java b/tez-api/src/test/java/org/apache/tez/common/TestTezCommonUtils.java
index 27f798d..e0975ab 100644
--- a/tez-api/src/test/java/org/apache/tez/common/TestTezCommonUtils.java
+++ b/tez-api/src/test/java/org/apache/tez/common/TestTezCommonUtils.java
@@ -136,10 +136,10 @@ public class TestTezCommonUtils {
public void testTezSessionJarStagingPath() throws Exception {
String strAppId = "testAppId";
Path stageDir = TezCommonUtils.getTezSystemStagingPath(conf, strAppId);
- Path confStageDir = TezCommonUtils.getTezSessionJarStagingPath(stageDir);
+ Path confStageDir = TezCommonUtils.getTezAMJarStagingPath(stageDir);
String expectedDir = RESOLVED_STAGE_DIR + File.separatorChar
+ TezCommonUtils.TEZ_SYSTEM_SUB_DIR + File.separatorChar + strAppId + File.separator
- + TezConstants.TEZ_SESSION_LOCAL_RESOURCES_PB_FILE_NAME;
+ + TezConstants.TEZ_AM_LOCAL_RESOURCES_PB_FILE_NAME;
Assert.assertEquals(confStageDir.toString(), expectedDir);
}
http://git-wip-us.apache.org/repos/asf/tez/blob/0a41e3a5/tez-api/src/test/java/org/apache/tez/dag/api/TestDAGPlan.java
----------------------------------------------------------------------
diff --git a/tez-api/src/test/java/org/apache/tez/dag/api/TestDAGPlan.java b/tez-api/src/test/java/org/apache/tez/dag/api/TestDAGPlan.java
index 6a74553..fbea24b 100644
--- a/tez-api/src/test/java/org/apache/tez/dag/api/TestDAGPlan.java
+++ b/tez-api/src/test/java/org/apache/tez/dag/api/TestDAGPlan.java
@@ -108,9 +108,9 @@ public class TestDAGPlan {
Vertex v1 = Vertex.create("v1", pd1, 10, Resource.newInstance(1024, 1));
Vertex v2 = Vertex.create("v2", pd2, 1, Resource.newInstance(1024, 1));
v1.setTaskLaunchCmdOpts("").setTaskEnvironment(new HashMap<String, String>())
- .setTaskLocalFiles(new HashMap<String, LocalResource>());
+ .addTaskLocalFiles(new HashMap<String, LocalResource>());
v2.setTaskLaunchCmdOpts("").setTaskEnvironment(new HashMap<String, String>())
- .setTaskLocalFiles(new HashMap<String, LocalResource>());
+ .addTaskLocalFiles(new HashMap<String, LocalResource>());
InputDescriptor inputDescriptor = InputDescriptor.create("input")
.setUserPayload(UserPayload.create(ByteBuffer.wrap("inputBytes".getBytes())));
@@ -144,9 +144,9 @@ public class TestDAGPlan {
Vertex v1 = Vertex.create("v1", pd1, 10, Resource.newInstance(1024, 1));
Vertex v2 = Vertex.create("v2", pd2, 1, Resource.newInstance(1024, 1));
v1.setTaskLaunchCmdOpts("").setTaskEnvironment(new HashMap<String, String>())
- .setTaskLocalFiles(new HashMap<String, LocalResource>());
+ .addTaskLocalFiles(new HashMap<String, LocalResource>());
v2.setTaskLaunchCmdOpts("").setTaskEnvironment(new HashMap<String, String>())
- .setTaskLocalFiles(new HashMap<String, LocalResource>());
+ .addTaskLocalFiles(new HashMap<String, LocalResource>());
InputDescriptor inputDescriptor = InputDescriptor.create("input").
setUserPayload(UserPayload.create(ByteBuffer.wrap("inputBytes".getBytes())));
@@ -208,11 +208,11 @@ public class TestDAGPlan {
Vertex v2 = Vertex.create("v2", pd2, 1, Resource.newInstance(1024, 1));
Vertex v3 = Vertex.create("v3", pd3, 1, Resource.newInstance(1024, 1));
v1.setTaskLaunchCmdOpts("").setTaskEnvironment(new HashMap<String, String>())
- .setTaskLocalFiles(new HashMap<String, LocalResource>());
+ .addTaskLocalFiles(new HashMap<String, LocalResource>());
v2.setTaskLaunchCmdOpts("").setTaskEnvironment(new HashMap<String, String>())
- .setTaskLocalFiles(new HashMap<String, LocalResource>());
+ .addTaskLocalFiles(new HashMap<String, LocalResource>());
v3.setTaskLaunchCmdOpts("").setTaskEnvironment(new HashMap<String, String>())
- .setTaskLocalFiles(new HashMap<String, LocalResource>());
+ .addTaskLocalFiles(new HashMap<String, LocalResource>());
InputDescriptor inputDescriptor = InputDescriptor.create("input").
setUserPayload(UserPayload.create(ByteBuffer.wrap("inputBytes".getBytes())));
@@ -278,9 +278,9 @@ public class TestDAGPlan {
Vertex v1 = Vertex.create("v1", pd1, 10, Resource.newInstance(1024, 1));
Vertex v2 = Vertex.create("v2", pd2, 1, Resource.newInstance(1024, 1));
v1.setTaskLaunchCmdOpts("").setTaskEnvironment(new HashMap<String, String>())
- .setTaskLocalFiles(new HashMap<String, LocalResource>());
+ .addTaskLocalFiles(new HashMap<String, LocalResource>());
v2.setTaskLaunchCmdOpts("").setTaskEnvironment(new HashMap<String, String>())
- .setTaskLocalFiles(new HashMap<String, LocalResource>());
+ .addTaskLocalFiles(new HashMap<String, LocalResource>());
InputDescriptor inputDescriptor = InputDescriptor.create("input").
setUserPayload(UserPayload.create(ByteBuffer.wrap("inputBytes".getBytes())));
http://git-wip-us.apache.org/repos/asf/tez/blob/0a41e3a5/tez-api/src/test/java/org/apache/tez/dag/api/TestDAGVerify.java
----------------------------------------------------------------------
diff --git a/tez-api/src/test/java/org/apache/tez/dag/api/TestDAGVerify.java b/tez-api/src/test/java/org/apache/tez/dag/api/TestDAGVerify.java
index 9885272..033d08b 100644
--- a/tez-api/src/test/java/org/apache/tez/dag/api/TestDAGVerify.java
+++ b/tez-api/src/test/java/org/apache/tez/dag/api/TestDAGVerify.java
@@ -19,10 +19,15 @@
package org.apache.tez.dag.api;
import java.util.Arrays;
+import java.util.Map;
import java.nio.ByteBuffer;
import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.yarn.api.records.LocalResource;
+import org.apache.hadoop.yarn.api.records.LocalResourceType;
+import org.apache.hadoop.yarn.api.records.LocalResourceVisibility;
import org.apache.hadoop.yarn.api.records.Resource;
+import org.apache.hadoop.yarn.api.records.URL;
import org.apache.tez.common.security.DAGAccessControls;
import org.apache.tez.dag.api.EdgeProperty.DataMovementType;
import org.apache.tez.dag.api.EdgeProperty.DataSourceType;
@@ -33,6 +38,8 @@ import org.apache.tez.dag.api.records.DAGProtos.PlanKeyValuePair;
import org.junit.Assert;
import org.junit.Test;
+import com.google.common.collect.Maps;
+
public class TestDAGVerify {
private final String dummyProcessorClassName = TestDAGVerify.class.getName();
@@ -896,8 +903,80 @@ public class TestDAGVerify {
dag.createDag(new TezConfiguration());
}
+
+
+ @Test(timeout = 5000)
+ public void testVerifyCommonFiles() {
+ Vertex v1 = Vertex.create("v1",
+ ProcessorDescriptor.create(dummyProcessorClassName),
+ dummyTaskCount, dummyTaskResource);
+ Vertex v2 = Vertex.create("v2",
+ ProcessorDescriptor.create("MapProcessor"),
+ dummyTaskCount, dummyTaskResource);
+ Edge e1 = Edge.create(v1, v2,
+ EdgeProperty.create(DataMovementType.SCATTER_GATHER,
+ DataSourceType.PERSISTED, SchedulingType.SEQUENTIAL,
+ OutputDescriptor.create(dummyOutputClassName),
+ InputDescriptor.create(dummyInputClassName)));
+ Map<String, LocalResource> lrs = Maps.newHashMap();
+ String lrName1 = "LR1";
+ lrs.put(lrName1, LocalResource.newInstance(URL.newInstance("file:///", "localhost", 0, "test"),
+ LocalResourceType.FILE, LocalResourceVisibility.PUBLIC, 1, 1));
+
+ DAG dag = new DAG("testDag");
+ dag.addVertex(v1);
+ dag.addVertex(v2);
+ dag.addEdge(e1);
+ dag.addTaskLocalFiles(lrs);
+ dag.createDag(new TezConfiguration());
+ Assert.assertTrue(v1.getTaskLocalFiles().containsKey(lrName1));
+ Assert.assertTrue(v2.getTaskLocalFiles().containsKey(lrName1));
+ }
-
+ @Test(timeout = 5000)
+ public void testVerifyCommonFilesFail() {
+ Vertex v1 = Vertex.create("v1",
+ ProcessorDescriptor.create(dummyProcessorClassName),
+ dummyTaskCount, dummyTaskResource);
+ Vertex v2 = Vertex.create("v2",
+ ProcessorDescriptor.create("MapProcessor"),
+ dummyTaskCount, dummyTaskResource);
+ Edge e1 = Edge.create(v1, v2,
+ EdgeProperty.create(DataMovementType.SCATTER_GATHER,
+ DataSourceType.PERSISTED, SchedulingType.SEQUENTIAL,
+ OutputDescriptor.create(dummyOutputClassName),
+ InputDescriptor.create(dummyInputClassName)));
+ Map<String, LocalResource> lrs = Maps.newHashMap();
+ String lrName1 = "LR1";
+ lrs.put(lrName1, LocalResource.newInstance(URL.newInstance("file:///", "localhost", 0, "test"),
+ LocalResourceType.FILE, LocalResourceVisibility.PUBLIC, 1, 1));
+ v1.addTaskLocalFiles(lrs);
+ try {
+ v1.addTaskLocalFiles(lrs);
+ Assert.fail();
+ } catch (TezUncheckedException e) {
+ Assert.assertTrue(e.getMessage().contains("Attempting to add duplicate resource"));
+ }
+ DAG dag = new DAG("testDag");
+ dag.addVertex(v1);
+ dag.addVertex(v2);
+ dag.addEdge(e1);
+ dag.addTaskLocalFiles(lrs);
+ try {
+ dag.addTaskLocalFiles(lrs);
+ Assert.fail();
+ } catch (TezUncheckedException e) {
+ Assert.assertTrue(e.getMessage().contains("Attempting to add duplicate resource"));
+ }
+ try {
+ // dag will add duplicate common files to vertex
+ dag.createDag(new TezConfiguration());
+ Assert.fail();
+ } catch (TezUncheckedException e) {
+ Assert.assertTrue(e.getMessage().contains("Attempting to add duplicate resource"));
+ }
+ }
+
@Test(timeout = 5000)
public void testDAGAccessControls() {
DAG dag = new DAG("testDag");
http://git-wip-us.apache.org/repos/asf/tez/blob/0a41e3a5/tez-dag/src/main/java/org/apache/tez/dag/app/AppContext.java
----------------------------------------------------------------------
diff --git a/tez-dag/src/main/java/org/apache/tez/dag/app/AppContext.java b/tez-dag/src/main/java/org/apache/tez/dag/app/AppContext.java
index 89c468b..9cc28cb 100644
--- a/tez-dag/src/main/java/org/apache/tez/dag/app/AppContext.java
+++ b/tez-dag/src/main/java/org/apache/tez/dag/app/AppContext.java
@@ -26,7 +26,6 @@ import org.apache.hadoop.fs.Path;
import org.apache.hadoop.yarn.api.records.ApplicationAccessType;
import org.apache.hadoop.yarn.api.records.ApplicationAttemptId;
import org.apache.hadoop.yarn.api.records.ApplicationId;
-import org.apache.hadoop.yarn.api.records.LocalResource;
import org.apache.hadoop.yarn.event.EventHandler;
import org.apache.hadoop.yarn.util.Clock;
import org.apache.tez.dag.app.dag.DAG;
@@ -79,8 +78,6 @@ public interface AppContext {
TaskSchedulerEventHandler getTaskScheduler();
- Map<String, LocalResource> getSessionResources();
-
boolean isSession();
DAGAppMasterState getAMState();
http://git-wip-us.apache.org/repos/asf/tez/blob/0a41e3a5/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 fa4b629..62004d2 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
@@ -216,8 +216,6 @@ public class DAGAppMaster extends AbstractService {
private HistoryEventHandler historyEventHandler;
private final Map<String, LocalResource> amResources = new HashMap<String, LocalResource>();
private final Map<String, LocalResource> cumulativeAdditionalResources = new HashMap<String, LocalResource>();
- private final Map<String, LocalResource> sessionResources =
- new HashMap<String, LocalResource>();
private boolean isLocal = false; //Local mode flag
@@ -398,15 +396,10 @@ public class DAGAppMaster extends AbstractService {
FileInputStream sessionResourcesStream = null;
try {
sessionResourcesStream = new FileInputStream(
- new File(workingDirectory, TezConstants.TEZ_SESSION_LOCAL_RESOURCES_PB_FILE_NAME));
- PlanLocalResourcesProto sessionLocalResourcesProto =
- PlanLocalResourcesProto.parseDelimitedFrom(sessionResourcesStream);
+ new File(workingDirectory, TezConstants.TEZ_AM_LOCAL_RESOURCES_PB_FILE_NAME));
PlanLocalResourcesProto amLocalResourceProto = PlanLocalResourcesProto
.parseDelimitedFrom(sessionResourcesStream);
- sessionResources.putAll(DagTypeConverters.convertFromPlanLocalResources(
- sessionLocalResourcesProto));
amResources.putAll(DagTypeConverters.convertFromPlanLocalResources(amLocalResourceProto));
- amResources.putAll(sessionResources);
} finally {
if (sessionResourcesStream != null) {
sessionResourcesStream.close();
@@ -1192,11 +1185,6 @@ public class DAGAppMaster extends AbstractService {
}
@Override
- public Map<String, LocalResource> getSessionResources() {
- return sessionResources;
- }
-
- @Override
public boolean isSession() {
return isSession;
}
http://git-wip-us.apache.org/repos/asf/tez/blob/0a41e3a5/tez-dag/src/main/java/org/apache/tez/dag/app/dag/impl/VertexImpl.java
----------------------------------------------------------------------
diff --git a/tez-dag/src/main/java/org/apache/tez/dag/app/dag/impl/VertexImpl.java b/tez-dag/src/main/java/org/apache/tez/dag/app/dag/impl/VertexImpl.java
index 6387b8f..1fb40a0 100644
--- a/tez-dag/src/main/java/org/apache/tez/dag/app/dag/impl/VertexImpl.java
+++ b/tez-dag/src/main/java/org/apache/tez/dag/app/dag/impl/VertexImpl.java
@@ -648,7 +648,6 @@ public class VertexImpl implements org.apache.tez.dag.app.dag.Vertex,
this.localResources = DagTypeConverters
.createLocalResourceMapFromDAGPlan(vertexPlan.getTaskConfig()
.getLocalResourceList());
- this.localResources.putAll(appContext.getSessionResources());
this.environment = DagTypeConverters
.createEnvironmentMapFromDAGPlan(vertexPlan.getTaskConfig()
.getEnvironmentSettingList());
http://git-wip-us.apache.org/repos/asf/tez/blob/0a41e3a5/tez-mapreduce/src/main/java/org/apache/tez/mapreduce/client/YARNRunner.java
----------------------------------------------------------------------
diff --git a/tez-mapreduce/src/main/java/org/apache/tez/mapreduce/client/YARNRunner.java b/tez-mapreduce/src/main/java/org/apache/tez/mapreduce/client/YARNRunner.java
index 94d5818..a966316 100644
--- a/tez-mapreduce/src/main/java/org/apache/tez/mapreduce/client/YARNRunner.java
+++ b/tez-mapreduce/src/main/java/org/apache/tez/mapreduce/client/YARNRunner.java
@@ -444,7 +444,7 @@ public class YARNRunner implements ClientProtocol {
: MRHelpers.getJavaOptsForMRReducer(stageConf);
vertex.setTaskEnvironment(taskEnv)
- .setTaskLocalFiles(taskLocalResources)
+ .addTaskLocalFiles(taskLocalResources)
.setLocationHint(VertexLocationHint.create(locations))
.setTaskLaunchCmdOpts(taskJavaOpts);
http://git-wip-us.apache.org/repos/asf/tez/blob/0a41e3a5/tez-mapreduce/src/test/java/org/apache/tez/mapreduce/hadoop/TestMRInputHelpers.java
----------------------------------------------------------------------
diff --git a/tez-mapreduce/src/test/java/org/apache/tez/mapreduce/hadoop/TestMRInputHelpers.java b/tez-mapreduce/src/test/java/org/apache/tez/mapreduce/hadoop/TestMRInputHelpers.java
index 3f9368f..c5f81a9 100644
--- a/tez-mapreduce/src/test/java/org/apache/tez/mapreduce/hadoop/TestMRInputHelpers.java
+++ b/tez-mapreduce/src/test/java/org/apache/tez/mapreduce/hadoop/TestMRInputHelpers.java
@@ -104,9 +104,9 @@ public class TestMRInputHelpers {
DataSourceDescriptor dataSource = generateDataSourceDescriptorMapReduce(newSplitsDir);
- Assert.assertTrue(dataSource.getAdditionalLocalResources()
+ Assert.assertTrue(dataSource.getAdditionalLocalFiles()
.containsKey(MRInputHelpers.JOB_SPLIT_RESOURCE_NAME));
- Assert.assertTrue(dataSource.getAdditionalLocalResources()
+ Assert.assertTrue(dataSource.getAdditionalLocalFiles()
.containsKey(MRInputHelpers.JOB_SPLIT_METAINFO_RESOURCE_NAME));
RemoteIterator<LocatedFileStatus> files =
@@ -141,8 +141,8 @@ public class TestMRInputHelpers {
public void testOldSplitsGen() throws Exception {
DataSourceDescriptor dataSource = generateDataSourceDescriptorMapRed(oldSplitsDir);
Assert.assertTrue(
- dataSource.getAdditionalLocalResources().containsKey(MRInputHelpers.JOB_SPLIT_RESOURCE_NAME));
- Assert.assertTrue(dataSource.getAdditionalLocalResources()
+ dataSource.getAdditionalLocalFiles().containsKey(MRInputHelpers.JOB_SPLIT_RESOURCE_NAME));
+ Assert.assertTrue(dataSource.getAdditionalLocalFiles()
.containsKey(MRInputHelpers.JOB_SPLIT_METAINFO_RESOURCE_NAME));
RemoteIterator<LocatedFileStatus> files =
@@ -177,7 +177,7 @@ public class TestMRInputHelpers {
public void testInputSplitLocalResourceCreation() throws Exception {
DataSourceDescriptor dataSource = generateDataSourceDescriptorMapRed(oldSplitsDir);
- Map<String, LocalResource> localResources = dataSource.getAdditionalLocalResources();
+ Map<String, LocalResource> localResources = dataSource.getAdditionalLocalFiles();
Assert.assertEquals(2, localResources.size());
Assert.assertTrue(localResources.containsKey(
http://git-wip-us.apache.org/repos/asf/tez/blob/0a41e3a5/tez-tests/src/main/java/org/apache/tez/mapreduce/examples/FilterLinesByWord.java
----------------------------------------------------------------------
diff --git a/tez-tests/src/main/java/org/apache/tez/mapreduce/examples/FilterLinesByWord.java b/tez-tests/src/main/java/org/apache/tez/mapreduce/examples/FilterLinesByWord.java
index 875bfcb..d9304e4 100644
--- a/tez-tests/src/main/java/org/apache/tez/mapreduce/examples/FilterLinesByWord.java
+++ b/tez-tests/src/main/java/org/apache/tez/mapreduce/examples/FilterLinesByWord.java
@@ -171,7 +171,7 @@ public class FilterLinesByWord extends Configured implements Tool {
// Setup stage1 Vertex
Vertex stage1Vertex = Vertex.create("stage1", ProcessorDescriptor.create(
FilterByWordInputProcessor.class.getName()).setUserPayload(stage1Payload))
- .setTaskLocalFiles(commonLocalResources);
+ .addTaskLocalFiles(commonLocalResources);
DataSourceDescriptor dsd;
if (generateSplitsInClient) {
@@ -189,7 +189,7 @@ public class FilterLinesByWord extends Configured implements Tool {
Vertex stage2Vertex = Vertex.create("stage2", ProcessorDescriptor.create(
FilterByWordOutputProcessor.class.getName()).setUserPayload(
TezUtils.createUserPayloadFromConf(stage2Conf)), 1);
- stage2Vertex.setTaskLocalFiles(commonLocalResources);
+ stage2Vertex.addTaskLocalFiles(commonLocalResources);
// Configure the Output for stage2
OutputDescriptor od = OutputDescriptor.create(MROutput.class.getName())
http://git-wip-us.apache.org/repos/asf/tez/blob/0a41e3a5/tez-tests/src/main/java/org/apache/tez/mapreduce/examples/FilterLinesByWordOneToOne.java
----------------------------------------------------------------------
diff --git a/tez-tests/src/main/java/org/apache/tez/mapreduce/examples/FilterLinesByWordOneToOne.java b/tez-tests/src/main/java/org/apache/tez/mapreduce/examples/FilterLinesByWordOneToOne.java
index 61d827e..ca2a288 100644
--- a/tez-tests/src/main/java/org/apache/tez/mapreduce/examples/FilterLinesByWordOneToOne.java
+++ b/tez-tests/src/main/java/org/apache/tez/mapreduce/examples/FilterLinesByWordOneToOne.java
@@ -162,7 +162,7 @@ public class FilterLinesByWordOneToOne extends Configured implements Tool {
// Setup stage1 Vertex
Vertex stage1Vertex = Vertex.create("stage1", ProcessorDescriptor.create(
FilterByWordInputProcessor.class.getName()).setUserPayload(stage1Payload))
- .setTaskLocalFiles(commonLocalResources);
+ .addTaskLocalFiles(commonLocalResources);
DataSourceDescriptor dsd;
if (generateSplitsInClient) {
@@ -180,7 +180,7 @@ public class FilterLinesByWordOneToOne extends Configured implements Tool {
Vertex stage2Vertex = Vertex.create("stage2", ProcessorDescriptor.create(
FilterByWordOutputProcessor.class.getName()).setUserPayload(TezUtils
.createUserPayloadFromConf(stage2Conf)), dsd.getNumberOfShards());
- stage2Vertex.setTaskLocalFiles(commonLocalResources);
+ stage2Vertex.addTaskLocalFiles(commonLocalResources);
// Configure the Output for stage2
stage2Vertex.addDataSink(
http://git-wip-us.apache.org/repos/asf/tez/blob/0a41e3a5/tez-tests/src/main/java/org/apache/tez/mapreduce/examples/MRRSleepJob.java
----------------------------------------------------------------------
diff --git a/tez-tests/src/main/java/org/apache/tez/mapreduce/examples/MRRSleepJob.java b/tez-tests/src/main/java/org/apache/tez/mapreduce/examples/MRRSleepJob.java
index e4c515b..38a1045 100644
--- a/tez-tests/src/main/java/org/apache/tez/mapreduce/examples/MRRSleepJob.java
+++ b/tez-tests/src/main/java/org/apache/tez/mapreduce/examples/MRRSleepJob.java
@@ -526,7 +526,7 @@ public class MRRSleepJob extends Configured implements Tool {
Vertex mapVertex = Vertex.create("map", ProcessorDescriptor.create(
MapProcessor.class.getName()).setUserPayload(mapUserPayload), numTasks)
- .setTaskLocalFiles(commonLocalResources);
+ .addTaskLocalFiles(commonLocalResources);
mapVertex.addDataSource("MRInput", dataSource);
vertices.add(mapVertex);
@@ -539,7 +539,7 @@ public class MRRSleepJob extends Configured implements Tool {
Vertex ivertex = Vertex.create("ireduce" + (i + 1),
ProcessorDescriptor.create(ReduceProcessor.class.getName()).
setUserPayload(iReduceUserPayload), numIReducer);
- ivertex.setTaskLocalFiles(commonLocalResources);
+ ivertex.addTaskLocalFiles(commonLocalResources);
vertices.add(ivertex);
}
}
@@ -549,7 +549,7 @@ public class MRRSleepJob extends Configured implements Tool {
UserPayload reducePayload = TezUtils.createUserPayloadFromConf(finalReduceConf);
finalReduceVertex = Vertex.create("reduce", ProcessorDescriptor.create(
ReduceProcessor.class.getName()).setUserPayload(reducePayload), numReducer);
- finalReduceVertex.setTaskLocalFiles(commonLocalResources);
+ finalReduceVertex.addTaskLocalFiles(commonLocalResources);
finalReduceVertex.addDataSink("MROutput", MROutputLegacy.createConfigBuilder(finalReduceConf,
NullOutputFormat.class).build());
vertices.add(finalReduceVertex);
http://git-wip-us.apache.org/repos/asf/tez/blob/0a41e3a5/tez-tests/src/main/java/org/apache/tez/mapreduce/examples/TestOrderedWordCount.java
----------------------------------------------------------------------
diff --git a/tez-tests/src/main/java/org/apache/tez/mapreduce/examples/TestOrderedWordCount.java b/tez-tests/src/main/java/org/apache/tez/mapreduce/examples/TestOrderedWordCount.java
index 3c4e66f..6157cbd 100644
--- a/tez-tests/src/main/java/org/apache/tez/mapreduce/examples/TestOrderedWordCount.java
+++ b/tez-tests/src/main/java/org/apache/tez/mapreduce/examples/TestOrderedWordCount.java
@@ -199,7 +199,7 @@ public class TestOrderedWordCount extends Configured implements Tool {
Vertex mapVertex = Vertex.create("initialmap", ProcessorDescriptor.create(
MapProcessor.class.getName()).setUserPayload(
TezUtils.createUserPayloadFromConf(mapStageConf))
- .setHistoryText(mapStageHistoryText)).setTaskLocalFiles(commonLocalResources);
+ .setHistoryText(mapStageHistoryText)).addTaskLocalFiles(commonLocalResources);
mapVertex.addDataSource("MRInput", dsd);
vertices.add(mapVertex);
@@ -210,7 +210,7 @@ public class TestOrderedWordCount extends Configured implements Tool {
ReduceProcessor.class.getName())
.setUserPayload(TezUtils.createUserPayloadFromConf(iReduceStageConf))
.setHistoryText(iReduceStageHistoryText), 2);
- ivertex.setTaskLocalFiles(commonLocalResources);
+ ivertex.addTaskLocalFiles(commonLocalResources);
vertices.add(ivertex);
ByteArrayOutputStream finalReduceOutputStream = new ByteArrayOutputStream(4096);
@@ -222,7 +222,7 @@ public class TestOrderedWordCount extends Configured implements Tool {
ReduceProcessor.class.getName())
.setUserPayload(finalReducePayload)
.setHistoryText(finalReduceStageHistoryText), 1);
- finalReduceVertex.setTaskLocalFiles(commonLocalResources);
+ finalReduceVertex.addTaskLocalFiles(commonLocalResources);
finalReduceVertex.addDataSink("MROutput",
MROutputLegacy.createConfigBuilder(finalReduceConf, TextOutputFormat.class, outputPath)
.build());
@@ -386,7 +386,7 @@ public class TestOrderedWordCount extends Configured implements Tool {
LOG.info("Pre-warming Session");
PreWarmVertex preWarmVertex = PreWarmVertex.create("PreWarm", preWarmNumContainers, dag
.getVertex("initialmap").getTaskResource());
- preWarmVertex.setTaskLocalFiles(dag.getVertex("initialmap").getTaskLocalFiles());
+ preWarmVertex.addTaskLocalFiles(dag.getVertex("initialmap").getTaskLocalFiles());
preWarmVertex.setTaskEnvironment(dag.getVertex("initialmap").getTaskEnvironment());
preWarmVertex.setTaskLaunchCmdOpts(dag.getVertex("initialmap").getTaskLaunchCmdOpts());
http://git-wip-us.apache.org/repos/asf/tez/blob/0a41e3a5/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 49de6c1..48faefe 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
@@ -785,7 +785,7 @@ public class TestMRRJobsDAGApi {
LOG.info("Submitting dag to tez session with appId=" + tezSession.getAppMasterApplicationId()
+ " and Dag Name=" + dag.getName());
if (additionalLocalResources != null) {
- tezSession.addAppMasterLocalResources(additionalLocalResources);
+ tezSession.addAppMasterLocalFiles(additionalLocalResources);
}
dagClient = tezSession.submitDAG(dag);
Assert.assertEquals(TezAppMasterStatus.RUNNING,
[2/4] git commit: TEZ-1472. Separate method calls for creating
InputDataInformationEvent with serialized/unserialized payloads (Siddharth
Seth via bikas)
Posted by bi...@apache.org.
TEZ-1472. Separate method calls for creating InputDataInformationEvent with serialized/unserialized payloads (Siddharth Seth via bikas)
Project: http://git-wip-us.apache.org/repos/asf/tez/repo
Commit: http://git-wip-us.apache.org/repos/asf/tez/commit/6dcd2313
Tree: http://git-wip-us.apache.org/repos/asf/tez/tree/6dcd2313
Diff: http://git-wip-us.apache.org/repos/asf/tez/diff/6dcd2313
Branch: refs/heads/branch-0.5.0
Commit: 6dcd2313e57e78557aaa93d37710f2eeabcf4bcf
Parents: 0a41e3a
Author: Bikas Saha <bi...@apache.org>
Authored: Wed Aug 20 19:20:41 2014 -0700
Committer: Bikas Saha <bi...@apache.org>
Committed: Thu Aug 21 13:47:25 2014 -0700
----------------------------------------------------------------------
CHANGES.txt | 4 +++-
.../api/events/InputDataInformationEvent.java | 10 ++++++----
.../dag/app/dag/impl/TestRootInputVertexManager.java | 6 ++++--
.../apache/tez/dag/app/dag/impl/TestVertexImpl.java | 9 ++++++---
.../mapreduce/common/MRInputAMSplitGenerator.java | 9 ++++++---
.../mapreduce/common/MRInputSplitDistributor.java | 10 ++++------
.../apache/tez/mapreduce/input/TestMultiMRInput.java | 15 ++++++++++-----
.../java/org/apache/tez/common/ProtoConverters.java | 2 +-
8 files changed, 40 insertions(+), 25 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/tez/blob/6dcd2313/CHANGES.txt
----------------------------------------------------------------------
diff --git a/CHANGES.txt b/CHANGES.txt
index d5852b0..729d5d4 100644
--- a/CHANGES.txt
+++ b/CHANGES.txt
@@ -79,7 +79,9 @@ INCOMPATIBLE CHANGES
TEZ-1455. Replace deprecated junit.framework.Assert with org.junit.Assert
TEZ-1465. Update and document IntersectExample. Change name to JoinExample
TEZ-1449. Change user payloads to work with a byte buffer
- TEZ-1472. AM/Session LRs are not shipped to vertices in new API use-case
+ TEZ-1469. AM/Session LRs are not shipped to vertices in new API use-case
+ TEZ-1472. Separate method calls for creating InputDataInformationEvent with
+ serialized/unserialized payloads
ALL CHANGES
http://git-wip-us.apache.org/repos/asf/tez/blob/6dcd2313/tez-api/src/main/java/org/apache/tez/runtime/api/events/InputDataInformationEvent.java
----------------------------------------------------------------------
diff --git a/tez-api/src/main/java/org/apache/tez/runtime/api/events/InputDataInformationEvent.java b/tez-api/src/main/java/org/apache/tez/runtime/api/events/InputDataInformationEvent.java
index ffeb7ff..a62a341 100644
--- a/tez-api/src/main/java/org/apache/tez/runtime/api/events/InputDataInformationEvent.java
+++ b/tez-api/src/main/java/org/apache/tez/runtime/api/events/InputDataInformationEvent.java
@@ -58,7 +58,7 @@ public final class InputDataInformationEvent extends Event {
this.userPayloadObject = null;
}
- private InputDataInformationEvent(int srcIndex, Object userPayloadDeserialized) {
+ private InputDataInformationEvent(int srcIndex, Object userPayloadDeserialized, Object sigChanged) {
this.sourceIndex = srcIndex;
this.userPayloadObject = userPayloadDeserialized;
this.userPayload = null;
@@ -69,12 +69,14 @@ public final class InputDataInformationEvent extends Event {
* @param srcIndex the src index
* @param userPayload the serialized payload
*/
- public static InputDataInformationEvent create(int srcIndex, ByteBuffer userPayload) {
+ public static InputDataInformationEvent createWithSerializedPayload(int srcIndex,
+ ByteBuffer userPayload) {
return new InputDataInformationEvent(srcIndex, userPayload);
}
- public static InputDataInformationEvent create(int srcIndex, Object userPayloadDeserialized) {
- return new InputDataInformationEvent(srcIndex, userPayloadDeserialized);
+ public static InputDataInformationEvent createWithObjectPayload(int srcIndex,
+ Object userPayloadDeserialized) {
+ return new InputDataInformationEvent(srcIndex, userPayloadDeserialized, null);
}
public int getSourceIndex() {
http://git-wip-us.apache.org/repos/asf/tez/blob/6dcd2313/tez-dag/src/test/java/org/apache/tez/dag/app/dag/impl/TestRootInputVertexManager.java
----------------------------------------------------------------------
diff --git a/tez-dag/src/test/java/org/apache/tez/dag/app/dag/impl/TestRootInputVertexManager.java b/tez-dag/src/test/java/org/apache/tez/dag/app/dag/impl/TestRootInputVertexManager.java
index b9aafa2..fc02bbf 100644
--- a/tez-dag/src/test/java/org/apache/tez/dag/app/dag/impl/TestRootInputVertexManager.java
+++ b/tez-dag/src/test/java/org/apache/tez/dag/app/dag/impl/TestRootInputVertexManager.java
@@ -48,14 +48,16 @@ public class TestRootInputVertexManager {
InputDescriptor id1 = mock(InputDescriptor.class);
List<Event> events1 = new LinkedList<Event>();
- InputDataInformationEvent diEvent11 = InputDataInformationEvent.create(0, null);
+ InputDataInformationEvent diEvent11 = InputDataInformationEvent.createWithSerializedPayload(0,
+ null);
events1.add(diEvent11);
rootInputVertexManager.onRootVertexInitialized("input1", id1, events1);
// All good so far, single input only.
InputDescriptor id2 = mock(InputDescriptor.class);
List<Event> events2 = new LinkedList<Event>();
- InputDataInformationEvent diEvent21 = InputDataInformationEvent.create(0, null);
+ InputDataInformationEvent diEvent21 = InputDataInformationEvent.createWithSerializedPayload(0,
+ null);
events2.add(diEvent21);
try {
// Should fail due to second input
http://git-wip-us.apache.org/repos/asf/tez/blob/6dcd2313/tez-dag/src/test/java/org/apache/tez/dag/app/dag/impl/TestVertexImpl.java
----------------------------------------------------------------------
diff --git a/tez-dag/src/test/java/org/apache/tez/dag/app/dag/impl/TestVertexImpl.java b/tez-dag/src/test/java/org/apache/tez/dag/app/dag/impl/TestVertexImpl.java
index 30b4275..75d0fff 100644
--- a/tez-dag/src/test/java/org/apache/tez/dag/app/dag/impl/TestVertexImpl.java
+++ b/tez-dag/src/test/java/org/apache/tez/dag/app/dag/impl/TestVertexImpl.java
@@ -2914,7 +2914,8 @@ public class TestVertexImpl {
VertexManagerEvent.create("vertex2", ByteBuffer.wrap(new byte[0])), new EventMetaData(
EventProducerConsumerType.PROCESSOR, "vertex1", "vertex2",
ta0_t0_v1)));
- events.add(new TezEvent(InputDataInformationEvent.create(0, ByteBuffer.wrap(new byte[0])),
+ events.add(new TezEvent(InputDataInformationEvent.createWithSerializedPayload(0,
+ ByteBuffer.wrap(new byte[0])),
new EventMetaData(EventProducerConsumerType.INPUT, "vertex2",
"NULL_VERTEX", null)));
dispatcher.getEventHandler().handle(
@@ -3259,7 +3260,8 @@ public class TestVertexImpl {
targetTasks, VertexLocationHint.create(locationHints), null);
events.add(configEvent);
for (int i = 0; i < targetTasks; i++) {
- InputDataInformationEvent diEvent = InputDataInformationEvent.create(i, null);
+ InputDataInformationEvent diEvent = InputDataInformationEvent.createWithSerializedPayload(i,
+ null);
events.add(diEvent);
}
eventHandler.handle(new VertexEventRootInputInitialized(vertexID, inputs
@@ -3462,7 +3464,8 @@ public class TestVertexImpl {
lock.unlock();
}
initComplete.set(true);
- InputDataInformationEvent diEvent = InputDataInformationEvent.create(0, ByteBuffer.wrap(new byte[]{0}));
+ InputDataInformationEvent diEvent = InputDataInformationEvent.createWithSerializedPayload(0,
+ ByteBuffer.wrap(new byte[]{0}));
List<Event> eventList = new LinkedList<Event>();
eventList.add(diEvent);
return eventList;
http://git-wip-us.apache.org/repos/asf/tez/blob/6dcd2313/tez-mapreduce/src/main/java/org/apache/tez/mapreduce/common/MRInputAMSplitGenerator.java
----------------------------------------------------------------------
diff --git a/tez-mapreduce/src/main/java/org/apache/tez/mapreduce/common/MRInputAMSplitGenerator.java b/tez-mapreduce/src/main/java/org/apache/tez/mapreduce/common/MRInputAMSplitGenerator.java
index 2408ddb..884054b 100644
--- a/tez-mapreduce/src/main/java/org/apache/tez/mapreduce/common/MRInputAMSplitGenerator.java
+++ b/tez-mapreduce/src/main/java/org/apache/tez/mapreduce/common/MRInputAMSplitGenerator.java
@@ -142,7 +142,8 @@ public class MRInputAMSplitGenerator extends InputInitializer {
int count = 0;
for (MRSplitProto mrSplit : splitsProto.getSplitsList()) {
// Unnecessary array copy, can be avoided by using ByteBuffer instead of a raw array.
- InputDataInformationEvent diEvent = InputDataInformationEvent.create(count++,
+ InputDataInformationEvent diEvent = InputDataInformationEvent.createWithSerializedPayload(
+ count++,
mrSplit.toByteString().asReadOnlyByteBuffer());
events.add(diEvent);
}
@@ -150,12 +151,14 @@ public class MRInputAMSplitGenerator extends InputInitializer {
int count = 0;
if (inputSplitInfo.holdsNewFormatSplits()) {
for (org.apache.hadoop.mapreduce.InputSplit split : inputSplitInfo.getNewFormatSplits()) {
- InputDataInformationEvent diEvent = InputDataInformationEvent.create(count++, split);
+ InputDataInformationEvent diEvent = InputDataInformationEvent.createWithObjectPayload(
+ count++, split);
events.add(diEvent);
}
} else {
for (org.apache.hadoop.mapred.InputSplit split : inputSplitInfo.getOldFormatSplits()) {
- InputDataInformationEvent diEvent = InputDataInformationEvent.create(count++, split);
+ InputDataInformationEvent diEvent = InputDataInformationEvent.createWithObjectPayload(
+ count++, split);
events.add(diEvent);
}
}
http://git-wip-us.apache.org/repos/asf/tez/blob/6dcd2313/tez-mapreduce/src/main/java/org/apache/tez/mapreduce/common/MRInputSplitDistributor.java
----------------------------------------------------------------------
diff --git a/tez-mapreduce/src/main/java/org/apache/tez/mapreduce/common/MRInputSplitDistributor.java b/tez-mapreduce/src/main/java/org/apache/tez/mapreduce/common/MRInputSplitDistributor.java
index e28a3a5..1307687 100644
--- a/tez-mapreduce/src/main/java/org/apache/tez/mapreduce/common/MRInputSplitDistributor.java
+++ b/tez-mapreduce/src/main/java/org/apache/tez/mapreduce/common/MRInputSplitDistributor.java
@@ -21,7 +21,6 @@ package org.apache.tez.mapreduce.common;
import java.io.IOException;
import java.util.List;
-import com.google.protobuf.ByteString;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.hadoop.classification.InterfaceAudience.Public;
@@ -39,14 +38,12 @@ import org.apache.tez.mapreduce.protos.MRRuntimeProtos.MRSplitsProto;
import org.apache.tez.runtime.api.Event;
import org.apache.tez.runtime.api.InputInitializer;
import org.apache.tez.runtime.api.InputInitializerContext;
-import org.apache.tez.runtime.api.events.DataMovementEvent;
import org.apache.tez.runtime.api.events.InputDataInformationEvent;
import org.apache.tez.runtime.api.events.InputInitializerEvent;
import org.apache.tez.runtime.api.events.InputUpdatePayloadEvent;
import com.google.common.base.Stopwatch;
import com.google.common.collect.Lists;
-import org.apache.tez.runtime.library.shuffle.impl.ShuffleUserPayloads;
/**
* Implements an {@link InputInitializer} that distributes Map Reduce
@@ -111,16 +108,17 @@ public class MRInputSplitDistributor extends InputInitializer {
if (sendSerializedEvents) {
// Unnecessary array copy, can be avoided by using ByteBuffer instead of
// a raw array.
- diEvent = InputDataInformationEvent.create(count++, mrSplit.toByteString().asReadOnlyByteBuffer());
+ diEvent = InputDataInformationEvent.createWithSerializedPayload(count++,
+ mrSplit.toByteString().asReadOnlyByteBuffer());
} else {
if (useNewApi) {
org.apache.hadoop.mapreduce.InputSplit newInputSplit = MRInputUtils
.getNewSplitDetailsFromEvent(mrSplit, conf);
- diEvent = InputDataInformationEvent.create(count++, newInputSplit);
+ diEvent = InputDataInformationEvent.createWithObjectPayload(count++, newInputSplit);
} else {
org.apache.hadoop.mapred.InputSplit oldInputSplit = MRInputUtils
.getOldSplitDetailsFromEvent(mrSplit, conf);
- diEvent = InputDataInformationEvent.create(count++, oldInputSplit);
+ diEvent = InputDataInformationEvent.createWithObjectPayload(count++, oldInputSplit);
}
}
events.add(diEvent);
http://git-wip-us.apache.org/repos/asf/tez/blob/6dcd2313/tez-mapreduce/src/test/java/org/apache/tez/mapreduce/input/TestMultiMRInput.java
----------------------------------------------------------------------
diff --git a/tez-mapreduce/src/test/java/org/apache/tez/mapreduce/input/TestMultiMRInput.java b/tez-mapreduce/src/test/java/org/apache/tez/mapreduce/input/TestMultiMRInput.java
index 121a975..05f6bbc 100644
--- a/tez-mapreduce/src/test/java/org/apache/tez/mapreduce/input/TestMultiMRInput.java
+++ b/tez-mapreduce/src/test/java/org/apache/tez/mapreduce/input/TestMultiMRInput.java
@@ -112,7 +112,8 @@ public class TestMultiMRInput {
MRSplitProto splitProto = MRInputHelpers.createSplitProto(splits[0]);
InputDataInformationEvent event =
- InputDataInformationEvent.create(0, splitProto.toByteString().asReadOnlyByteBuffer());
+ InputDataInformationEvent.createWithSerializedPayload(0,
+ splitProto.toByteString().asReadOnlyByteBuffer());
eventList.clear();
eventList.add(event);
@@ -172,11 +173,13 @@ public class TestMultiMRInput {
MRSplitProto splitProto1 = MRInputHelpers.createSplitProto(splits[0]);
InputDataInformationEvent event1 =
- InputDataInformationEvent.create(0, splitProto1.toByteString().asReadOnlyByteBuffer());
+ InputDataInformationEvent.createWithSerializedPayload(0,
+ splitProto1.toByteString().asReadOnlyByteBuffer());
MRSplitProto splitProto2 = MRInputHelpers.createSplitProto(splits[1]);
InputDataInformationEvent event2 =
- InputDataInformationEvent.create(0, splitProto2.toByteString().asReadOnlyByteBuffer());
+ InputDataInformationEvent.createWithSerializedPayload(0,
+ splitProto2.toByteString().asReadOnlyByteBuffer());
eventList.clear();
eventList.add(event1);
@@ -225,9 +228,11 @@ public class TestMultiMRInput {
MRSplitProto splitProto = MRInputHelpers.createSplitProto(splits[0]);
InputDataInformationEvent event1 =
- InputDataInformationEvent.create(0, splitProto.toByteString().asReadOnlyByteBuffer());
+ InputDataInformationEvent.createWithSerializedPayload(0,
+ splitProto.toByteString().asReadOnlyByteBuffer());
InputDataInformationEvent event2 =
- InputDataInformationEvent.create(1, splitProto.toByteString().asReadOnlyByteBuffer());
+ InputDataInformationEvent.createWithSerializedPayload(1,
+ splitProto.toByteString().asReadOnlyByteBuffer());
eventList.clear();
eventList.add(event1);
http://git-wip-us.apache.org/repos/asf/tez/blob/6dcd2313/tez-runtime-internals/src/main/java/org/apache/tez/common/ProtoConverters.java
----------------------------------------------------------------------
diff --git a/tez-runtime-internals/src/main/java/org/apache/tez/common/ProtoConverters.java b/tez-runtime-internals/src/main/java/org/apache/tez/common/ProtoConverters.java
index 0650a90..f765f42 100644
--- a/tez-runtime-internals/src/main/java/org/apache/tez/common/ProtoConverters.java
+++ b/tez-runtime-internals/src/main/java/org/apache/tez/common/ProtoConverters.java
@@ -101,7 +101,7 @@ public class ProtoConverters {
public static InputDataInformationEvent
convertRootInputDataInformationEventFromProto(
EventProtos.RootInputDataInformationEventProto proto) {
- InputDataInformationEvent diEvent = InputDataInformationEvent.create(
+ InputDataInformationEvent diEvent = InputDataInformationEvent.createWithSerializedPayload(
proto.getSourceIndex(),
proto.hasUserPayload() ? proto.getUserPayload().asReadOnlyByteBuffer() : null);
diEvent.setTargetIndex(proto.getTargetIndex());
[3/4] git commit: TEZ-1436. Fix javadoc warnings (Jonathan Eagles via
bikas)
Posted by bi...@apache.org.
TEZ-1436. Fix javadoc warnings (Jonathan Eagles via bikas)
Project: http://git-wip-us.apache.org/repos/asf/tez/repo
Commit: http://git-wip-us.apache.org/repos/asf/tez/commit/c90918c6
Tree: http://git-wip-us.apache.org/repos/asf/tez/tree/c90918c6
Diff: http://git-wip-us.apache.org/repos/asf/tez/diff/c90918c6
Branch: refs/heads/branch-0.5.0
Commit: c90918c6e3ba6be2d4cc2c5feb5b66ab9f678b36
Parents: 6dcd231
Author: Bikas Saha <bi...@apache.org>
Authored: Thu Aug 21 13:45:00 2014 -0700
Committer: Bikas Saha <bi...@apache.org>
Committed: Thu Aug 21 13:47:59 2014 -0700
----------------------------------------------------------------------
.../tez/common/security/DAGAccessControls.java | 16 ++++++++--------
.../apache/tez/dag/api/DataSourceDescriptor.java | 2 +-
.../src/main/java/org/apache/tez/dag/api/Edge.java | 2 +-
.../org/apache/tez/dag/api/EntityDescriptor.java | 3 ++-
.../java/org/apache/tez/dag/api/PreWarmVertex.java | 2 +-
.../main/java/org/apache/tez/dag/api/Vertex.java | 5 ++++-
.../java/org/apache/tez/dag/api/VertexGroup.java | 3 ++-
.../org/apache/tez/runtime/api/InputSpecUpdate.java | 4 ++--
.../runtime/api/ProcessorFrameworkInterface.java | 1 -
.../org/apache/tez/dag/history/SummaryEvent.java | 2 +-
.../apache/tez/mapreduce/hadoop/InputSplitInfo.java | 4 ++--
.../apache/tez/mapreduce/hadoop/MRInputHelpers.java | 2 +-
.../tez/runtime/library/common/sort/impl/IFile.java | 5 +++--
.../tez/runtime/library/conf/BaseConfigBuilder.java | 6 +++---
.../library/conf/OrderedGroupedKVInputConfig.java | 6 +++---
.../conf/OrderedPartitionedKVEdgeConfig.java | 4 ++--
.../conf/OrderedPartitionedKVOutputConfig.java | 4 ++--
.../runtime/library/conf/UnorderedKVEdgeConfig.java | 4 ++--
.../library/conf/UnorderedKVInputConfig.java | 4 ++--
.../library/conf/UnorderedKVOutputConfig.java | 4 ++--
.../conf/UnorderedPartitionedKVEdgeConfig.java | 4 ++--
.../conf/UnorderedPartitionedKVOutputConfig.java | 4 ++--
.../library/shuffle/common/HttpConnection.java | 10 ++++++----
23 files changed, 54 insertions(+), 47 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/tez/blob/c90918c6/tez-api/src/main/java/org/apache/tez/common/security/DAGAccessControls.java
----------------------------------------------------------------------
diff --git a/tez-api/src/main/java/org/apache/tez/common/security/DAGAccessControls.java b/tez-api/src/main/java/org/apache/tez/common/security/DAGAccessControls.java
index 1b1e90e..87e08c3 100644
--- a/tez-api/src/main/java/org/apache/tez/common/security/DAGAccessControls.java
+++ b/tez-api/src/main/java/org/apache/tez/common/security/DAGAccessControls.java
@@ -85,10 +85,10 @@ public class DAGAccessControls {
}
/**
- * Set the list of users that view permissions on the DAG. If all users are allowed,
+ * Sets the list of users with view permissions on the DAG. If all users are allowed,
* pass in a single entry "*"
* @param users Set of users with view permissions
- * @return
+ * @return this object for further chained method calls
*/
public synchronized DAGAccessControls setUsersWithViewACLs(Collection<String> users) {
this.usersWithViewACLs.clear();
@@ -97,10 +97,10 @@ public class DAGAccessControls {
}
/**
- * Set the list of users that have modify permissions on the DAG. If all users are allowed,
+ * Sets the list of users with modify permissions on the DAG. If all users are allowed,
* pass in a single entry "*"
* @param users Set of users with modify permissions
- * @return
+ * @return this object for further chained method calls
*/
public synchronized DAGAccessControls setUsersWithModifyACLs(Collection<String> users) {
this.usersWithModifyACLs.clear();
@@ -109,9 +109,9 @@ public class DAGAccessControls {
}
/**
- * Set the list of groups that have view permissions on the DAG.
+ * Sets the list of groups with view permissions on the DAG.
* @param groups Set of groups with view permissions
- * @return
+ * @return this object for further chained method calls
*/
public synchronized DAGAccessControls setGroupsWithViewACLs(Collection<String> groups) {
this.groupsWithViewACLs.clear();
@@ -120,9 +120,9 @@ public class DAGAccessControls {
}
/**
- * Set the list of groups that have modify permissions on the DAG.
+ * Sets the list of groups with modify permissions on the DAG.
* @param groups Set of groups with modify permissions
- * @return
+ * @return this object for further chained method calls
*/
public synchronized DAGAccessControls setGroupsWithModifyACLs(Collection<String> groups) {
this.groupsWithModifyACLs.clear();
http://git-wip-us.apache.org/repos/asf/tez/blob/c90918c6/tez-api/src/main/java/org/apache/tez/dag/api/DataSourceDescriptor.java
----------------------------------------------------------------------
diff --git a/tez-api/src/main/java/org/apache/tez/dag/api/DataSourceDescriptor.java b/tez-api/src/main/java/org/apache/tez/dag/api/DataSourceDescriptor.java
index f8e6072..8d03705 100644
--- a/tez-api/src/main/java/org/apache/tez/dag/api/DataSourceDescriptor.java
+++ b/tez-api/src/main/java/org/apache/tez/dag/api/DataSourceDescriptor.java
@@ -161,7 +161,7 @@ public class DataSourceDescriptor {
/**
* Get the list of additional local files which were specified during creation.
- * @return
+ * @return Map of additional local files or null if there are none
*/
@InterfaceAudience.Private
public @Nullable Map<String, LocalResource> getAdditionalLocalFiles() {
http://git-wip-us.apache.org/repos/asf/tez/blob/c90918c6/tez-api/src/main/java/org/apache/tez/dag/api/Edge.java
----------------------------------------------------------------------
diff --git a/tez-api/src/main/java/org/apache/tez/dag/api/Edge.java b/tez-api/src/main/java/org/apache/tez/dag/api/Edge.java
index 151d0db..e348fa1 100644
--- a/tez-api/src/main/java/org/apache/tez/dag/api/Edge.java
+++ b/tez-api/src/main/java/org/apache/tez/dag/api/Edge.java
@@ -70,7 +70,7 @@ public class Edge {
/**
* The @link {@link Vertex} that reads output from the edge
- * @retur {@link Vertex}
+ * @return {@link Vertex}
*/
public Vertex getOutputVertex() {
return outputVertex;
http://git-wip-us.apache.org/repos/asf/tez/blob/c90918c6/tez-api/src/main/java/org/apache/tez/dag/api/EntityDescriptor.java
----------------------------------------------------------------------
diff --git a/tez-api/src/main/java/org/apache/tez/dag/api/EntityDescriptor.java b/tez-api/src/main/java/org/apache/tez/dag/api/EntityDescriptor.java
index 6c86207..1ad004f 100644
--- a/tez-api/src/main/java/org/apache/tez/dag/api/EntityDescriptor.java
+++ b/tez-api/src/main/java/org/apache/tez/dag/api/EntityDescriptor.java
@@ -60,7 +60,7 @@ public abstract class EntityDescriptor<T extends EntityDescriptor<T>> implements
/**
* Set user payload for this entity descriptor
* @param userPayload User Payload
- * @return
+ * @return this object for further chained method calls
*/
public T setUserPayload(UserPayload userPayload) {
Preconditions.checkNotNull(userPayload);
@@ -72,6 +72,7 @@ public abstract class EntityDescriptor<T extends EntityDescriptor<T>> implements
* Provide a human-readable version of the user payload that can be
* used in the History UI
* @param historyText History text
+ * @return this object for further chained method calls
*/
public T setHistoryText(String historyText) {
this.historyText = historyText;
http://git-wip-us.apache.org/repos/asf/tez/blob/c90918c6/tez-api/src/main/java/org/apache/tez/dag/api/PreWarmVertex.java
----------------------------------------------------------------------
diff --git a/tez-api/src/main/java/org/apache/tez/dag/api/PreWarmVertex.java b/tez-api/src/main/java/org/apache/tez/dag/api/PreWarmVertex.java
index 4ec553a..62979b5 100644
--- a/tez-api/src/main/java/org/apache/tez/dag/api/PreWarmVertex.java
+++ b/tez-api/src/main/java/org/apache/tez/dag/api/PreWarmVertex.java
@@ -69,7 +69,7 @@ public class PreWarmVertex extends Vertex {
* Create a config builder for the @link {@link PreWarmVertex}. This may be used to construct the
* pre-warm vertex more flexibly.
* @param conf
- * @return
+ * @return a new config builder for {@link PreWarmVertex}
*/
public static PreWarmVertexConfigBuilder createConfigBuilder(Configuration conf) {
return new PreWarmVertexConfigBuilder(conf);
http://git-wip-us.apache.org/repos/asf/tez/blob/c90918c6/tez-api/src/main/java/org/apache/tez/dag/api/Vertex.java
----------------------------------------------------------------------
diff --git a/tez-api/src/main/java/org/apache/tez/dag/api/Vertex.java b/tez-api/src/main/java/org/apache/tez/dag/api/Vertex.java
index 6397c82..1df548a 100644
--- a/tez-api/src/main/java/org/apache/tez/dag/api/Vertex.java
+++ b/tez-api/src/main/java/org/apache/tez/dag/api/Vertex.java
@@ -121,6 +121,7 @@ public class Vertex {
* @param taskResource
* Physical resources like memory/cpu thats used by each task of this
* vertex.
+ * @return a new Vertex with the given parameters
*/
public static Vertex create(String vertexName,
ProcessorDescriptor processorDescriptor,
@@ -145,6 +146,7 @@ public class Vertex {
* @param processorDescriptor
* Description of the processor that is executed in every task of
* this vertex
+ * @return a new Vertex with the given parameters
*/
public static Vertex create(String vertexName, ProcessorDescriptor processorDescriptor) {
return new Vertex(vertexName, processorDescriptor);
@@ -169,6 +171,7 @@ public class Vertex {
* Number of tasks in this vertex. Set to -1 if this is going to be
* decided at runtime. Parallelism may change at runtime due to graph
* reconfigurations.
+ * @return a new Vertex with the given parameters
*/
public static Vertex create(String vertexName, ProcessorDescriptor processorDescriptor,
int parallelism) {
@@ -186,7 +189,7 @@ public class Vertex {
/**
* Get the vertex task processor descriptor
- * @return
+ * @return process descriptor
*/
public ProcessorDescriptor getProcessorDescriptor() {
return this.processorDescriptor;
http://git-wip-us.apache.org/repos/asf/tez/blob/c90918c6/tez-api/src/main/java/org/apache/tez/dag/api/VertexGroup.java
----------------------------------------------------------------------
diff --git a/tez-api/src/main/java/org/apache/tez/dag/api/VertexGroup.java b/tez-api/src/main/java/org/apache/tez/dag/api/VertexGroup.java
index 0594835..f1887a1 100644
--- a/tez-api/src/main/java/org/apache/tez/dag/api/VertexGroup.java
+++ b/tez-api/src/main/java/org/apache/tez/dag/api/VertexGroup.java
@@ -89,7 +89,8 @@ public class VertexGroup {
/**
* Add an common data sink to the group of vertices.
- * Refer to {@link Vertex#addDataSink(String, OutputDescriptor, OutputCommitterDescriptor)}
+ * Refer to {@link Vertex#addDataSink(String, DataSinkDescriptor)}
+ * @return this object for further chained method calls
*/
public VertexGroup addDataSink(String outputName, DataSinkDescriptor dataSinkDescriptor) {
RootInputLeafOutput<OutputDescriptor, OutputCommitterDescriptor> leafOutput =
http://git-wip-us.apache.org/repos/asf/tez/blob/c90918c6/tez-api/src/main/java/org/apache/tez/runtime/api/InputSpecUpdate.java
----------------------------------------------------------------------
diff --git a/tez-api/src/main/java/org/apache/tez/runtime/api/InputSpecUpdate.java b/tez-api/src/main/java/org/apache/tez/runtime/api/InputSpecUpdate.java
index f0d166f..b81f15c 100644
--- a/tez-api/src/main/java/org/apache/tez/runtime/api/InputSpecUpdate.java
+++ b/tez-api/src/main/java/org/apache/tez/runtime/api/InputSpecUpdate.java
@@ -48,7 +48,7 @@ public class InputSpecUpdate {
*
* @param numPhysicalInputs
* the number of physical inputs for all work units which will use the LogicalInput
- * @return
+ * @return a new InputSpecUpdate instance given the parameters
*/
public static InputSpecUpdate createAllTaskInputSpecUpdate(int numPhysicalInputs) {
return new InputSpecUpdate(numPhysicalInputs);
@@ -61,7 +61,7 @@ public class InputSpecUpdate {
* @param perWorkUnitNumPhysicalInputs
* A list containing one entry per work unit. The order in the list corresponds to task
* index or equivalently the order of {@link InputDataInformationEvent}s being sent.
- * @return
+ * @return a new InputSpecUpdate instance given the parameters
*/
public static InputSpecUpdate createPerTaskInputSpecUpdate(
List<Integer> perWorkUnitNumPhysicalInputs) {
http://git-wip-us.apache.org/repos/asf/tez/blob/c90918c6/tez-api/src/main/java/org/apache/tez/runtime/api/ProcessorFrameworkInterface.java
----------------------------------------------------------------------
diff --git a/tez-api/src/main/java/org/apache/tez/runtime/api/ProcessorFrameworkInterface.java b/tez-api/src/main/java/org/apache/tez/runtime/api/ProcessorFrameworkInterface.java
index 317ba39..f0ba9c9 100644
--- a/tez-api/src/main/java/org/apache/tez/runtime/api/ProcessorFrameworkInterface.java
+++ b/tez-api/src/main/java/org/apache/tez/runtime/api/ProcessorFrameworkInterface.java
@@ -36,7 +36,6 @@ public interface ProcessorFrameworkInterface {
/**
* Initializes the <code>Processor</code>
*
- * @param processorContext
* @throws java.io.IOException
* if an error occurs
*/
http://git-wip-us.apache.org/repos/asf/tez/blob/c90918c6/tez-dag/src/main/java/org/apache/tez/dag/history/SummaryEvent.java
----------------------------------------------------------------------
diff --git a/tez-dag/src/main/java/org/apache/tez/dag/history/SummaryEvent.java b/tez-dag/src/main/java/org/apache/tez/dag/history/SummaryEvent.java
index eaae813..bdbea59 100644
--- a/tez-dag/src/main/java/org/apache/tez/dag/history/SummaryEvent.java
+++ b/tez-dag/src/main/java/org/apache/tez/dag/history/SummaryEvent.java
@@ -32,7 +32,7 @@ public interface SummaryEvent {
/**
* Whether to write this event immediately to the DAG recovery file
* Summary events are always written immediately to summary file.
- * @return
+ * @return true if event should be written immediately to the DAG recovery file
*/
public boolean writeToRecoveryImmediately();
http://git-wip-us.apache.org/repos/asf/tez/blob/c90918c6/tez-mapreduce/src/main/java/org/apache/tez/mapreduce/hadoop/InputSplitInfo.java
----------------------------------------------------------------------
diff --git a/tez-mapreduce/src/main/java/org/apache/tez/mapreduce/hadoop/InputSplitInfo.java b/tez-mapreduce/src/main/java/org/apache/tez/mapreduce/hadoop/InputSplitInfo.java
index 15fe463..cb362ac 100644
--- a/tez-mapreduce/src/main/java/org/apache/tez/mapreduce/hadoop/InputSplitInfo.java
+++ b/tez-mapreduce/src/main/java/org/apache/tez/mapreduce/hadoop/InputSplitInfo.java
@@ -88,13 +88,13 @@ public interface InputSplitInfo {
/**
* Get new format splits. Should only be used if the mapreduce API is being used
- * @return
+ * @return array of mapreduce format splits
*/
public org.apache.hadoop.mapreduce.InputSplit[] getNewFormatSplits();
/**
* Get old format splits. Should only be used if the mapred API is being used
- * @return
+ * @return array of mapred format splits
*/
public org.apache.hadoop.mapred.InputSplit[] getOldFormatSplits();
}
http://git-wip-us.apache.org/repos/asf/tez/blob/c90918c6/tez-mapreduce/src/main/java/org/apache/tez/mapreduce/hadoop/MRInputHelpers.java
----------------------------------------------------------------------
diff --git a/tez-mapreduce/src/main/java/org/apache/tez/mapreduce/hadoop/MRInputHelpers.java b/tez-mapreduce/src/main/java/org/apache/tez/mapreduce/hadoop/MRInputHelpers.java
index 71d294f..9d6d56a 100644
--- a/tez-mapreduce/src/main/java/org/apache/tez/mapreduce/hadoop/MRInputHelpers.java
+++ b/tez-mapreduce/src/main/java/org/apache/tez/mapreduce/hadoop/MRInputHelpers.java
@@ -253,7 +253,7 @@ public class MRInputHelpers {
}
/**
- * Generates Input splits and stores them in a {@link org.apache.hadoop.mapreduce.v2.proto.MRProtos} instance.
+ * Generates Input splits and stores them in a {@link org.apache.tez.mapreduce.protos.MRRuntimeProtos.MRSplitsProto} instance.
*
* Returns an instance of {@link InputSplitInfoMem}
*
http://git-wip-us.apache.org/repos/asf/tez/blob/c90918c6/tez-runtime-library/src/main/java/org/apache/tez/runtime/library/common/sort/impl/IFile.java
----------------------------------------------------------------------
diff --git a/tez-runtime-library/src/main/java/org/apache/tez/runtime/library/common/sort/impl/IFile.java b/tez-runtime-library/src/main/java/org/apache/tez/runtime/library/common/sort/impl/IFile.java
index c9b6df4..f96f9ed 100644
--- a/tez-runtime-library/src/main/java/org/apache/tez/runtime/library/common/sort/impl/IFile.java
+++ b/tez-runtime-library/src/main/java/org/apache/tez/runtime/library/common/sort/impl/IFile.java
@@ -642,7 +642,7 @@ public class IFile {
}
/**
- * Read upto len bytes into buf starting at offset off.
+ * Read up to len bytes into buf starting at offset off.
*
* @param buf buffer
* @param off offset
@@ -687,7 +687,8 @@ public class IFile {
* Reset key length and value length for next record in the file
*
* @param dIn
- * @return
+ * @return true if key length and value length were set to the next
+ * false if end of file (EOF) marker was reached
* @throws IOException
*/
protected boolean positionToNextRecord(DataInput dIn) throws IOException {
http://git-wip-us.apache.org/repos/asf/tez/blob/c90918c6/tez-runtime-library/src/main/java/org/apache/tez/runtime/library/conf/BaseConfigBuilder.java
----------------------------------------------------------------------
diff --git a/tez-runtime-library/src/main/java/org/apache/tez/runtime/library/conf/BaseConfigBuilder.java b/tez-runtime-library/src/main/java/org/apache/tez/runtime/library/conf/BaseConfigBuilder.java
index c172804..d991e89 100644
--- a/tez-runtime-library/src/main/java/org/apache/tez/runtime/library/conf/BaseConfigBuilder.java
+++ b/tez-runtime-library/src/main/java/org/apache/tez/runtime/library/conf/BaseConfigBuilder.java
@@ -34,7 +34,7 @@ interface BaseConfigBuilder<T> {
*
* @param key the key to set
* @param value the corresponding value
- * @return
+ * @return this object for further chained method calls
*/
public T setAdditionalConfiguration(String key, String value);
@@ -45,7 +45,7 @@ interface BaseConfigBuilder<T> {
* made available to the combiner.
*
* @param confMap map of configuration key-value pairs
- * @return
+ * @return this object for further chained method calls
*/
public T setAdditionalConfiguration(Map<String, String> confMap);
@@ -55,7 +55,7 @@ interface BaseConfigBuilder<T> {
* compatibility and ease of use for existing systems which rely heavily on Configuration.
*
* @param conf
- * @return
+ * @return this object for further chained method calls
*/
@InterfaceAudience.LimitedPrivate({"hive, pig"})
public T setFromConfiguration(Configuration conf);
http://git-wip-us.apache.org/repos/asf/tez/blob/c90918c6/tez-runtime-library/src/main/java/org/apache/tez/runtime/library/conf/OrderedGroupedKVInputConfig.java
----------------------------------------------------------------------
diff --git a/tez-runtime-library/src/main/java/org/apache/tez/runtime/library/conf/OrderedGroupedKVInputConfig.java b/tez-runtime-library/src/main/java/org/apache/tez/runtime/library/conf/OrderedGroupedKVInputConfig.java
index 27817ef..9a81f2e 100644
--- a/tez-runtime-library/src/main/java/org/apache/tez/runtime/library/conf/OrderedGroupedKVInputConfig.java
+++ b/tez-runtime-library/src/main/java/org/apache/tez/runtime/library/conf/OrderedGroupedKVInputConfig.java
@@ -93,7 +93,7 @@ public class OrderedGroupedKVInputConfig {
* Enable the memory to memory merger
*
* @param enable whether to enable the memory to memory merger
- * @return
+ * @return instance of the current builder
*/
public T setMemToMemMerger(boolean enable); // Not super useful until additional params are used.
@@ -460,7 +460,7 @@ public class OrderedGroupedKVInputConfig {
* @param serializerConf the serializer configuration. This can be null, and is a
* {@link java.util.Map} of key-value pairs. The keys should be limited
* to the ones required by the comparator.
- * @return
+ * @return this object for further chained method calls
*/
public Builder setKeySerializationClass(String serializationClassName,
String comparatorClassName, @Nullable Map<String, String> serializerConf) {
@@ -486,7 +486,7 @@ public class OrderedGroupedKVInputConfig {
* @param serializerConf the serializer configuration. This can be null, and is a
* {@link java.util.Map} of key-value pairs. The keys should be limited
* to the ones required by the comparator.
- * @return
+ * @return this object for further chained method calls
*/
public Builder setValueSerializationClass(String serializationClassName,
@Nullable Map<String, String> serializerConf) {
http://git-wip-us.apache.org/repos/asf/tez/blob/c90918c6/tez-runtime-library/src/main/java/org/apache/tez/runtime/library/conf/OrderedPartitionedKVEdgeConfig.java
----------------------------------------------------------------------
diff --git a/tez-runtime-library/src/main/java/org/apache/tez/runtime/library/conf/OrderedPartitionedKVEdgeConfig.java b/tez-runtime-library/src/main/java/org/apache/tez/runtime/library/conf/OrderedPartitionedKVEdgeConfig.java
index c495c12..76f73ac 100644
--- a/tez-runtime-library/src/main/java/org/apache/tez/runtime/library/conf/OrderedPartitionedKVEdgeConfig.java
+++ b/tez-runtime-library/src/main/java/org/apache/tez/runtime/library/conf/OrderedPartitionedKVEdgeConfig.java
@@ -210,7 +210,7 @@ public class OrderedPartitionedKVEdgeConfig
* @param serializerConf the serializer configuration. This can be null, and is a
* {@link java.util.Map} of key-value pairs. The keys should be limited
* to the ones required by the comparator.
- * @return
+ * @return this object for further chained method calls
*/
public Builder setKeySerializationClass(String serializationClassName,
String comparatorClassName, @Nullable Map<String, String> serializerConf) {
@@ -226,7 +226,7 @@ public class OrderedPartitionedKVEdgeConfig
* @param serializerConf the serializer configuration. This can be null, and is a
* {@link java.util.Map} of key-value pairs. The keys should be limited
* to the ones required by the comparator.
- * @return
+ * @return this object for further chained method calls
*/
public Builder setValueSerializationClass(String serializationClassName,
@Nullable Map<String, String> serializerConf) {
http://git-wip-us.apache.org/repos/asf/tez/blob/c90918c6/tez-runtime-library/src/main/java/org/apache/tez/runtime/library/conf/OrderedPartitionedKVOutputConfig.java
----------------------------------------------------------------------
diff --git a/tez-runtime-library/src/main/java/org/apache/tez/runtime/library/conf/OrderedPartitionedKVOutputConfig.java b/tez-runtime-library/src/main/java/org/apache/tez/runtime/library/conf/OrderedPartitionedKVOutputConfig.java
index f9b468c..4a8e38f 100644
--- a/tez-runtime-library/src/main/java/org/apache/tez/runtime/library/conf/OrderedPartitionedKVOutputConfig.java
+++ b/tez-runtime-library/src/main/java/org/apache/tez/runtime/library/conf/OrderedPartitionedKVOutputConfig.java
@@ -386,7 +386,7 @@ public class OrderedPartitionedKVOutputConfig {
* @param serializerConf the serializer configuration. This can be null, and is a
* {@link java.util.Map} of key-value pairs. The keys should be limited
* to the ones required by the comparator.
- * @return
+ * @return this object for further chained method calls
*/
public Builder setKeySerializationClass(String serializationClassName,
String comparatorClassName, @Nullable Map<String, String> serializerConf) {
@@ -412,7 +412,7 @@ public class OrderedPartitionedKVOutputConfig {
* @param serializerConf the serializer configuration. This can be null, and is a
* {@link java.util.Map} of key-value pairs. The keys should be limited
* to the ones required by the comparator.
- * @return
+ * @return this object for further chained method calls
*/
public Builder setValueSerializationClass(String serializationClassName,
@Nullable Map<String, String> serializerConf) {
http://git-wip-us.apache.org/repos/asf/tez/blob/c90918c6/tez-runtime-library/src/main/java/org/apache/tez/runtime/library/conf/UnorderedKVEdgeConfig.java
----------------------------------------------------------------------
diff --git a/tez-runtime-library/src/main/java/org/apache/tez/runtime/library/conf/UnorderedKVEdgeConfig.java b/tez-runtime-library/src/main/java/org/apache/tez/runtime/library/conf/UnorderedKVEdgeConfig.java
index 66b2ded..3864565 100644
--- a/tez-runtime-library/src/main/java/org/apache/tez/runtime/library/conf/UnorderedKVEdgeConfig.java
+++ b/tez-runtime-library/src/main/java/org/apache/tez/runtime/library/conf/UnorderedKVEdgeConfig.java
@@ -207,7 +207,7 @@ public class UnorderedKVEdgeConfig extends HadoopKeyValuesBasedBaseEdgeConfig {
* @param serializerConf the serializer configuration. This can be null, and is a
* {@link java.util.Map} of key-value pairs. The keys should be limited
* to the ones required by the comparator.
- * @return
+ * @return this object for further chained method calls
*/
public Builder setKeySerializationClass(String serializationClassName,
@Nullable Map<String, String> serializerConf) {
@@ -223,7 +223,7 @@ public class UnorderedKVEdgeConfig extends HadoopKeyValuesBasedBaseEdgeConfig {
* @param serializerConf the serializer configuration. This can be null, and is a
* {@link java.util.Map} of key-value pairs. The keys should be limited
* to the ones required by the comparator.
- * @return
+ * @return this object for further chained method calls
*/
public Builder setValueSerializationClass(String serializationClassName,
@Nullable Map<String, String> serializerConf) {
http://git-wip-us.apache.org/repos/asf/tez/blob/c90918c6/tez-runtime-library/src/main/java/org/apache/tez/runtime/library/conf/UnorderedKVInputConfig.java
----------------------------------------------------------------------
diff --git a/tez-runtime-library/src/main/java/org/apache/tez/runtime/library/conf/UnorderedKVInputConfig.java b/tez-runtime-library/src/main/java/org/apache/tez/runtime/library/conf/UnorderedKVInputConfig.java
index 9686cbe..e541341 100644
--- a/tez-runtime-library/src/main/java/org/apache/tez/runtime/library/conf/UnorderedKVInputConfig.java
+++ b/tez-runtime-library/src/main/java/org/apache/tez/runtime/library/conf/UnorderedKVInputConfig.java
@@ -307,7 +307,7 @@ public class UnorderedKVInputConfig {
* @param serializerConf the serializer configuration. This can be null, and is a
* {@link java.util.Map} of key-value pairs. The keys should be limited
* to the ones required by the comparator.
- * @return
+ * @return this object for further chained method calls
*/
public Builder setKeySerializationClass(String serializationClassName,
@Nullable Map<String, String> serializerConf) {
@@ -330,7 +330,7 @@ public class UnorderedKVInputConfig {
* @param serializerConf the serializer configuration. This can be null, and is a
* {@link java.util.Map} of key-value pairs. The keys should be limited
* to the ones required by the comparator.
- * @return
+ * @return this object for further chained method calls
*/
public Builder setValueSerializationClass(String serializationClassName,
@Nullable Map<String, String> serializerConf) {
http://git-wip-us.apache.org/repos/asf/tez/blob/c90918c6/tez-runtime-library/src/main/java/org/apache/tez/runtime/library/conf/UnorderedKVOutputConfig.java
----------------------------------------------------------------------
diff --git a/tez-runtime-library/src/main/java/org/apache/tez/runtime/library/conf/UnorderedKVOutputConfig.java b/tez-runtime-library/src/main/java/org/apache/tez/runtime/library/conf/UnorderedKVOutputConfig.java
index b4e2c2c..3c3a7cf 100644
--- a/tez-runtime-library/src/main/java/org/apache/tez/runtime/library/conf/UnorderedKVOutputConfig.java
+++ b/tez-runtime-library/src/main/java/org/apache/tez/runtime/library/conf/UnorderedKVOutputConfig.java
@@ -217,7 +217,7 @@ public class UnorderedKVOutputConfig {
* @param serializerConf the serializer configuration. This can be null, and is a
* {@link java.util.Map} of key-value pairs. The keys should be limited
* to the ones required by the comparator.
- * @return
+ * @return this object for further chained method calls
*/
public Builder setKeySerializationClass(String serializationClassName,
@Nullable Map<String, String> serializerConf) {
@@ -240,7 +240,7 @@ public class UnorderedKVOutputConfig {
* @param serializerConf the serializer configuration. This can be null, and is a
* {@link java.util.Map} of key-value pairs. The keys should be limited
* to the ones required by the comparator.
- * @return
+ * @return this object for further chained method calls
*/
public Builder setValueSerializationClass(String serializationClassName,
@Nullable Map<String, String> serializerConf) {
http://git-wip-us.apache.org/repos/asf/tez/blob/c90918c6/tez-runtime-library/src/main/java/org/apache/tez/runtime/library/conf/UnorderedPartitionedKVEdgeConfig.java
----------------------------------------------------------------------
diff --git a/tez-runtime-library/src/main/java/org/apache/tez/runtime/library/conf/UnorderedPartitionedKVEdgeConfig.java b/tez-runtime-library/src/main/java/org/apache/tez/runtime/library/conf/UnorderedPartitionedKVEdgeConfig.java
index 54ea5ce..44ca74e 100644
--- a/tez-runtime-library/src/main/java/org/apache/tez/runtime/library/conf/UnorderedPartitionedKVEdgeConfig.java
+++ b/tez-runtime-library/src/main/java/org/apache/tez/runtime/library/conf/UnorderedPartitionedKVEdgeConfig.java
@@ -212,7 +212,7 @@ public class UnorderedPartitionedKVEdgeConfig
* @param serializerConf the serializer configuration. This can be null, and is a
* {@link java.util.Map} of key-value pairs. The keys should be limited
* to the ones required by the comparator.
- * @return
+ * @return this object for further chained method calls
*/
public Builder setKeySerializationClass(String serializationClassName,
@Nullable Map<String, String> serializerConf) {
@@ -228,7 +228,7 @@ public class UnorderedPartitionedKVEdgeConfig
* @param serializerConf the serializer configuration. This can be null, and is a
* {@link java.util.Map} of key-value pairs. The keys should be limited
* to the ones required by the comparator.
- * @return
+ * @return this object for further chained method calls
*/
public Builder setValueSerializationClass(String serializationClassName, @Nullable Map<String, String> serializerConf) {
outputBuilder.setValueSerializationClass(serializationClassName, serializerConf);
http://git-wip-us.apache.org/repos/asf/tez/blob/c90918c6/tez-runtime-library/src/main/java/org/apache/tez/runtime/library/conf/UnorderedPartitionedKVOutputConfig.java
----------------------------------------------------------------------
diff --git a/tez-runtime-library/src/main/java/org/apache/tez/runtime/library/conf/UnorderedPartitionedKVOutputConfig.java b/tez-runtime-library/src/main/java/org/apache/tez/runtime/library/conf/UnorderedPartitionedKVOutputConfig.java
index fbadc58..74fb209 100644
--- a/tez-runtime-library/src/main/java/org/apache/tez/runtime/library/conf/UnorderedPartitionedKVOutputConfig.java
+++ b/tez-runtime-library/src/main/java/org/apache/tez/runtime/library/conf/UnorderedPartitionedKVOutputConfig.java
@@ -277,7 +277,7 @@ public class UnorderedPartitionedKVOutputConfig {
* @param serializerConf the serializer configuration. This can be null, and is a
* {@link java.util.Map} of key-value pairs. The keys should be limited
* to the ones required by the comparator.
- * @return
+ * @return this object for further chained method calls
*/
public Builder setKeySerializationClass(String serializationClassName,
@Nullable Map<String, String> serializerConf) {
@@ -299,7 +299,7 @@ public class UnorderedPartitionedKVOutputConfig {
* @param serializerConf the serializer configuration. This can be null, and is a
* {@link java.util.Map} of key-value pairs. The keys should be limited
* to the ones required by the comparator.
- * @return
+ * @return this object for further chained method calls
*/
public Builder setValueSerializationClass(String serializationClassName,
@Nullable Map<String, String> serializerConf) {
http://git-wip-us.apache.org/repos/asf/tez/blob/c90918c6/tez-runtime-library/src/main/java/org/apache/tez/runtime/library/shuffle/common/HttpConnection.java
----------------------------------------------------------------------
diff --git a/tez-runtime-library/src/main/java/org/apache/tez/runtime/library/shuffle/common/HttpConnection.java b/tez-runtime-library/src/main/java/org/apache/tez/runtime/library/shuffle/common/HttpConnection.java
index 1dd36b4..7208c16 100644
--- a/tez-runtime-library/src/main/java/org/apache/tez/runtime/library/shuffle/common/HttpConnection.java
+++ b/tez-runtime-library/src/main/java/org/apache/tez/runtime/library/shuffle/common/HttpConnection.java
@@ -122,8 +122,9 @@ public class HttpConnection {
/**
* Connect to source
*
- * @return
- * @throws IOException
+ * @return true if connection was successful
+ * false if connection was previously cleaned up
+ * @throws IOException upon connection failure
*/
public boolean connect() throws IOException {
return connect(httpConnParams.connectionTimeout);
@@ -133,8 +134,9 @@ public class HttpConnection {
* Connect to source with specific timeout
*
* @param connectionTimeout
- * @return
- * @throws IOException
+ * @return true if connection was successful
+ * false if connection was previously cleaned up
+ * @throws IOException upon connection failure
*/
public boolean connect(int connectionTimeout) throws IOException {
stopWatch.reset().start();
[4/4] git commit: Updating CHANGES.txt etc
Posted by bi...@apache.org.
Updating CHANGES.txt etc
Project: http://git-wip-us.apache.org/repos/asf/tez/repo
Commit: http://git-wip-us.apache.org/repos/asf/tez/commit/d0452969
Tree: http://git-wip-us.apache.org/repos/asf/tez/tree/d0452969
Diff: http://git-wip-us.apache.org/repos/asf/tez/diff/d0452969
Branch: refs/heads/branch-0.5.0
Commit: d045296963280038f05ad44d6b81fa37a3438ae2
Parents: c90918c
Author: Bikas Saha <bi...@apache.org>
Authored: Thu Aug 21 13:54:14 2014 -0700
Committer: Bikas Saha <bi...@apache.org>
Committed: Thu Aug 21 13:54:14 2014 -0700
----------------------------------------------------------------------
CHANGES.txt | 6 ++++++
INSTALL.txt | 8 ++++----
2 files changed, 10 insertions(+), 4 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/tez/blob/d0452969/CHANGES.txt
----------------------------------------------------------------------
diff --git a/CHANGES.txt b/CHANGES.txt
index 729d5d4..184f353 100644
--- a/CHANGES.txt
+++ b/CHANGES.txt
@@ -4,6 +4,9 @@ Apache Tez Change Log
Release 0.5.0: 2014-08-22
INCOMPATIBLE CHANGES
+ TEZ-1472. Separate method calls for creating InputDataInformationEvent with
+ serialized/unserialized payloads
+ TEZ-1469. AM/Session LRs are not shipped to vertices in new API use-case
TEZ-960. VertexManagerPluginContext::getTotalAVailableResource() changed to
VertexManagerPluginContext::getTotalAvailableResource()
TEZ-1025. Rename tez.am.max.task.attempts to tez.am.task.max.failed.attempts
@@ -85,6 +88,9 @@ INCOMPATIBLE CHANGES
ALL CHANGES
+ TEZ-1436. Fix javadoc warnings (Jonathan Eagles via bikas)
+ TEZ-1472. Separate method calls for creating InputDataInformationEvent with serialized/unserialized payloads (Siddharth Seth via bikas)
+ TEZ-1469. AM/Session LRs are not shipped to vertices in new API use-case (bikas)
TEZ-1449. Change user payloads to work with a byte buffer (Siddharth Seth via bikas)
TEZ-1325. RecoveryParser can find incorrect last DAG ID. (Jeff Zhang via hitesh)
TEZ-1466. Fix JDK8 builds of Tez (gopalv)
http://git-wip-us.apache.org/repos/asf/tez/blob/d0452969/INSTALL.txt
----------------------------------------------------------------------
diff --git a/INSTALL.txt b/INSTALL.txt
index 17e7413..25d87c9 100644
--- a/INSTALL.txt
+++ b/INSTALL.txt
@@ -16,8 +16,8 @@ Install/Deploy Instructions
3) Copy the relevant tez tarball into HDFS, and configure tez-site.xml
- A tez tarball containing tez and hadoop libraries will be found at tez-dist/target/tez-0.5.0.tar.gz
- Assuming that the tez jars are put in /apps/ on HDFS, the command would be
- "hadoop dfs -mkdir /apps/tez-0.5.0"
- "hadoop dfs -copyFromLocal tez-dist/target/tez-0.5.0.tar.gz /apps/tez-0.5.0/"
+ "hadoop fs -mkdir /apps/tez-0.5.0"
+ "hadoop fs -copyFromLocal tez-dist/target/tez-0.5.0.tar.gz /apps/tez-0.5.0/"
- tez-site.xml configuration
- Set tez.lib.uris to point to the tar.gz uploaded to HDFS. Assuming the steps mentioned so far were followed,
set tez.lib.uris to "${fs.defaultFS}/apps/tez-0.5.0/tez-0.5.0.tar.gz"
@@ -71,8 +71,8 @@ follow this alternate machanism to setup Tez to use Hadoop libraries from the cl
Step 3 above changes as follows. Also subsequent steps would use tez-dist/target/tez-0.5.0-minimal.tar.gz instead of tez-dist/target/tez-0.5.0.tar.gz
- A tez build without Hadoop dependencies will be available at tez-dist/target/tez-0.5.0-minimal.tar.gz
- Assuming that the tez jars are put in /apps/ on HDFS, the command would be
- "hadoop dfs -mkdir /apps/tez-0.5.0"
- "hadoop dfs -copyFromLocal tez-dist/target/tez-0.5.0-minimal.tar.gz /apps/tez-0.5.0"
+ "hadoop fs -mkdir /apps/tez-0.5.0"
+ "hadoop fs -copyFromLocal tez-dist/target/tez-0.5.0-minimal.tar.gz /apps/tez-0.5.0"
- tez-site.xml configuration
- Set tez.lib.uris to point to the paths in HDFS containing the tez jars. Assuming the steps mentioned so far were followed,
set tez.lib.uris to "${fs.defaultFS}/apps/tez-0.5.0/tez-0.5.0-minimal.tar.gz