Return-Path: X-Original-To: apmail-tez-commits-archive@minotaur.apache.org Delivered-To: apmail-tez-commits-archive@minotaur.apache.org Received: from mail.apache.org (hermes.apache.org [140.211.11.3]) by minotaur.apache.org (Postfix) with SMTP id 47B291843F for ; Wed, 6 May 2015 07:41:05 +0000 (UTC) Received: (qmail 16821 invoked by uid 500); 6 May 2015 07:41:05 -0000 Delivered-To: apmail-tez-commits-archive@tez.apache.org Received: (qmail 16739 invoked by uid 500); 6 May 2015 07:41:05 -0000 Mailing-List: contact commits-help@tez.apache.org; run by ezmlm Precedence: bulk List-Help: List-Unsubscribe: List-Post: List-Id: Reply-To: dev@tez.apache.org Delivered-To: mailing list commits@tez.apache.org Received: (qmail 16314 invoked by uid 99); 6 May 2015 07:41:04 -0000 Received: from git1-us-west.apache.org (HELO git1-us-west.apache.org) (140.211.11.23) by apache.org (qpsmtpd/0.29) with ESMTP; Wed, 06 May 2015 07:41:04 +0000 Received: by git1-us-west.apache.org (ASF Mail Server at git1-us-west.apache.org, from userid 33) id C9F55E17C1; Wed, 6 May 2015 07:41:04 +0000 (UTC) Content-Type: text/plain; charset="us-ascii" MIME-Version: 1.0 Content-Transfer-Encoding: 7bit From: sseth@apache.org To: commits@tez.apache.org Date: Wed, 06 May 2015 07:41:45 -0000 Message-Id: In-Reply-To: <3cb91b30f7424e8bae96b62852f5fb15@git.apache.org> References: <3cb91b30f7424e8bae96b62852f5fb15@git.apache.org> X-Mailer: ASF-Git Admin Mailer Subject: [43/50] [abbrv] tez git commit: TEZ-2285. Allow TaskCommunicators to indicate task/container liveness. (sseth) TEZ-2285. Allow TaskCommunicators to indicate task/container liveness. (sseth) Project: http://git-wip-us.apache.org/repos/asf/tez/repo Commit: http://git-wip-us.apache.org/repos/asf/tez/commit/2f8a2735 Tree: http://git-wip-us.apache.org/repos/asf/tez/tree/2f8a2735 Diff: http://git-wip-us.apache.org/repos/asf/tez/diff/2f8a2735 Branch: refs/heads/TEZ-2003 Commit: 2f8a2735c8ed3ac35c6137b0a3f85ba30cad3a87 Parents: 854a439 Author: Siddharth Seth Authored: Tue Apr 7 13:22:09 2015 -0700 Committer: Siddharth Seth Committed: Wed May 6 00:18:08 2015 -0700 ---------------------------------------------------------------------- TEZ-2003-CHANGES.txt | 1 + .../apache/tez/dag/api/TaskCommunicatorContext.java | 4 ++++ .../tez/dag/app/TaskAttemptListenerImpTezDag.java | 10 ++++++++++ .../apache/tez/dag/app/TezTaskCommunicatorImpl.java | 16 +++++++++------- 4 files changed, 24 insertions(+), 7 deletions(-) ---------------------------------------------------------------------- http://git-wip-us.apache.org/repos/asf/tez/blob/2f8a2735/TEZ-2003-CHANGES.txt ---------------------------------------------------------------------- diff --git a/TEZ-2003-CHANGES.txt b/TEZ-2003-CHANGES.txt index e2c428d..9d6b220 100644 --- a/TEZ-2003-CHANGES.txt +++ b/TEZ-2003-CHANGES.txt @@ -14,5 +14,6 @@ ALL CHANGES: TEZ-2241. Miscellaneous fixes after last reabse. TEZ-2283. Fixes after rebase 04/07. TEZ-2284. Separate TaskReporter into an interface. + TEZ-2285. Allow TaskCommunicators to indicate task/container liveness. INCOMPATIBLE CHANGES: http://git-wip-us.apache.org/repos/asf/tez/blob/2f8a2735/tez-dag/src/main/java/org/apache/tez/dag/api/TaskCommunicatorContext.java ---------------------------------------------------------------------- diff --git a/tez-dag/src/main/java/org/apache/tez/dag/api/TaskCommunicatorContext.java b/tez-dag/src/main/java/org/apache/tez/dag/api/TaskCommunicatorContext.java index a85fb7f..0c3bac3 100644 --- a/tez-dag/src/main/java/org/apache/tez/dag/api/TaskCommunicatorContext.java +++ b/tez-dag/src/main/java/org/apache/tez/dag/api/TaskCommunicatorContext.java @@ -43,6 +43,10 @@ public interface TaskCommunicatorContext { boolean isKnownContainer(ContainerId containerId); + void taskAlive(TezTaskAttemptID taskAttemptId); + + void containerAlive(ContainerId containerId); + // TODO TEZ-2003 Move to vertex, taskIndex, version. Rename to taskAttempt* void taskStartedRemotely(TezTaskAttemptID taskAttemptID, ContainerId containerId); http://git-wip-us.apache.org/repos/asf/tez/blob/2f8a2735/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 aaaa01c..a6ccbfa 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 @@ -259,6 +259,16 @@ public class TaskAttemptListenerImpTezDag extends AbstractService implements } @Override + public void taskAlive(TezTaskAttemptID taskAttemptId) { + taskHeartbeatHandler.pinged(taskAttemptId); + } + + @Override + public void containerAlive(ContainerId containerId) { + pingContainerHeartbeatHandler(containerId); + } + + @Override public void taskStartedRemotely(TezTaskAttemptID taskAttemptID, ContainerId containerId) { context.getEventHandler() .handle(new TaskAttemptEventStartedRemotely(taskAttemptID, containerId, null)); http://git-wip-us.apache.org/repos/asf/tez/blob/2f8a2735/tez-dag/src/main/java/org/apache/tez/dag/app/TezTaskCommunicatorImpl.java ---------------------------------------------------------------------- diff --git a/tez-dag/src/main/java/org/apache/tez/dag/app/TezTaskCommunicatorImpl.java b/tez-dag/src/main/java/org/apache/tez/dag/app/TezTaskCommunicatorImpl.java index 9d0c031..ef4f764 100644 --- a/tez-dag/src/main/java/org/apache/tez/dag/app/TezTaskCommunicatorImpl.java +++ b/tez-dag/src/main/java/org/apache/tez/dag/app/TezTaskCommunicatorImpl.java @@ -65,17 +65,19 @@ public class TezTaskCommunicatorImpl extends TaskCommunicator { null, true, null, null, false); private final TaskCommunicatorContext taskCommunicatorContext; + private final TezTaskUmbilicalProtocol taskUmbilical; - private final ConcurrentMap registeredContainers = + protected final ConcurrentMap registeredContainers = new ConcurrentHashMap(); - private final ConcurrentMap attemptToContainerMap = + protected final ConcurrentMap attemptToContainerMap = new ConcurrentHashMap(); - private final TezTaskUmbilicalProtocol taskUmbilical; - private final String tokenIdentifier; - private final Token sessionToken; + + protected final String tokenIdentifier; + protected final Token sessionToken; protected InetSocketAddress address; - private Server server; + + protected volatile Server server; public static final class ContainerInfo { @@ -438,7 +440,7 @@ public class TezTaskCommunicatorImpl extends TaskCommunicator { // Holder for Task information, which eventually will likely be VertexImplm taskIndex, attemptIndex - private static class TaskAttempt { + protected static class TaskAttempt { // TODO TEZ-2003 Change this to work with VertexName, int id, int version // TODO TEZ-2003 Avoid constructing this unit all over the place private TezTaskAttemptID taskAttemptId;