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/02/25 19:11:04 UTC
git commit: TEZ-865. TezTaskContext.getDAGName() does not return DAG
name (Tsuyoshi OZAWA via bikas)
Repository: incubator-tez
Updated Branches:
refs/heads/master b38c9313e -> f38b33090
TEZ-865. TezTaskContext.getDAGName() does not return DAG name (Tsuyoshi OZAWA via bikas)
Project: http://git-wip-us.apache.org/repos/asf/incubator-tez/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-tez/commit/f38b3309
Tree: http://git-wip-us.apache.org/repos/asf/incubator-tez/tree/f38b3309
Diff: http://git-wip-us.apache.org/repos/asf/incubator-tez/diff/f38b3309
Branch: refs/heads/master
Commit: f38b330900a71f3aa643d4ff5e3975fc6c58bd37
Parents: b38c931
Author: Bikas Saha <bi...@apache.org>
Authored: Tue Feb 25 10:10:56 2014 -0800
Committer: Bikas Saha <bi...@apache.org>
Committed: Tue Feb 25 10:10:56 2014 -0800
----------------------------------------------------------------------
.../apache/tez/dag/app/dag/impl/TaskAttemptImpl.java | 4 +++-
.../apache/tez/dag/app/rm/TestContainerReuse.java | 2 +-
.../org/apache/tez/mapreduce/processor/MapUtils.java | 4 ++--
.../mapreduce/processor/map/TestMapProcessor.java | 3 ++-
.../processor/reduce/TestReduceProcessor.java | 5 +++--
.../tez/runtime/LogicalIOProcessorRuntimeTask.java | 10 +++++++---
.../org/apache/tez/runtime/api/impl/TaskSpec.java | 15 +++++++++++++--
.../tez/runtime/api/impl/TezInputContextImpl.java | 4 ++--
.../tez/runtime/api/impl/TezOutputContextImpl.java | 5 +++--
.../runtime/api/impl/TezProcessorContextImpl.java | 4 ++--
.../tez/runtime/api/impl/TezTaskContextImpl.java | 8 ++++----
.../runtime/TestLogicalIOProcessorRuntimeTask.java | 10 ++++++----
.../library/output/TestOnFileUnorderedKVOutput.java | 3 ++-
13 files changed, 50 insertions(+), 27 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-tez/blob/f38b3309/tez-dag/src/main/java/org/apache/tez/dag/app/dag/impl/TaskAttemptImpl.java
----------------------------------------------------------------------
diff --git a/tez-dag/src/main/java/org/apache/tez/dag/app/dag/impl/TaskAttemptImpl.java b/tez-dag/src/main/java/org/apache/tez/dag/app/dag/impl/TaskAttemptImpl.java
index d555e04..7fe07af 100644
--- a/tez-dag/src/main/java/org/apache/tez/dag/app/dag/impl/TaskAttemptImpl.java
+++ b/tez-dag/src/main/java/org/apache/tez/dag/app/dag/impl/TaskAttemptImpl.java
@@ -515,7 +515,9 @@ public class TaskAttemptImpl implements TaskAttempt,
Vertex vertex = getVertex();
ProcessorDescriptor procDesc = vertex.getProcessorDescriptor();
int taskId = getTaskID().getId();
- return new TaskSpec(getID(), vertex.getName(), procDesc,
+ return new TaskSpec(getID(),
+ vertex.getDAG().getName(),
+ vertex.getName(), procDesc,
vertex.getInputSpecList(taskId), vertex.getOutputSpecList(taskId),
vertex.getGroupInputSpecList(taskId));
}
http://git-wip-us.apache.org/repos/asf/incubator-tez/blob/f38b3309/tez-dag/src/test/java/org/apache/tez/dag/app/rm/TestContainerReuse.java
----------------------------------------------------------------------
diff --git a/tez-dag/src/test/java/org/apache/tez/dag/app/rm/TestContainerReuse.java b/tez-dag/src/test/java/org/apache/tez/dag/app/rm/TestContainerReuse.java
index 61d60ae..4c8ef1d 100644
--- a/tez-dag/src/test/java/org/apache/tez/dag/app/rm/TestContainerReuse.java
+++ b/tez-dag/src/test/java/org/apache/tez/dag/app/rm/TestContainerReuse.java
@@ -895,7 +895,7 @@ public class TestContainerReuse {
String[] hosts, String[] racks, Priority priority,
ContainerContext containerContext) {
AMSchedulerEventTALaunchRequest lr = new AMSchedulerEventTALaunchRequest(
- taID, capability, new TaskSpec(taID, "vertexName",
+ taID, capability, new TaskSpec(taID, "dagName", "vertexName",
new ProcessorDescriptor("processorClassName"),
Collections.singletonList(new InputSpec("vertexName",
new InputDescriptor("inputClassName"), 1)),
http://git-wip-us.apache.org/repos/asf/incubator-tez/blob/f38b3309/tez-mapreduce/src/test/java/org/apache/tez/mapreduce/processor/MapUtils.java
----------------------------------------------------------------------
diff --git a/tez-mapreduce/src/test/java/org/apache/tez/mapreduce/processor/MapUtils.java b/tez-mapreduce/src/test/java/org/apache/tez/mapreduce/processor/MapUtils.java
index 700acb0..df7f2a9 100644
--- a/tez-mapreduce/src/test/java/org/apache/tez/mapreduce/processor/MapUtils.java
+++ b/tez-mapreduce/src/test/java/org/apache/tez/mapreduce/processor/MapUtils.java
@@ -193,7 +193,7 @@ public class MapUtils {
public static LogicalIOProcessorRuntimeTask createLogicalTask(FileSystem fs, Path workDir,
JobConf jobConf, int mapId, Path mapInput,
- TezUmbilical umbilical,
+ TezUmbilical umbilical, String dagName,
String vertexName, List<InputSpec> inputSpecs,
List<OutputSpec> outputSpecs) throws Exception {
jobConf.setInputFormat(SequenceFileInputFormat.class);
@@ -205,7 +205,7 @@ public class MapUtils {
TaskSpec taskSpec = new TaskSpec(
TezTestUtils.getMockTaskAttemptId(0, 0, mapId, 0),
- vertexName,
+ dagName, vertexName,
mapProcessorDesc,
inputSpecs,
outputSpecs, null);
http://git-wip-us.apache.org/repos/asf/incubator-tez/blob/f38b3309/tez-mapreduce/src/test/java/org/apache/tez/mapreduce/processor/map/TestMapProcessor.java
----------------------------------------------------------------------
diff --git a/tez-mapreduce/src/test/java/org/apache/tez/mapreduce/processor/map/TestMapProcessor.java b/tez-mapreduce/src/test/java/org/apache/tez/mapreduce/processor/map/TestMapProcessor.java
index 78191a4..744d680 100644
--- a/tez-mapreduce/src/test/java/org/apache/tez/mapreduce/processor/map/TestMapProcessor.java
+++ b/tez-mapreduce/src/test/java/org/apache/tez/mapreduce/processor/map/TestMapProcessor.java
@@ -101,6 +101,7 @@ public class TestMapProcessor {
@Test
public void testMapProcessor() throws Exception {
+ String dagName = "mrdag0";
String vertexName = MultiStageMRConfigUtil.getInitialMapVertexName();
JobConf jobConf = new JobConf(defaultConf);
setUpJobConf(jobConf);
@@ -129,7 +130,7 @@ public class TestMapProcessor {
OutputSpec mapOutputSpec = new OutputSpec("NullDestVertex", new OutputDescriptor(LocalOnFileSorterOutput.class.getName()), 1);
LogicalIOProcessorRuntimeTask task = MapUtils.createLogicalTask(localFs, workDir, job, 0,
- new Path(workDir, "map0"), new TestUmbilical(), vertexName,
+ new Path(workDir, "map0"), new TestUmbilical(), dagName, vertexName,
Collections.singletonList(mapInputSpec),
Collections.singletonList(mapOutputSpec));
http://git-wip-us.apache.org/repos/asf/incubator-tez/blob/f38b3309/tez-mapreduce/src/test/java/org/apache/tez/mapreduce/processor/reduce/TestReduceProcessor.java
----------------------------------------------------------------------
diff --git a/tez-mapreduce/src/test/java/org/apache/tez/mapreduce/processor/reduce/TestReduceProcessor.java b/tez-mapreduce/src/test/java/org/apache/tez/mapreduce/processor/reduce/TestReduceProcessor.java
index 5a2a83e..bbe47d9 100644
--- a/tez-mapreduce/src/test/java/org/apache/tez/mapreduce/processor/reduce/TestReduceProcessor.java
+++ b/tez-mapreduce/src/test/java/org/apache/tez/mapreduce/processor/reduce/TestReduceProcessor.java
@@ -114,7 +114,7 @@ public class TestReduceProcessor {
@Test
public void testReduceProcessor() throws Exception {
-
+ final String dagName = "mrdag0";
String mapVertexName = MultiStageMRConfigUtil.getInitialMapVertexName();
String reduceVertexName = MultiStageMRConfigUtil.getFinalReduceVertexName();
JobConf jobConf = new JobConf(defaultConf);
@@ -142,7 +142,7 @@ public class TestReduceProcessor {
OutputSpec mapOutputSpec = new OutputSpec("NullDestVertex", new OutputDescriptor(LocalOnFileSorterOutput.class.getName()), 1);
// Run a map
LogicalIOProcessorRuntimeTask mapTask = MapUtils.createLogicalTask(localFs, workDir, mapConf, 0,
- mapInput, new TestUmbilical(), mapVertexName,
+ mapInput, new TestUmbilical(), dagName, mapVertexName,
Collections.singletonList(mapInputSpec),
Collections.singletonList(mapOutputSpec));
@@ -172,6 +172,7 @@ public class TestReduceProcessor {
// Now run a reduce
TaskSpec taskSpec = new TaskSpec(
TezTestUtils.getMockTaskAttemptId(0, 1, 0, 0),
+ dagName,
reduceVertexName,
reduceProcessorDesc,
Collections.singletonList(reduceInputSpec),
http://git-wip-us.apache.org/repos/asf/incubator-tez/blob/f38b3309/tez-runtime-internals/src/main/java/org/apache/tez/runtime/LogicalIOProcessorRuntimeTask.java
----------------------------------------------------------------------
diff --git a/tez-runtime-internals/src/main/java/org/apache/tez/runtime/LogicalIOProcessorRuntimeTask.java b/tez-runtime-internals/src/main/java/org/apache/tez/runtime/LogicalIOProcessorRuntimeTask.java
index ae30b70..41326b9 100644
--- a/tez-runtime-internals/src/main/java/org/apache/tez/runtime/LogicalIOProcessorRuntimeTask.java
+++ b/tez-runtime-internals/src/main/java/org/apache/tez/runtime/LogicalIOProcessorRuntimeTask.java
@@ -461,7 +461,8 @@ public class LogicalIOProcessorRuntimeTask extends RuntimeTask {
private TezInputContext createInputContext(Input input, InputSpec inputSpec, int inputIndex) {
TezInputContext inputContext = new TezInputContextImpl(tezConf,
- appAttemptNumber, tezUmbilical, taskSpec.getVertexName(),
+ appAttemptNumber, tezUmbilical,
+ taskSpec.getDAGName(), taskSpec.getVertexName(),
inputSpec.getSourceVertexName(), taskSpec.getTaskAttemptID(),
tezCounters, inputIndex,
inputSpec.getInputDescriptor().getUserPayload() == null ? taskSpec
@@ -474,7 +475,8 @@ public class LogicalIOProcessorRuntimeTask extends RuntimeTask {
private TezOutputContext createOutputContext(OutputSpec outputSpec, int outputIndex) {
TezOutputContext outputContext = new TezOutputContextImpl(tezConf,
- appAttemptNumber, tezUmbilical, taskSpec.getVertexName(),
+ appAttemptNumber, tezUmbilical,
+ taskSpec.getDAGName(), taskSpec.getVertexName(),
outputSpec.getDestinationVertexName(), taskSpec.getTaskAttemptID(),
tezCounters, outputIndex,
outputSpec.getOutputDescriptor().getUserPayload() == null ? taskSpec
@@ -487,7 +489,9 @@ public class LogicalIOProcessorRuntimeTask extends RuntimeTask {
private TezProcessorContext createProcessorContext() {
TezProcessorContext processorContext = new TezProcessorContextImpl(tezConf,
- appAttemptNumber, tezUmbilical, taskSpec.getVertexName(), taskSpec.getTaskAttemptID(),
+ appAttemptNumber, tezUmbilical,
+ taskSpec.getDAGName(), taskSpec.getVertexName(),
+ taskSpec.getTaskAttemptID(),
tezCounters, processorDescriptor.getUserPayload(), this,
serviceConsumerMetadata, System.getenv(), initialMemoryDistributor,
processorDescriptor, inputReadyTracker);
http://git-wip-us.apache.org/repos/asf/incubator-tez/blob/f38b3309/tez-runtime-internals/src/main/java/org/apache/tez/runtime/api/impl/TaskSpec.java
----------------------------------------------------------------------
diff --git a/tez-runtime-internals/src/main/java/org/apache/tez/runtime/api/impl/TaskSpec.java b/tez-runtime-internals/src/main/java/org/apache/tez/runtime/api/impl/TaskSpec.java
index 8752521..6466cdd 100644
--- a/tez-runtime-internals/src/main/java/org/apache/tez/runtime/api/impl/TaskSpec.java
+++ b/tez-runtime-internals/src/main/java/org/apache/tez/runtime/api/impl/TaskSpec.java
@@ -33,6 +33,7 @@ import com.google.common.collect.Lists;
public class TaskSpec implements Writable {
private TezTaskAttemptID taskAttemptId;
+ private String dagName;
private String vertexName;
private ProcessorDescriptor processorDescriptor;
private List<InputSpec> inputSpecList;
@@ -43,10 +44,13 @@ public class TaskSpec implements Writable {
}
public TaskSpec(TezTaskAttemptID taskAttemptID,
- String vertexName, ProcessorDescriptor processorDescriptor,
+ String dagName, String vertexName,
+ ProcessorDescriptor processorDescriptor,
List<InputSpec> inputSpecList, List<OutputSpec> outputSpecList,
List<GroupInputSpec> groupInputSpecList) {
+ // TODO: TEZ-884 null check
this.taskAttemptId = taskAttemptID;
+ this.dagName = StringInterner.weakIntern(dagName);
this.vertexName = StringInterner.weakIntern(vertexName);
this.processorDescriptor = processorDescriptor;
this.inputSpecList = inputSpecList;
@@ -54,6 +58,10 @@ public class TaskSpec implements Writable {
this.groupInputSpecList = groupInputSpecList;
}
+ public String getDAGName() {
+ return dagName;
+ }
+
public String getVertexName() {
return vertexName;
}
@@ -81,6 +89,7 @@ public class TaskSpec implements Writable {
@Override
public void write(DataOutput out) throws IOException {
taskAttemptId.write(out);
+ out.writeUTF(dagName);
out.writeUTF(vertexName);
processorDescriptor.write(out);
out.writeInt(inputSpecList.size());
@@ -105,6 +114,7 @@ public class TaskSpec implements Writable {
@Override
public void readFields(DataInput in) throws IOException {
taskAttemptId = TezTaskAttemptID.readTezTaskAttemptID(in);
+ dagName = StringInterner.weakIntern(in.readUTF());
vertexName = StringInterner.weakIntern(in.readUTF());
// TODO TEZ-305 convert this to PB
processorDescriptor = new ProcessorDescriptor();
@@ -138,7 +148,8 @@ public class TaskSpec implements Writable {
@Override
public String toString() {
StringBuffer sb = new StringBuffer();
- sb.append("VertexName: " + vertexName);
+ sb.append("DAGName : " + dagName);
+ sb.append(", VertexName: " + vertexName);
sb.append(", TaskAttemptID:" + taskAttemptId);
sb.append(", processorName=" + processorDescriptor.getClassName()
+ ", inputSpecListSize=" + inputSpecList.size()
http://git-wip-us.apache.org/repos/asf/incubator-tez/blob/f38b3309/tez-runtime-internals/src/main/java/org/apache/tez/runtime/api/impl/TezInputContextImpl.java
----------------------------------------------------------------------
diff --git a/tez-runtime-internals/src/main/java/org/apache/tez/runtime/api/impl/TezInputContextImpl.java b/tez-runtime-internals/src/main/java/org/apache/tez/runtime/api/impl/TezInputContextImpl.java
index d4b07d1..a80d90b 100644
--- a/tez-runtime-internals/src/main/java/org/apache/tez/runtime/api/impl/TezInputContextImpl.java
+++ b/tez-runtime-internals/src/main/java/org/apache/tez/runtime/api/impl/TezInputContextImpl.java
@@ -48,13 +48,13 @@ public class TezInputContextImpl extends TezTaskContextImpl
@Private
public TezInputContextImpl(Configuration conf, int appAttemptNumber,
- TezUmbilical tezUmbilical, String taskVertexName,
+ TezUmbilical tezUmbilical, String dagName, String taskVertexName,
String sourceVertexName, TezTaskAttemptID taskAttemptID,
TezCounters counters, int inputIndex, byte[] userPayload,
RuntimeTask runtimeTask, Map<String, ByteBuffer> serviceConsumerMetadata,
Map<String, String> auxServiceEnv, MemoryDistributor memDist,
InputDescriptor inputDescriptor, Input input, InputReadyTracker inputReadyTracker) {
- super(conf, appAttemptNumber, taskVertexName, taskAttemptID,
+ super(conf, appAttemptNumber, dagName, taskVertexName, taskAttemptID,
counters, runtimeTask, tezUmbilical, serviceConsumerMetadata,
auxServiceEnv, memDist, inputDescriptor);
this.userPayload = userPayload;
http://git-wip-us.apache.org/repos/asf/incubator-tez/blob/f38b3309/tez-runtime-internals/src/main/java/org/apache/tez/runtime/api/impl/TezOutputContextImpl.java
----------------------------------------------------------------------
diff --git a/tez-runtime-internals/src/main/java/org/apache/tez/runtime/api/impl/TezOutputContextImpl.java b/tez-runtime-internals/src/main/java/org/apache/tez/runtime/api/impl/TezOutputContextImpl.java
index 6f45d39..5bb261d 100644
--- a/tez-runtime-internals/src/main/java/org/apache/tez/runtime/api/impl/TezOutputContextImpl.java
+++ b/tez-runtime-internals/src/main/java/org/apache/tez/runtime/api/impl/TezOutputContextImpl.java
@@ -44,14 +44,15 @@ public class TezOutputContextImpl extends TezTaskContextImpl
@Private
public TezOutputContextImpl(Configuration conf, int appAttemptNumber,
- TezUmbilical tezUmbilical, String taskVertexName,
+ TezUmbilical tezUmbilical, String dagName,
+ String taskVertexName,
String destinationVertexName,
TezTaskAttemptID taskAttemptID, TezCounters counters, int outputIndex,
byte[] userPayload, RuntimeTask runtimeTask,
Map<String, ByteBuffer> serviceConsumerMetadata,
Map<String, String> auxServiceEnv, MemoryDistributor memDist,
OutputDescriptor outputDescriptor) {
- super(conf, appAttemptNumber, taskVertexName, taskAttemptID,
+ super(conf, appAttemptNumber, dagName, taskVertexName, taskAttemptID,
counters, runtimeTask, tezUmbilical, serviceConsumerMetadata,
auxServiceEnv, memDist, outputDescriptor);
this.userPayload = userPayload;
http://git-wip-us.apache.org/repos/asf/incubator-tez/blob/f38b3309/tez-runtime-internals/src/main/java/org/apache/tez/runtime/api/impl/TezProcessorContextImpl.java
----------------------------------------------------------------------
diff --git a/tez-runtime-internals/src/main/java/org/apache/tez/runtime/api/impl/TezProcessorContextImpl.java b/tez-runtime-internals/src/main/java/org/apache/tez/runtime/api/impl/TezProcessorContextImpl.java
index 106251c..2f9ebfd 100644
--- a/tez-runtime-internals/src/main/java/org/apache/tez/runtime/api/impl/TezProcessorContextImpl.java
+++ b/tez-runtime-internals/src/main/java/org/apache/tez/runtime/api/impl/TezProcessorContextImpl.java
@@ -48,13 +48,13 @@ public class TezProcessorContextImpl extends TezTaskContextImpl implements TezPr
private final InputReadyTracker inputReadyTracker;
public TezProcessorContextImpl(Configuration conf, int appAttemptNumber,
- TezUmbilical tezUmbilical, String vertexName,
+ TezUmbilical tezUmbilical, String dagName, String vertexName,
TezTaskAttemptID taskAttemptID, TezCounters counters,
byte[] userPayload, RuntimeTask runtimeTask,
Map<String, ByteBuffer> serviceConsumerMetadata,
Map<String, String> auxServiceEnv, MemoryDistributor memDist,
ProcessorDescriptor processorDescriptor, InputReadyTracker inputReadyTracker) {
- super(conf, appAttemptNumber, vertexName, taskAttemptID,
+ super(conf, appAttemptNumber, dagName, vertexName, taskAttemptID,
counters, runtimeTask, tezUmbilical, serviceConsumerMetadata,
auxServiceEnv, memDist, processorDescriptor);
this.userPayload = userPayload;
http://git-wip-us.apache.org/repos/asf/incubator-tez/blob/f38b3309/tez-runtime-internals/src/main/java/org/apache/tez/runtime/api/impl/TezTaskContextImpl.java
----------------------------------------------------------------------
diff --git a/tez-runtime-internals/src/main/java/org/apache/tez/runtime/api/impl/TezTaskContextImpl.java b/tez-runtime-internals/src/main/java/org/apache/tez/runtime/api/impl/TezTaskContextImpl.java
index e15f6f2..53f6a56 100644
--- a/tez-runtime-internals/src/main/java/org/apache/tez/runtime/api/impl/TezTaskContextImpl.java
+++ b/tez-runtime-internals/src/main/java/org/apache/tez/runtime/api/impl/TezTaskContextImpl.java
@@ -56,15 +56,17 @@ public abstract class TezTaskContextImpl implements TezTaskContext {
private final Map<String, String> auxServiceEnv;
protected final MemoryDistributor initialMemoryDistributor;
protected final TezEntityDescriptor descriptor;
+ private final String dagName;
@Private
public TezTaskContextImpl(Configuration conf, int appAttemptNumber,
- String taskVertexName, TezTaskAttemptID taskAttemptID,
+ String dagName, String taskVertexName, TezTaskAttemptID taskAttemptID,
TezCounters counters, RuntimeTask runtimeTask,
TezUmbilical tezUmbilical, Map<String, ByteBuffer> serviceConsumerMetadata,
Map<String, String> auxServiceEnv, MemoryDistributor memDist,
TezEntityDescriptor descriptor) {
this.conf = conf;
+ this.dagName = dagName;
this.taskVertexName = taskVertexName;
this.taskAttemptID = taskAttemptID;
this.counters = counters;
@@ -106,9 +108,7 @@ public abstract class TezTaskContextImpl implements TezTaskContext {
@Override
public String getDAGName() {
- // TODO NEWTEZ Change to some form of the DAG name, for now using dagId as
- // the unique identifier.
- return taskAttemptID.getTaskID().getVertexID().getDAGId().toString();
+ return dagName;
}
@Override
http://git-wip-us.apache.org/repos/asf/incubator-tez/blob/f38b3309/tez-runtime-internals/src/test/java/org/apache/tez/runtime/TestLogicalIOProcessorRuntimeTask.java
----------------------------------------------------------------------
diff --git a/tez-runtime-internals/src/test/java/org/apache/tez/runtime/TestLogicalIOProcessorRuntimeTask.java b/tez-runtime-internals/src/test/java/org/apache/tez/runtime/TestLogicalIOProcessorRuntimeTask.java
index c2dd68b..27a250e 100644
--- a/tez-runtime-internals/src/test/java/org/apache/tez/runtime/TestLogicalIOProcessorRuntimeTask.java
+++ b/tez-runtime-internals/src/test/java/org/apache/tez/runtime/TestLogicalIOProcessorRuntimeTask.java
@@ -65,10 +65,10 @@ public class TestLogicalIOProcessorRuntimeTask {
TezConfiguration tezConf = new TezConfiguration();
TezTaskAttemptID taId1 = createTaskAttemptID(vertexId, 1);
- TaskSpec task1 = createTaskSpec(taId1, "vertex1");
+ TaskSpec task1 = createTaskSpec(taId1, "dag1", "vertex1");
TezTaskAttemptID taId2 = createTaskAttemptID(vertexId, 2);
- TaskSpec task2 = createTaskSpec(taId2, "vertex1");
+ TaskSpec task2 = createTaskSpec(taId2, "dag2", "vertex1");
LogicalIOProcessorRuntimeTask lio1 = new LogicalIOProcessorRuntimeTask(task1, 0, tezConf,
umbilical, serviceConsumerMetadata, startedInputsMap);
@@ -96,9 +96,11 @@ public class TestLogicalIOProcessorRuntimeTask {
}
- private TaskSpec createTaskSpec(TezTaskAttemptID taskAttemptID, String vertexName) {
+ private TaskSpec createTaskSpec(TezTaskAttemptID taskAttemptID,
+ String dagName, String vertexName) {
ProcessorDescriptor processorDesc = createProcessorDescriptor();
- TaskSpec taskSpec = new TaskSpec(taskAttemptID, vertexName, processorDesc,
+ TaskSpec taskSpec = new TaskSpec(taskAttemptID,
+ dagName, vertexName, processorDesc,
createInputSpecList(), createOutputSpecList(), null);
return taskSpec;
}
http://git-wip-us.apache.org/repos/asf/incubator-tez/blob/f38b3309/tez-runtime-library/src/test/java/org/apache/tez/runtime/library/output/TestOnFileUnorderedKVOutput.java
----------------------------------------------------------------------
diff --git a/tez-runtime-library/src/test/java/org/apache/tez/runtime/library/output/TestOnFileUnorderedKVOutput.java b/tez-runtime-library/src/test/java/org/apache/tez/runtime/library/output/TestOnFileUnorderedKVOutput.java
index 4f48ad2..61d7019 100644
--- a/tez-runtime-library/src/test/java/org/apache/tez/runtime/library/output/TestOnFileUnorderedKVOutput.java
+++ b/tez-runtime-library/src/test/java/org/apache/tez/runtime/library/output/TestOnFileUnorderedKVOutput.java
@@ -100,6 +100,7 @@ public class TestOnFileUnorderedKVOutput {
int appAttemptNumber = 1;
TezUmbilical tezUmbilical = null;
+ String dagName = "currentDAG";
String taskVertexName = "currentVertex";
String destinationVertexName = "destinationVertex";
TezDAGID dagID = TezDAGID.getInstance("2000", 1, 1);
@@ -119,7 +120,7 @@ public class TestOnFileUnorderedKVOutput {
TezOutputContext outputContext = new TezOutputContextImpl(conf,
- appAttemptNumber, tezUmbilical, taskVertexName, destinationVertexName,
+ appAttemptNumber, tezUmbilical, dagName, taskVertexName, destinationVertexName,
taskAttemptID, counters, 0, userPayload, runtimeTask,
null, auxEnv, new MemoryDistributor(1, 1, conf) , mock(OutputDescriptor.class));