git commit: TAJO-40: Adopt YarnClient to GlobalEngine. (jihoon)

[ji...@apache.org]

Updated Branches:
  refs/heads/master ba0699c71 -> 16a0f2dfd


TAJO-40: Adopt YarnClient to GlobalEngine. (jihoon)


Project: http://git-wip-us.apache.org/repos/asf/incubator-tajo/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-tajo/commit/16a0f2df
Tree: http://git-wip-us.apache.org/repos/asf/incubator-tajo/tree/16a0f2df
Diff: http://git-wip-us.apache.org/repos/asf/incubator-tajo/diff/16a0f2df

Branch: refs/heads/master
Commit: 16a0f2dfd6f2735afe6d9b6277aaef16f1999d3e
Parents: ba0699c
Author: Jihoon Son <ji...@apache.org>
Authored: Mon Jun 17 15:49:12 2013 +0900
Committer: Jihoon Son <ji...@apache.org>
Committed: Mon Jun 17 15:49:12 2013 +0900

----------------------------------------------------------------------
 CHANGES.txt                                     |  2 ++
 .../src/main/java/tajo/master/GlobalEngine.java | 34 ++++++++------------
 2 files changed, 15 insertions(+), 21 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/incubator-tajo/blob/16a0f2df/CHANGES.txt
----------------------------------------------------------------------
diff --git a/CHANGES.txt b/CHANGES.txt
index dea57b3..990ae35 100644
--- a/CHANGES.txt
+++ b/CHANGES.txt
@@ -10,6 +10,8 @@ Release 0.2.0 - unreleased
 
   IMPROVEMENTS
 
+    TAJO-40: Adopt YarnClient to GlobalEngine. (jihoon)
+
     TAJO-69: Add .reviewboardrc file to the ignore lists of Rat and .gitignore. (hyunsik)
 
     TAJO-66: Support compiling and running Tajo with Java7 (Tejas Patil via hsaputra)

http://git-wip-us.apache.org/repos/asf/incubator-tajo/blob/16a0f2df/tajo-core/tajo-core-backend/src/main/java/tajo/master/GlobalEngine.java
----------------------------------------------------------------------
diff --git a/tajo-core/tajo-core-backend/src/main/java/tajo/master/GlobalEngine.java b/tajo-core/tajo-core-backend/src/main/java/tajo/master/GlobalEngine.java
index 080dbdd..2a52ef7 100644
--- a/tajo-core/tajo-core-backend/src/main/java/tajo/master/GlobalEngine.java
+++ b/tajo-core/tajo-core-backend/src/main/java/tajo/master/GlobalEngine.java
@@ -23,9 +23,10 @@ import org.apache.commons.logging.LogFactory;
 import org.apache.hadoop.conf.Configuration;
 import org.apache.hadoop.fs.FileSystem;
 import org.apache.hadoop.security.UserGroupInformation;
-import org.apache.hadoop.yarn.api.ClientRMProtocol;
-import org.apache.hadoop.yarn.api.protocolrecords.*;
+import org.apache.hadoop.yarn.api.protocolrecords.GetNewApplicationResponse;
 import org.apache.hadoop.yarn.api.records.*;
+import org.apache.hadoop.yarn.client.YarnClient;
+import org.apache.hadoop.yarn.client.YarnClientImpl;
 import org.apache.hadoop.yarn.conf.YarnConfiguration;
 import org.apache.hadoop.yarn.exceptions.YarnRemoteException;
 import org.apache.hadoop.yarn.factories.RecordFactory;
@@ -80,7 +81,7 @@ public class GlobalEngine extends AbstractService {
   // Yarn
   private final RecordFactory recordFactory =
       RecordFactoryProvider.getRecordFactory(null);
-  protected ClientRMProtocol rmClient;
+  protected YarnClient yarnClient;
   protected InetSocketAddress rmAddress;
 
   public GlobalEngine(final MasterContext context, final StorageManager sm)
@@ -92,7 +93,7 @@ public class GlobalEngine extends AbstractService {
 
   public void start() {
     try  {
-      connectRMClient();
+      connectYarnClient();
       analyzer = new QueryAnalyzer(context.getCatalog());
       planner = new LogicalPlanner(context.getCatalog());
 
@@ -108,6 +109,7 @@ public class GlobalEngine extends AbstractService {
 
   public void stop() {
     super.stop();
+    yarnClient.stop();
   }
 
   private String createTable(LogicalRootNode root) throws IOException {
@@ -221,9 +223,7 @@ public class GlobalEngine extends AbstractService {
 
     // Submit the application to the applications manager
     LOG.info("Submitting application to ASM");
-    SubmitApplicationRequest req = Records.newRecord(SubmitApplicationRequest.class);
-    req.setApplicationSubmissionContext(appContext);
-    rmClient.submitApplication(req);
+    yarnClient.submitApplication(appContext);
 
     // Monitor the application to wait for launch state
     ApplicationReport appReport = monitorApplication(appId,
@@ -272,12 +272,10 @@ public class GlobalEngine extends AbstractService {
     }
   }
 
-  private void connectRMClient() {
-    this.rmAddress = getRmAddress(getConfig());
-    this.rmClient =
-        (ClientRMProtocol) context.getYarnRPC().getProxy(ClientRMProtocol.class,
-            rmAddress, getConfig());
-    LOG.debug("Connecting to ResourceManager at " + rmAddress);
+  private void connectYarnClient() {
+    this.yarnClient = new YarnClientImpl();
+    this.yarnClient.init(getConfig());
+    this.yarnClient.start();
   }
 
   private static InetSocketAddress getRmAddress(Configuration conf) {
@@ -287,9 +285,7 @@ public class GlobalEngine extends AbstractService {
 
   public GetNewApplicationResponse getNewApplication()
       throws YarnRemoteException {
-    GetNewApplicationRequest request =
-        Records.newRecord(GetNewApplicationRequest.class);
-    return rmClient.getNewApplication(request);
+    return yarnClient.getNewApplication();
   }
 
   /**
@@ -314,11 +310,7 @@ public class GlobalEngine extends AbstractService {
       }
 
       // Get application report for the appId we are interested in
-      GetApplicationReportRequest req = Records.newRecord(GetApplicationReportRequest.class);
-      req.setApplicationId(appId);
-      GetApplicationReportResponse res = rmClient.getApplicationReport(req);
-      ApplicationReport report = res.getApplicationReport();
-
+      ApplicationReport report = yarnClient.getApplicationReport(appId);
 
       LOG.info("Got application report from ASM for" + ", appId="
           + appId.getId() + ", appAttemptId="