You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@tez.apache.org by hi...@apache.org on 2013/09/16 23:56:20 UTC
git commit: TEZ-453. Fix umbilical heartbeat handshake with AM.
(hitesh)
Updated Branches:
refs/heads/TEZ-398 2d5302a4b -> 55a849332
TEZ-453. Fix umbilical heartbeat handshake with AM. (hitesh)
Project: http://git-wip-us.apache.org/repos/asf/incubator-tez/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-tez/commit/55a84933
Tree: http://git-wip-us.apache.org/repos/asf/incubator-tez/tree/55a84933
Diff: http://git-wip-us.apache.org/repos/asf/incubator-tez/diff/55a84933
Branch: refs/heads/TEZ-398
Commit: 55a8493324d54728184c13c4c67ff63a005a8c9e
Parents: 2d5302a
Author: Hitesh Shah <hi...@apache.org>
Authored: Mon Sep 16 14:55:56 2013 -0700
Committer: Hitesh Shah <hi...@apache.org>
Committed: Mon Sep 16 14:55:56 2013 -0700
----------------------------------------------------------------------
.../apache/hadoop/mapred/YarnTezDagChild.java | 6 ++++-
.../dag/app/TaskAttemptListenerImpTezDag.java | 16 +++++++++---
.../engine/newapi/impl/TezHeartbeatRequest.java | 26 +++++++++++++-------
3 files changed, 35 insertions(+), 13 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-tez/blob/55a84933/tez-dag/src/main/java/org/apache/hadoop/mapred/YarnTezDagChild.java
----------------------------------------------------------------------
diff --git a/tez-dag/src/main/java/org/apache/hadoop/mapred/YarnTezDagChild.java b/tez-dag/src/main/java/org/apache/hadoop/mapred/YarnTezDagChild.java
index 1a3905b..e988fa5 100644
--- a/tez-dag/src/main/java/org/apache/hadoop/mapred/YarnTezDagChild.java
+++ b/tez-dag/src/main/java/org/apache/hadoop/mapred/YarnTezDagChild.java
@@ -190,7 +190,11 @@ public class YarnTezDagChild {
}
try {
taskLock.readLock().lock();
- if (currentTask != null) {
+ if (taskAttemptID == null
+ || !taskAttemptID.equals(currentTaskAttemptID)) {
+ return;
+ }
+ if (currentTask != null && response.getEvents() != null) {
currentTask.handleEvents(response.getEvents());
}
} finally {
http://git-wip-us.apache.org/repos/asf/incubator-tez/blob/55a84933/tez-dag/src/main/java/org/apache/tez/dag/app/TaskAttemptListenerImpTezDag.java
----------------------------------------------------------------------
diff --git a/tez-dag/src/main/java/org/apache/tez/dag/app/TaskAttemptListenerImpTezDag.java b/tez-dag/src/main/java/org/apache/tez/dag/app/TaskAttemptListenerImpTezDag.java
index 3db7a1e..3c32d36 100644
--- a/tez-dag/src/main/java/org/apache/tez/dag/app/TaskAttemptListenerImpTezDag.java
+++ b/tez-dag/src/main/java/org/apache/tez/dag/app/TaskAttemptListenerImpTezDag.java
@@ -586,20 +586,30 @@ public class TaskAttemptListenerImpTezDag extends AbstractService implements
public TezHeartbeatResponse heartbeat(TezHeartbeatRequest request)
throws IOException, TezException {
long requestId = request.getRequestId();
+ LOG.info("Received request id " + requestId + " from child JVM");
TezTaskAttemptID taskAttemptID = request.getCurrentTaskAttemptID();
+ if (taskAttemptID == null) {
+ TezHeartbeatResponse response = new TezHeartbeatResponse();
+ response.setLastRequestId(requestId);
+ return response;
+ }
AttemptInfo attemptInfo = attemptToInfoMap.get(taskAttemptID);
if(attemptInfo == null) {
throw new TezException("Attempt " + taskAttemptID
+ " is not recognized for heartbeat");
}
- synchronized (attemptInfo) {
+ synchronized (attemptInfo) {
if(attemptInfo.lastRequestId == requestId) {
LOG.warn("Old sequenceId received: " + requestId + ", Re-sending last response to client");
return attemptInfo.lastReponse;
}
- if(attemptInfo.lastRequestId+1 < requestId) {
+ if(attemptInfo.lastRequestId != -1
+ && attemptInfo.lastRequestId+1 < requestId) {
+ // TODO NEWTEZ fix checking of last req id to ensure heartbeat works
+ // across attempts with container reuse
throw new TezException("Attempt " + taskAttemptID
- + " has invalid request id. Expected: " + attemptInfo.lastRequestId+1
+ + " has invalid request id. Expected: "
+ + attemptInfo.lastRequestId+1
+ " and actual: " + requestId);
}
http://git-wip-us.apache.org/repos/asf/incubator-tez/blob/55a84933/tez-engine/src/main/java/org/apache/tez/engine/newapi/impl/TezHeartbeatRequest.java
----------------------------------------------------------------------
diff --git a/tez-engine/src/main/java/org/apache/tez/engine/newapi/impl/TezHeartbeatRequest.java b/tez-engine/src/main/java/org/apache/tez/engine/newapi/impl/TezHeartbeatRequest.java
index e47f14b..d21de95 100644
--- a/tez-engine/src/main/java/org/apache/tez/engine/newapi/impl/TezHeartbeatRequest.java
+++ b/tez-engine/src/main/java/org/apache/tez/engine/newapi/impl/TezHeartbeatRequest.java
@@ -43,6 +43,7 @@ public class TezHeartbeatRequest implements Writable {
public TezHeartbeatRequest(long requestId, List<TezEvent> events,
TezTaskAttemptID taskAttemptID,
int startIndex, int maxEvents) {
+ this.requestId = requestId;
this.events = Collections.unmodifiableList(events);
this.startIndex = startIndex;
this.maxEvents = maxEvents;
@@ -71,9 +72,14 @@ public class TezHeartbeatRequest implements Writable {
@Override
public void write(DataOutput out) throws IOException {
- out.writeInt(events.size());
- for (TezEvent e : events) {
- e.write(out);
+ if (events != null) {
+ out.writeBoolean(true);
+ out.writeInt(events.size());
+ for (TezEvent e : events) {
+ e.write(out);
+ }
+ } else {
+ out.writeBoolean(false);
}
if (currentTaskAttemptID != null) {
out.writeBoolean(true);
@@ -88,12 +94,14 @@ public class TezHeartbeatRequest implements Writable {
@Override
public void readFields(DataInput in) throws IOException {
- int eventsCount = in.readInt();
- events = new ArrayList<TezEvent>(eventsCount);
- for (int i = 0; i < eventsCount; ++i) {
- TezEvent e = new TezEvent();
- e.readFields(in);
- events.add(e);
+ if (in.readBoolean()) {
+ int eventsCount = in.readInt();
+ events = new ArrayList<TezEvent>(eventsCount);
+ for (int i = 0; i < eventsCount; ++i) {
+ TezEvent e = new TezEvent();
+ e.readFields(in);
+ events.add(e);
+ }
}
if (in.readBoolean()) {
currentTaskAttemptID = new TezTaskAttemptID();