You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@tez.apache.org by zj...@apache.org on 2014/11/12 05:56:05 UTC
tez git commit: TEZ-1737. Should add taskNum in VertexFinishedEvent
(zjffdu)
Repository: tez
Updated Branches:
refs/heads/master 103839b6c -> 0ebfc1b10
TEZ-1737. Should add taskNum in VertexFinishedEvent (zjffdu)
Project: http://git-wip-us.apache.org/repos/asf/tez/repo
Commit: http://git-wip-us.apache.org/repos/asf/tez/commit/0ebfc1b1
Tree: http://git-wip-us.apache.org/repos/asf/tez/tree/0ebfc1b1
Diff: http://git-wip-us.apache.org/repos/asf/tez/diff/0ebfc1b1
Branch: refs/heads/master
Commit: 0ebfc1b10b5000baadee930326b7f6b81516ea39
Parents: 103839b
Author: Jeff Zhang <zj...@apache.org>
Authored: Wed Nov 12 12:54:57 2014 +0800
Committer: Jeff Zhang <zj...@apache.org>
Committed: Wed Nov 12 12:54:57 2014 +0800
----------------------------------------------------------------------
CHANGES.txt | 1 +
.../apache/tez/dag/app/dag/impl/VertexImpl.java | 3 +-
.../dag/history/events/VertexFinishedEvent.java | 10 ++-
tez-dag/src/main/proto/HistoryEvents.proto | 1 +
.../dag/app/dag/impl/TestVertexRecovery.java | 89 +++++++++++++++++++-
.../TestHistoryEventsProtoConversion.java | 4 +-
.../impl/TestHistoryEventJsonConversion.java | 3 +-
.../ats/TestHistoryEventTimelineConversion.java | 4 +-
8 files changed, 105 insertions(+), 10 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/tez/blob/0ebfc1b1/CHANGES.txt
----------------------------------------------------------------------
diff --git a/CHANGES.txt b/CHANGES.txt
index 34e2c7a..7ad6903 100644
--- a/CHANGES.txt
+++ b/CHANGES.txt
@@ -22,6 +22,7 @@ ALL CHANGES:
TEZ-1750. Add a DAGScheduler which schedules tasks only when sources have been scheduled.
TEZ-1761. TestRecoveryParser::testGetLastInProgressDAG fails in similar manner to TEZ-1686.
TEZ-1687. Use logIdentifier of Vertex for logging.
+ TEZ-1737. Should add taskNum in VertexFinishedEvent.
Release 0.5.2: 2014-11-07
http://git-wip-us.apache.org/repos/asf/tez/blob/0ebfc1b1/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 8130c18..d19c4cc 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
@@ -1135,6 +1135,7 @@ public class VertexImpl implements org.apache.tez.dag.app.dag.Vertex,
} else {
vertexCompleteSeen = true;
}
+ numTasks = finishedEvent.getNumTasks();
recoveryCommitInProgress = false;
recoveredState = finishedEvent.getState();
diagnostics.add(finishedEvent.getDiagnostics());
@@ -1610,7 +1611,7 @@ public class VertexImpl implements org.apache.tez.dag.app.dag.Vertex,
taskStats.put(ATSConstants.NUM_FAILED_TASKS_ATTEMPTS, failedTaskAttemptCount.get());
taskStats.put(ATSConstants.NUM_KILLED_TASKS_ATTEMPTS, killedTaskAttemptCount.get());
- VertexFinishedEvent finishEvt = new VertexFinishedEvent(vertexId, vertexName, initTimeRequested,
+ VertexFinishedEvent finishEvt = new VertexFinishedEvent(vertexId, vertexName, numTasks, initTimeRequested,
initedTime, startTimeRequested, startedTime, finishTime, finalState, diagnostics,
getAllCounters(), getVertexStats(), taskStats);
this.appContext.getHistoryHandler().handleCriticalEvent(
http://git-wip-us.apache.org/repos/asf/tez/blob/0ebfc1b1/tez-dag/src/main/java/org/apache/tez/dag/history/events/VertexFinishedEvent.java
----------------------------------------------------------------------
diff --git a/tez-dag/src/main/java/org/apache/tez/dag/history/events/VertexFinishedEvent.java b/tez-dag/src/main/java/org/apache/tez/dag/history/events/VertexFinishedEvent.java
index d9cafc7..d1d7eef 100644
--- a/tez-dag/src/main/java/org/apache/tez/dag/history/events/VertexFinishedEvent.java
+++ b/tez-dag/src/main/java/org/apache/tez/dag/history/events/VertexFinishedEvent.java
@@ -43,6 +43,7 @@ public class VertexFinishedEvent implements HistoryEvent, SummaryEvent {
private TezVertexID vertexID;
private String vertexName;
+ private int numTasks;
private long initRequestedTime;
private long initedTime;
private long startRequestedTime;
@@ -55,13 +56,14 @@ public class VertexFinishedEvent implements HistoryEvent, SummaryEvent {
private VertexStats vertexStats;
private Map<String, Integer> vertexTaskStats;
- public VertexFinishedEvent(TezVertexID vertexId, String vertexName, long initRequestedTime,
+ public VertexFinishedEvent(TezVertexID vertexId, String vertexName, int numTasks, long initRequestedTime,
long initedTime, long startRequestedTime, long startedTime,
long finishTime, VertexState state, String diagnostics,
TezCounters counters, VertexStats vertexStats,
Map<String, Integer> vertexTaskStats) {
this.vertexName = vertexName;
this.vertexID = vertexId;
+ this.numTasks = numTasks;
this.initRequestedTime = initRequestedTime;
this.initedTime = initedTime;
this.startRequestedTime = startRequestedTime;
@@ -182,12 +184,17 @@ public class VertexFinishedEvent implements HistoryEvent, SummaryEvent {
return vertexTaskStats;
}
+ public int getNumTasks() {
+ return numTasks;
+ }
+
@Override
public void toSummaryProtoStream(OutputStream outputStream) throws IOException {
VertexFinishStateProto finishStateProto =
VertexFinishStateProto.newBuilder()
.setState(state.ordinal())
.setVertexId(vertexID.toString())
+ .setNumTasks(numTasks)
.build();
SummaryEventProto.Builder builder = RecoveryProtos.SummaryEventProto.newBuilder()
@@ -205,6 +212,7 @@ public class VertexFinishedEvent implements HistoryEvent, SummaryEvent {
VertexFinishStateProto.parseFrom(proto.getEventPayload());
this.vertexID = TezVertexID.fromString(finishStateProto.getVertexId());
this.state = VertexState.values()[finishStateProto.getState()];
+ this.numTasks = finishStateProto.getNumTasks();
this.finishTime = proto.getTimestamp();
this.fromSummary = true;
}
http://git-wip-us.apache.org/repos/asf/tez/blob/0ebfc1b1/tez-dag/src/main/proto/HistoryEvents.proto
----------------------------------------------------------------------
diff --git a/tez-dag/src/main/proto/HistoryEvents.proto b/tez-dag/src/main/proto/HistoryEvents.proto
index 93f217f..5d18b0a 100644
--- a/tez-dag/src/main/proto/HistoryEvents.proto
+++ b/tez-dag/src/main/proto/HistoryEvents.proto
@@ -198,4 +198,5 @@ message SummaryEventProto {
message VertexFinishStateProto {
optional string vertex_id = 1;
optional int32 state = 2;
+ optional int32 num_tasks = 3;
}
http://git-wip-us.apache.org/repos/asf/tez/blob/0ebfc1b1/tez-dag/src/test/java/org/apache/tez/dag/app/dag/impl/TestVertexRecovery.java
----------------------------------------------------------------------
diff --git a/tez-dag/src/test/java/org/apache/tez/dag/app/dag/impl/TestVertexRecovery.java b/tez-dag/src/test/java/org/apache/tez/dag/app/dag/impl/TestVertexRecovery.java
index 8fa574c..1b8ac2f 100644
--- a/tez-dag/src/test/java/org/apache/tez/dag/app/dag/impl/TestVertexRecovery.java
+++ b/tez-dag/src/test/java/org/apache/tez/dag/app/dag/impl/TestVertexRecovery.java
@@ -47,6 +47,7 @@ import org.apache.tez.dag.api.records.DAGProtos.PlanEdgeSchedulingType;
import org.apache.tez.dag.api.records.DAGProtos.PlanTaskConfiguration;
import org.apache.tez.dag.api.records.DAGProtos.PlanTaskLocationHint;
import org.apache.tez.dag.api.records.DAGProtos.PlanVertexType;
+import org.apache.tez.dag.api.records.DAGProtos.RootInputLeafOutputProto;
import org.apache.tez.dag.api.records.DAGProtos.TezEntityDescriptorProto;
import org.apache.tez.dag.api.records.DAGProtos.VertexPlan;
import org.apache.tez.dag.app.AppContext;
@@ -67,11 +68,14 @@ import org.apache.tez.dag.app.dag.event.VertexEventRecoverVertex;
import org.apache.tez.dag.app.dag.event.VertexEventType;
import org.apache.tez.dag.app.dag.impl.AMUserCodeException.Source;
import org.apache.tez.dag.app.dag.impl.TestVertexImpl.CountingOutputCommitter;
+import org.apache.tez.dag.history.HistoryEventType;
import org.apache.tez.dag.history.events.VertexRecoverableEventsGeneratedEvent;
import org.apache.tez.dag.history.events.VertexFinishedEvent;
import org.apache.tez.dag.history.events.VertexInitializedEvent;
import org.apache.tez.dag.history.events.VertexStartedEvent;
import org.apache.tez.dag.records.TezDAGID;
+import org.apache.tez.dag.recovery.records.RecoveryProtos.SummaryEventProto;
+import org.apache.tez.dag.recovery.records.RecoveryProtos.VertexFinishStateProto;
import org.apache.tez.runtime.api.OutputCommitter;
import org.apache.tez.runtime.api.events.InputDataInformationEvent;
import org.apache.tez.runtime.api.impl.EventMetaData;
@@ -215,6 +219,49 @@ public class TestVertexRecovery {
return dag;
}
+ private DAGPlan createDAGPlanSingleVertex() {
+ DAGPlan dag =
+ DAGPlan
+ .newBuilder()
+ .setName("testverteximpl")
+ .addVertex(
+ VertexPlan
+ .newBuilder()
+ .setName("vertex1")
+ .setType(PlanVertexType.NORMAL)
+ .addTaskLocationHint(
+ PlanTaskLocationHint.newBuilder().addHost("host1")
+ .addRack("rack1").build())
+ .setTaskConfig(
+ PlanTaskConfiguration.newBuilder().setNumTasks(-1)
+ .setVirtualCores(4).setMemoryMb(1024)
+ .setJavaOpts("").setTaskModule("x1.y1").build())
+ .addInputs(RootInputLeafOutputProto.newBuilder()
+ .setIODescriptor(
+ TezEntityDescriptorProto.newBuilder()
+ .setClassName("input").build())
+ .setName("inputx")
+ .setControllerDescriptor(
+ TezEntityDescriptorProto
+ .newBuilder()
+ .setClassName("inputinitlizer"))
+ .build())
+ .addOutputs(
+ DAGProtos.RootInputLeafOutputProto
+ .newBuilder()
+ .setIODescriptor(
+ TezEntityDescriptorProto.newBuilder()
+ .setClassName("output").build())
+ .setName("outputx")
+ .setControllerDescriptor(
+ TezEntityDescriptorProto
+ .newBuilder()
+ .setClassName(
+ CountingOutputCommitter.class.getName())))
+ .build()).build();
+ return dag;
+ }
+
class VertexEventHanlder implements EventHandler<VertexEvent> {
private List<VertexEvent> events = new ArrayList<VertexEvent>();
@@ -316,6 +363,44 @@ public class TestVertexRecovery {
}
/**
+ * vertex1(New) -> StartRecoveryTransition(SUCCEEDED)
+ * @throws IOException
+ */
+ @Test
+ public void testRecovery_Desired_SUCCEEDED_OnlySummaryLog() throws IOException {
+ DAGPlan dagPlan = createDAGPlanSingleVertex();
+ dag =
+ new DAGImpl(dagId, new Configuration(), dagPlan,
+ dispatcher.getEventHandler(), mock(TaskAttemptListener.class),
+ new Credentials(), new SystemClock(), user,
+ mock(TaskHeartbeatHandler.class), mockAppContext);
+ when(mockAppContext.getCurrentDAG()).thenReturn(dag);
+ dag.handle(new DAGEvent(dagId, DAGEventType.DAG_INIT));
+
+ VertexImpl vertex1 = (VertexImpl) dag.getVertex("vertex1");
+ VertexFinishedEvent vertexFinishEvent = new VertexFinishedEvent();
+ vertexFinishEvent.fromSummaryProtoStream(SummaryEventProto.newBuilder()
+ .setDagId(dag.getID().toString())
+ .setEventType(HistoryEventType.VERTEX_FINISHED.ordinal())
+ .setTimestamp(100L)
+ .setEventPayload(VertexFinishStateProto.newBuilder()
+ .setNumTasks(2)
+ .setState(VertexState.SUCCEEDED.ordinal())
+ .setVertexId(vertex1.getVertexId().toString()).build().toByteString())
+ .build());
+ VertexState recoveredState = vertex1.restoreFromEvent(vertexFinishEvent);
+ // numTasks is recovered from summary log
+ assertEquals(2, vertex1.numTasks);
+ assertEquals(VertexState.SUCCEEDED, recoveredState);
+ vertex1.handle(new VertexEventRecoverVertex(vertex1.getVertexId(),
+ VertexState.SUCCEEDED));
+ dispatcher.await();
+ assertEquals(VertexState.SUCCEEDED, vertex1.getState());
+ assertEquals(vertex1.numTasks, vertex1.succeededTaskCount);
+ assertEquals(vertex1.numTasks, vertex1.completedTaskCount);
+ }
+
+ /**
* vertex1(New) -> StartRecoveryTransition(FAILED)
*/
@Test
@@ -576,7 +661,7 @@ public class TestVertexRecovery {
long finishTime = startTime + 100L;
recoveredState =
vertex1.restoreFromEvent(new VertexFinishedEvent(vertex1.getVertexId(),
- "vertex1", initRequestedTime, initedTime, startRequestedTime,
+ "vertex1", 1, initRequestedTime, initedTime, startRequestedTime,
startTime, finishTime, VertexState.SUCCEEDED, "",
new TezCounters(), new VertexStats(), null));
assertEquals(finishTime, vertex1.finishTime);
@@ -847,7 +932,7 @@ public class TestVertexRecovery {
assertEquals(VertexState.RUNNING, recoveredState);
recoveredState = vertex1.restoreFromEvent(new VertexFinishedEvent(vertex1.getVertexId(),
- "vertex1", initRequestedTime, initedTime, initRequestedTime + 300L,
+ "vertex1", 1, initRequestedTime, initedTime, initRequestedTime + 300L,
initRequestedTime + 400L, initRequestedTime + 500L,
VertexState.SUCCEEDED, "", new TezCounters(), new VertexStats(), null));
assertEquals(VertexState.SUCCEEDED, recoveredState);
http://git-wip-us.apache.org/repos/asf/tez/blob/0ebfc1b1/tez-dag/src/test/java/org/apache/tez/dag/history/events/TestHistoryEventsProtoConversion.java
----------------------------------------------------------------------
diff --git a/tez-dag/src/test/java/org/apache/tez/dag/history/events/TestHistoryEventsProtoConversion.java b/tez-dag/src/test/java/org/apache/tez/dag/history/events/TestHistoryEventsProtoConversion.java
index a7a23db..8913287 100644
--- a/tez-dag/src/test/java/org/apache/tez/dag/history/events/TestHistoryEventsProtoConversion.java
+++ b/tez-dag/src/test/java/org/apache/tez/dag/history/events/TestHistoryEventsProtoConversion.java
@@ -361,7 +361,7 @@ public class TestHistoryEventsProtoConversion {
VertexFinishedEvent event =
new VertexFinishedEvent(TezVertexID.getInstance(
TezDAGID.getInstance(ApplicationId.newInstance(0, 1), 1), 111),
- "vertex1", 1000l, 15000l, 16000l, 20000l, 1344400l, VertexState.ERROR,
+ "vertex1", 1, 1000l, 15000l, 16000l, 20000l, 1344400l, VertexState.ERROR,
null, null, null, null);
VertexFinishedEvent deserializedEvent = (VertexFinishedEvent)
testProtoConversion(event);
@@ -376,7 +376,7 @@ public class TestHistoryEventsProtoConversion {
VertexFinishedEvent event =
new VertexFinishedEvent(TezVertexID.getInstance(
TezDAGID.getInstance(ApplicationId.newInstance(0, 1), 1), 111),
- "vertex1", 1000l, 15000l, 16000l, 20000l, 1344400l, VertexState.ERROR,
+ "vertex1", 1, 1000l, 15000l, 16000l, 20000l, 1344400l, VertexState.ERROR,
"diagnose", new TezCounters(), new VertexStats(), null);
VertexFinishedEvent deserializedEvent = (VertexFinishedEvent)
testProtoConversion(event);
http://git-wip-us.apache.org/repos/asf/tez/blob/0ebfc1b1/tez-dag/src/test/java/org/apache/tez/dag/history/logging/impl/TestHistoryEventJsonConversion.java
----------------------------------------------------------------------
diff --git a/tez-dag/src/test/java/org/apache/tez/dag/history/logging/impl/TestHistoryEventJsonConversion.java b/tez-dag/src/test/java/org/apache/tez/dag/history/logging/impl/TestHistoryEventJsonConversion.java
index c3d51c3..a20c9fe 100644
--- a/tez-dag/src/test/java/org/apache/tez/dag/history/logging/impl/TestHistoryEventJsonConversion.java
+++ b/tez-dag/src/test/java/org/apache/tez/dag/history/logging/impl/TestHistoryEventJsonConversion.java
@@ -27,7 +27,6 @@ import org.apache.hadoop.yarn.api.records.ApplicationAttemptId;
import org.apache.hadoop.yarn.api.records.ApplicationId;
import org.apache.hadoop.yarn.api.records.ContainerId;
import org.apache.hadoop.yarn.api.records.NodeId;
-import org.apache.hadoop.yarn.api.records.timeline.TimelineEvent;
import org.apache.tez.common.ATSConstants;
import org.apache.tez.dag.api.EdgeManagerPluginDescriptor;
import org.apache.tez.dag.api.oldrecords.TaskAttemptState;
@@ -139,7 +138,7 @@ public class TestHistoryEventJsonConversion {
event = new VertexParallelismUpdatedEvent(tezVertexID, 1, null, null, null, 10);
break;
case VERTEX_FINISHED:
- event = new VertexFinishedEvent(tezVertexID, "v1", random.nextInt(), random.nextInt(),
+ event = new VertexFinishedEvent(tezVertexID, "v1", 1, random.nextInt(), random.nextInt(),
random.nextInt(), random.nextInt(), random.nextInt(), VertexState.ERROR,
null, null, null, null);
break;
http://git-wip-us.apache.org/repos/asf/tez/blob/0ebfc1b1/tez-plugins/tez-yarn-timeline-history/src/test/java/org/apache/tez/dag/history/logging/ats/TestHistoryEventTimelineConversion.java
----------------------------------------------------------------------
diff --git a/tez-plugins/tez-yarn-timeline-history/src/test/java/org/apache/tez/dag/history/logging/ats/TestHistoryEventTimelineConversion.java b/tez-plugins/tez-yarn-timeline-history/src/test/java/org/apache/tez/dag/history/logging/ats/TestHistoryEventTimelineConversion.java
index 2fed18b..a40aa74 100644
--- a/tez-plugins/tez-yarn-timeline-history/src/test/java/org/apache/tez/dag/history/logging/ats/TestHistoryEventTimelineConversion.java
+++ b/tez-plugins/tez-yarn-timeline-history/src/test/java/org/apache/tez/dag/history/logging/ats/TestHistoryEventTimelineConversion.java
@@ -140,7 +140,7 @@ public class TestHistoryEventTimelineConversion {
event = new VertexParallelismUpdatedEvent(tezVertexID, 1, null, null, null, 1);
break;
case VERTEX_FINISHED:
- event = new VertexFinishedEvent(tezVertexID, "v1", random.nextInt(), random.nextInt(),
+ event = new VertexFinishedEvent(tezVertexID, "v1", 1, random.nextInt(), random.nextInt(),
random.nextInt(), random.nextInt(), random.nextInt(), VertexState.ERROR,
null, null, null, null);
break;
@@ -448,7 +448,7 @@ public class TestHistoryEventTimelineConversion {
taskStats.put("BAR", 200);
VertexStats vertexStats = new VertexStats();
- VertexFinishedEvent event = new VertexFinishedEvent(tezVertexID, "v1", initRequestedTime,
+ VertexFinishedEvent event = new VertexFinishedEvent(tezVertexID, "v1", 1,initRequestedTime,
initedTime, startRequestedTime, startTime, finishTime, VertexState.ERROR,
"diagnostics", null, vertexStats, taskStats);