You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@tajo.apache.org by ji...@apache.org on 2013/12/28 07:36:31 UTC
[34/50] [abbrv] git commit: TAJO-135: Bump up hadoop to 2.2.0.
(jihoon)
TAJO-135: Bump up hadoop to 2.2.0. (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/ecc749e8
Tree: http://git-wip-us.apache.org/repos/asf/incubator-tajo/tree/ecc749e8
Diff: http://git-wip-us.apache.org/repos/asf/incubator-tajo/diff/ecc749e8
Branch: refs/heads/DAG-execplan
Commit: ecc749e811b9f978c9d9bb2cd9bfee6f795653df
Parents: 074a341
Author: Jihoon Son <ji...@apache.org>
Authored: Mon Dec 23 18:55:34 2013 +0900
Committer: Jihoon Son <ji...@apache.org>
Committed: Mon Dec 23 18:55:34 2013 +0900
----------------------------------------------------------------------
BUILDING.txt | 2 +-
CHANGES.txt | 2 +
.../src/main/proto/CatalogProtos.proto | 4 +-
.../org/apache/tajo/catalog/CatalogServer.java | 2 +-
tajo-common/src/main/proto/Security.proto | 6 +
tajo-common/src/main/proto/yarn_protos.proto | 177 +++++++++----------
tajo-core/tajo-core-backend/pom.xml | 1 -
.../tajo/master/AbstractTaskScheduler.java | 2 +-
.../org/apache/tajo/master/GlobalEngine.java | 2 +-
.../apache/tajo/master/TajoAsyncDispatcher.java | 6 +-
.../java/org/apache/tajo/master/TajoMaster.java | 13 +-
.../tajo/master/TajoMasterClientService.java | 2 +-
.../apache/tajo/master/TajoMasterService.java | 2 +-
.../apache/tajo/master/YarnContainerProxy.java | 64 ++++---
.../tajo/master/YarnTaskRunnerLauncherImpl.java | 2 +-
.../apache/tajo/master/querymaster/Query.java | 2 +-
.../master/querymaster/QueryInProgress.java | 6 +-
.../master/querymaster/QueryJobManager.java | 4 +-
.../tajo/master/querymaster/QueryMaster.java | 13 +-
.../querymaster/QueryMasterManagerService.java | 2 +-
.../master/querymaster/QueryMasterRunner.java | 2 +-
.../master/querymaster/QueryMasterTask.java | 4 +-
.../tajo/master/rm/TajoWorkerContainer.java | 26 +--
.../tajo/master/rm/TajoWorkerContainerId.java | 5 +
.../master/rm/TajoWorkerResourceManager.java | 8 +-
.../master/rm/YarnRMContainerAllocator.java | 31 ++--
.../tajo/master/rm/YarnTajoResourceManager.java | 63 +++----
.../apache/tajo/util/ApplicationIdUtils.java | 2 +-
.../tajo/worker/AbstractResourceAllocator.java | 2 +-
.../tajo/worker/TajoResourceAllocator.java | 8 +-
.../java/org/apache/tajo/worker/TajoWorker.java | 2 +-
.../tajo/worker/TajoWorkerClientService.java | 2 +-
.../tajo/worker/TajoWorkerManagerService.java | 2 +-
.../java/org/apache/tajo/worker/TaskRunner.java | 2 +-
.../apache/tajo/worker/TaskRunnerManager.java | 2 +-
.../tajo/worker/YarnResourceAllocator.java | 7 +-
.../src/main/proto/InternalTypes.proto | 2 +-
.../src/main/proto/TajoMasterProtocol.proto | 4 +-
.../src/main/proto/TajoWorkerProtocol.proto | 2 +-
.../org/apache/tajo/MiniTajoYarnCluster.java | 10 +-
.../org/apache/tajo/TajoTestingCluster.java | 3 +-
.../test/java/org/apache/tajo/TestTajoIds.java | 2 +-
.../tajo/pullserver/FadvisedChunkedFile.java | 6 +-
.../tajo/pullserver/FadvisedFileRegion.java | 6 +-
.../tajo/pullserver/PullServerAuxService.java | 22 +--
.../tajo/pullserver/TajoPullServerService.java | 2 +-
tajo-project/pom.xml | 6 +-
47 files changed, 263 insertions(+), 284 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-tajo/blob/ecc749e8/BUILDING.txt
----------------------------------------------------------------------
diff --git a/BUILDING.txt b/BUILDING.txt
index cceacc1..8dcc179 100644
--- a/BUILDING.txt
+++ b/BUILDING.txt
@@ -6,7 +6,7 @@ Requirements:
* Unix System
* JDK 1.6 or higher
* Maven 3.0 or higher
-* Protocol Buffers 2.4.1
+* Protocol Buffers 2.5.0
* Internet connection for first build (to fetch all Maven and Tajo dependencies)
--------------------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-tajo/blob/ecc749e8/CHANGES.txt
----------------------------------------------------------------------
diff --git a/CHANGES.txt b/CHANGES.txt
index 602f6e8..5593cd3 100644
--- a/CHANGES.txt
+++ b/CHANGES.txt
@@ -103,6 +103,8 @@ Release 0.8.0 - unreleased
IMPROVEMENTS
+ TAJO-135: Bump up hadoop to 2.2.0. (jihoon)
+
TAJO-435: Improve intermediate file. (jinho)
TAJO-424: Make serializer/deserializer configurable in CSVFile. (jinho)
http://git-wip-us.apache.org/repos/asf/incubator-tajo/blob/ecc749e8/tajo-catalog/tajo-catalog-common/src/main/proto/CatalogProtos.proto
----------------------------------------------------------------------
diff --git a/tajo-catalog/tajo-catalog-common/src/main/proto/CatalogProtos.proto b/tajo-catalog/tajo-catalog-common/src/main/proto/CatalogProtos.proto
index ad6ef3e..2d5119d 100644
--- a/tajo-catalog/tajo-catalog-common/src/main/proto/CatalogProtos.proto
+++ b/tajo-catalog/tajo-catalog-common/src/main/proto/CatalogProtos.proto
@@ -203,8 +203,8 @@ enum StatType {
TABLE_AVG_ROWS = 3;
TABLE_NUM_BYTES = 4;
- COLUMN_NUM_NDV = 4;
- COLUMN_NUM_NULLS = 5;
+ COLUMN_NUM_NDV = 5;
+ COLUMN_NUM_NULLS = 6;
}
message StatSetProto {
http://git-wip-us.apache.org/repos/asf/incubator-tajo/blob/ecc749e8/tajo-catalog/tajo-catalog-server/src/main/java/org/apache/tajo/catalog/CatalogServer.java
----------------------------------------------------------------------
diff --git a/tajo-catalog/tajo-catalog-server/src/main/java/org/apache/tajo/catalog/CatalogServer.java b/tajo-catalog/tajo-catalog-server/src/main/java/org/apache/tajo/catalog/CatalogServer.java
index e6566af..6c16558 100644
--- a/tajo-catalog/tajo-catalog-server/src/main/java/org/apache/tajo/catalog/CatalogServer.java
+++ b/tajo-catalog/tajo-catalog-server/src/main/java/org/apache/tajo/catalog/CatalogServer.java
@@ -23,7 +23,7 @@ import com.google.protobuf.ServiceException;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.hadoop.conf.Configuration;
-import org.apache.hadoop.yarn.service.AbstractService;
+import org.apache.hadoop.service.AbstractService;
import org.apache.tajo.catalog.CatalogProtocol.CatalogProtocolService;
import org.apache.tajo.catalog.exception.*;
import org.apache.tajo.catalog.proto.CatalogProtos.*;
http://git-wip-us.apache.org/repos/asf/incubator-tajo/blob/ecc749e8/tajo-common/src/main/proto/Security.proto
----------------------------------------------------------------------
diff --git a/tajo-common/src/main/proto/Security.proto b/tajo-common/src/main/proto/Security.proto
index 961ee9d..5ff571d 100644
--- a/tajo-common/src/main/proto/Security.proto
+++ b/tajo-common/src/main/proto/Security.proto
@@ -16,6 +16,12 @@
* limitations under the License.
*/
+/**
+ * These .proto interfaces are private and stable.
+ * Please see http://wiki.apache.org/hadoop/Compatibility
+ * for what changes are allowed for a *stable* .proto interface.
+ */
+
option java_package = "org.apache.hadoop.security.proto";
option java_outer_classname = "SecurityProtos";
option java_generic_services = true;
http://git-wip-us.apache.org/repos/asf/incubator-tajo/blob/ecc749e8/tajo-common/src/main/proto/yarn_protos.proto
----------------------------------------------------------------------
diff --git a/tajo-common/src/main/proto/yarn_protos.proto b/tajo-common/src/main/proto/yarn_protos.proto
index 3fe519f..3192306 100644
--- a/tajo-common/src/main/proto/yarn_protos.proto
+++ b/tajo-common/src/main/proto/yarn_protos.proto
@@ -16,18 +16,25 @@
* limitations under the License.
*/
+/**
+ * These .proto interfaces are public and stable.
+ * Please see http://wiki.apache.org/hadoop/Compatibility
+ * for what changes are allowed for a *stable* .proto interface.
+ */
+
option java_package = "org.apache.hadoop.yarn.proto";
option java_outer_classname = "YarnProtos";
option java_generic_services = true;
option java_generate_equals_and_hash = true;
+package hadoop.yarn;
import "Security.proto";
-message YarnRemoteExceptionProto {
+message SerializedExceptionProto {
optional string message = 1;
optional string trace = 2;
optional string class_name = 3;
- optional YarnRemoteExceptionProto cause = 4;
+ optional SerializedExceptionProto cause = 4;
}
message ApplicationIdProto {
@@ -67,19 +74,18 @@ message ContainerProto {
optional string node_http_address = 3;
optional ResourceProto resource = 4;
optional PriorityProto priority = 5;
- optional ContainerStateProto state = 6;
- optional hadoop.common.TokenProto container_token = 7;
- optional ContainerStatusProto container_status = 8;
+ optional hadoop.common.TokenProto container_token = 6;
}
enum YarnApplicationStateProto {
NEW = 1;
- SUBMITTED = 2;
- RUNNING = 3;
- FINISHED = 4;
- FAILED = 5;
- KILLED = 6;
- ACCEPTED = 7;
+ NEW_SAVING = 2;
+ SUBMITTED = 3;
+ ACCEPTED = 4;
+ RUNNING = 5;
+ FINISHED = 6;
+ FAILED = 7;
+ KILLED = 8;
}
enum FinalApplicationStatusProto {
@@ -89,30 +95,12 @@ enum FinalApplicationStatusProto {
APP_KILLED = 3;
}
-message ApplicationStatusProto {
- optional int32 response_id = 1;
- optional ApplicationAttemptIdProto application_attempt_id = 2;
- optional float progress = 3;
-}
-
-message ApplicationMasterProto {
- optional ApplicationIdProto application_id = 1;
- optional string host = 2;
- optional int32 rpc_port = 3;
- optional string trackingUrl = 4;
- optional ApplicationStatusProto status = 5;
- optional YarnApplicationStateProto state = 6;
- optional hadoop.common.TokenProto client_token = 7;
- optional int32 containerCount = 8;
- optional int32 amFailCount = 9;
- optional string diagnostics = 10 [default = ""];
-}
-
message URLProto {
optional string scheme = 1;
optional string host = 2;
optional int32 port = 3;
optional string file = 4;
+ optional string userInfo = 5;
}
enum LocalResourceVisibilityProto {
@@ -151,18 +139,19 @@ message ApplicationReportProto {
optional string name = 4;
optional string host = 5;
optional int32 rpc_port = 6;
- optional hadoop.common.TokenProto client_token = 7;
- optional ApplicationStatusProto status = 8;
- optional YarnApplicationStateProto yarn_application_state = 9;
- optional ContainerProto masterContainer = 10;
- optional string trackingUrl = 11;
- optional string diagnostics = 12 [default = "N/A"];
- optional int64 startTime = 13;
- optional int64 finishTime = 14;
- optional FinalApplicationStatusProto final_application_status = 15;
- optional ApplicationResourceUsageReportProto app_resource_Usage = 16;
- optional string originalTrackingUrl = 17;
- optional ApplicationAttemptIdProto currentApplicationAttemptId = 18;
+ optional hadoop.common.TokenProto client_to_am_token = 7;
+ optional YarnApplicationStateProto yarn_application_state = 8;
+ optional string trackingUrl = 9;
+ optional string diagnostics = 10 [default = "N/A"];
+ optional int64 startTime = 11;
+ optional int64 finishTime = 12;
+ optional FinalApplicationStatusProto final_application_status = 13;
+ optional ApplicationResourceUsageReportProto app_resource_Usage = 14;
+ optional string originalTrackingUrl = 15;
+ optional ApplicationAttemptIdProto currentApplicationAttemptId = 16;
+ optional float progress = 17;
+ optional string applicationType = 18;
+ optional hadoop.common.TokenProto am_rm_token = 19;
}
enum NodeStateProto {
@@ -179,12 +168,6 @@ message NodeIdProto {
optional int32 port = 2;
}
-message NodeHealthStatusProto {
- optional bool is_node_healthy = 1;
- optional string health_report = 2;
- optional int64 last_health_report_time = 3;
-}
-
message NodeReportProto {
optional NodeIdProto nodeId = 1;
optional string httpAddress = 2;
@@ -192,8 +175,9 @@ message NodeReportProto {
optional ResourceProto used = 4;
optional ResourceProto capability = 5;
optional int32 numContainers = 6;
- optional NodeHealthStatusProto node_health_status = 8;
- optional NodeStateProto node_state = 9;
+ optional NodeStateProto node_state = 7;
+ optional string health_report = 8;
+ optional int64 last_health_report_time = 9;
}
@@ -202,20 +186,43 @@ message NodeReportProto {
////////////////////////////////////////////////////////////////////////
message ResourceRequestProto {
optional PriorityProto priority = 1;
- optional string host_name = 2;
+ optional string resource_name = 2;
optional ResourceProto capability = 3;
optional int32 num_containers = 4;
+ optional bool relax_locality = 5 [default = true];
+}
+
+enum AMCommandProto {
+ AM_RESYNC = 1;
+ AM_SHUTDOWN = 2;
+}
+
+message PreemptionMessageProto {
+ optional StrictPreemptionContractProto strictContract = 1;
+ optional PreemptionContractProto contract = 2;
}
-message AMResponseProto {
- optional bool reboot = 1;
- optional int32 response_id = 2;
- repeated ContainerProto allocated_containers = 3;
- repeated ContainerStatusProto completed_container_statuses = 4;
- optional ResourceProto limit = 5;
- repeated NodeReportProto updated_nodes = 6;
+message StrictPreemptionContractProto {
+ repeated PreemptionContainerProto container = 1;
}
+message PreemptionContractProto {
+ repeated PreemptionResourceRequestProto resource = 1;
+ repeated PreemptionContainerProto container = 2;
+}
+
+message PreemptionContainerProto {
+ optional ContainerIdProto id = 1;
+}
+
+message PreemptionResourceRequestProto {
+ optional ResourceRequestProto resource = 1;
+}
+
+message ResourceBlacklistRequestProto {
+ repeated string blacklist_additions = 1;
+ repeated string blacklist_removals = 2;
+}
////////////////////////////////////////////////////////////////////////
////// From client_RM_Protocol /////////////////////////////////////////
@@ -223,12 +230,14 @@ message AMResponseProto {
message ApplicationSubmissionContextProto {
optional ApplicationIdProto application_id = 1;
optional string application_name = 2 [default = "N/A"];
- optional string user = 3;
- optional string queue = 4 [default = "default"];
- optional PriorityProto priority = 5;
- optional ContainerLaunchContextProto am_container_spec = 6;
- optional bool cancel_tokens_when_complete = 7 [default = true];
- optional bool unmanaged_am = 8 [default = false];
+ optional string queue = 3 [default = "default"];
+ optional PriorityProto priority = 4;
+ optional ContainerLaunchContextProto am_container_spec = 5;
+ optional bool cancel_tokens_when_complete = 6 [default = true];
+ optional bool unmanaged_am = 7 [default = false];
+ optional int32 maxAppAttempts = 8 [default = 0];
+ optional ResourceProto resource = 9;
+ optional string applicationType = 10 [default = "YARN"];
}
enum ApplicationAccessTypeProto {
@@ -275,15 +284,12 @@ message QueueUserACLInfoProto {
////////////////////////////////////////////////////////////////////////
message ContainerLaunchContextProto {
- optional ContainerIdProto container_id = 1;
- optional string user = 2;
- optional ResourceProto resource = 3;
- repeated StringLocalResourceMapProto localResources = 4;
- optional bytes container_tokens = 5;
- repeated StringBytesMapProto service_data = 6;
- repeated StringStringMapProto environment = 7;
- repeated string command = 8;
- repeated ApplicationACLMapProto application_ACLs = 9;
+ repeated StringLocalResourceMapProto localResources = 1;
+ optional bytes tokens = 2;
+ repeated StringBytesMapProto service_data = 3;
+ repeated StringStringMapProto environment = 4;
+ repeated string command = 5;
+ repeated ApplicationACLMapProto application_ACLs = 6;
}
message ContainerStatusProto {
@@ -293,16 +299,16 @@ message ContainerStatusProto {
optional int32 exit_status = 4 [default = -1000];
}
-
+enum ContainerExitStatusProto {
+ SUCCESS = 0;
+ INVALID = -1000;
+ ABORTED = -100;
+ DISKS_FAILED = -101;
+}
////////////////////////////////////////////////////////////////////////
////// From common//////////////////////////////////////////////////////
////////////////////////////////////////////////////////////////////////
-message StringURLMapProto {
- optional string key = 1;
- optional URLProto value = 2;
-}
-
message StringLocalResourceMapProto {
optional string key = 1;
optional LocalResourceProto value = 2;
@@ -317,16 +323,3 @@ message StringBytesMapProto {
optional string key = 1;
optional bytes value = 2;
}
-
-////////////////////////////////////////////////////////////////////////
-////// From recovery////////////////////////////////////////////////////
-////////////////////////////////////////////////////////////////////////
-message ApplicationStateDataProto {
- optional int64 submit_time = 1;
- optional ApplicationSubmissionContextProto application_submission_context = 2;
-}
-
-message ApplicationAttemptStateDataProto {
- optional ApplicationAttemptIdProto attemptId = 1;
- optional ContainerProto master_container = 2;
-}
http://git-wip-us.apache.org/repos/asf/incubator-tajo/blob/ecc749e8/tajo-core/tajo-core-backend/pom.xml
----------------------------------------------------------------------
diff --git a/tajo-core/tajo-core-backend/pom.xml b/tajo-core/tajo-core-backend/pom.xml
index 07117a6..ff59935 100644
--- a/tajo-core/tajo-core-backend/pom.xml
+++ b/tajo-core/tajo-core-backend/pom.xml
@@ -261,7 +261,6 @@
<dependency>
<groupId>org.apache.hadoop</groupId>
<artifactId>hadoop-yarn-server-common</artifactId>
- <scope>test</scope>
</dependency>
<dependency>
<groupId>org.apache.hadoop</groupId>
http://git-wip-us.apache.org/repos/asf/incubator-tajo/blob/ecc749e8/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/master/AbstractTaskScheduler.java
----------------------------------------------------------------------
diff --git a/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/master/AbstractTaskScheduler.java b/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/master/AbstractTaskScheduler.java
index ed2529a..3f4998a 100644
--- a/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/master/AbstractTaskScheduler.java
+++ b/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/master/AbstractTaskScheduler.java
@@ -18,8 +18,8 @@
package org.apache.tajo.master;
+import org.apache.hadoop.service.AbstractService;
import org.apache.hadoop.yarn.event.EventHandler;
-import org.apache.hadoop.yarn.service.AbstractService;
import org.apache.tajo.master.event.TaskRequestEvent;
import org.apache.tajo.master.event.TaskSchedulerEvent;
http://git-wip-us.apache.org/repos/asf/incubator-tajo/blob/ecc749e8/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/master/GlobalEngine.java
----------------------------------------------------------------------
diff --git a/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/master/GlobalEngine.java b/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/master/GlobalEngine.java
index 4575bf8..d17ec91 100644
--- a/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/master/GlobalEngine.java
+++ b/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/master/GlobalEngine.java
@@ -23,8 +23,8 @@ import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.service.AbstractService;
import org.apache.hadoop.util.StringUtils;
-import org.apache.hadoop.yarn.service.AbstractService;
import org.apache.tajo.QueryId;
import org.apache.tajo.QueryIdFactory;
import org.apache.tajo.TajoProtos;
http://git-wip-us.apache.org/repos/asf/incubator-tajo/blob/ecc749e8/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/master/TajoAsyncDispatcher.java
----------------------------------------------------------------------
diff --git a/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/master/TajoAsyncDispatcher.java b/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/master/TajoAsyncDispatcher.java
index 5875a6c..751b21b 100644
--- a/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/master/TajoAsyncDispatcher.java
+++ b/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/master/TajoAsyncDispatcher.java
@@ -21,12 +21,12 @@ package org.apache.tajo.master;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.service.AbstractService;
import org.apache.hadoop.util.ShutdownHookManager;
-import org.apache.hadoop.yarn.YarnException;
import org.apache.hadoop.yarn.event.Dispatcher;
import org.apache.hadoop.yarn.event.Event;
import org.apache.hadoop.yarn.event.EventHandler;
-import org.apache.hadoop.yarn.service.AbstractService;
+import org.apache.hadoop.yarn.exceptions.YarnRuntimeException;
import java.util.ArrayList;
import java.util.HashMap;
@@ -201,7 +201,7 @@ public class TajoAsyncDispatcher extends AbstractService implements Dispatcher
if (!stopped) {
LOG.warn("AsyncDispatcher thread interrupted", e);
}
- throw new YarnException(e);
+ throw new YarnRuntimeException(e);
}
}
}
http://git-wip-us.apache.org/repos/asf/incubator-tajo/blob/ecc749e8/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/master/TajoMaster.java
----------------------------------------------------------------------
diff --git a/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/master/TajoMaster.java b/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/master/TajoMaster.java
index 0f036bb..40277cb 100644
--- a/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/master/TajoMaster.java
+++ b/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/master/TajoMaster.java
@@ -27,16 +27,15 @@ import org.apache.hadoop.fs.FSDataOutputStream;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.fs.permission.FsPermission;
+import org.apache.hadoop.service.CompositeService;
import org.apache.hadoop.util.ShutdownHookManager;
import org.apache.hadoop.util.StringUtils;
-import org.apache.hadoop.yarn.Clock;
-import org.apache.hadoop.yarn.SystemClock;
import org.apache.hadoop.yarn.conf.YarnConfiguration;
import org.apache.hadoop.yarn.event.AsyncDispatcher;
import org.apache.hadoop.yarn.event.EventHandler;
-import org.apache.hadoop.yarn.service.CompositeService;
-import org.apache.hadoop.yarn.service.Service;
+import org.apache.hadoop.yarn.util.Clock;
import org.apache.hadoop.yarn.util.RackResolver;
+import org.apache.hadoop.yarn.util.SystemClock;
import org.apache.tajo.TajoConstants;
import org.apache.tajo.catalog.*;
import org.apache.tajo.catalog.proto.CatalogProtos.FunctionType;
@@ -842,12 +841,6 @@ public class TajoMaster extends CompositeService {
return this.context;
}
- protected void addIfService(Object object) {
- if (object instanceof Service) {
- addService((Service) object);
- }
- }
-
@Override
public void start() {
LOG.info("TajoMaster startup");
http://git-wip-us.apache.org/repos/asf/incubator-tajo/blob/ecc749e8/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/master/TajoMasterClientService.java
----------------------------------------------------------------------
diff --git a/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/master/TajoMasterClientService.java b/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/master/TajoMasterClientService.java
index 861147a..3e9720f 100644
--- a/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/master/TajoMasterClientService.java
+++ b/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/master/TajoMasterClientService.java
@@ -26,7 +26,7 @@ import org.apache.commons.logging.LogFactory;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.security.UserGroupInformation;
-import org.apache.hadoop.yarn.service.AbstractService;
+import org.apache.hadoop.service.AbstractService;
import org.apache.tajo.QueryId;
import org.apache.tajo.QueryIdFactory;
import org.apache.tajo.TajoIdProtos;
http://git-wip-us.apache.org/repos/asf/incubator-tajo/blob/ecc749e8/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/master/TajoMasterService.java
----------------------------------------------------------------------
diff --git a/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/master/TajoMasterService.java b/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/master/TajoMasterService.java
index aac4114..07a4d59 100644
--- a/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/master/TajoMasterService.java
+++ b/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/master/TajoMasterService.java
@@ -22,8 +22,8 @@ import com.google.protobuf.RpcCallback;
import com.google.protobuf.RpcController;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
+import org.apache.hadoop.service.AbstractService;
import org.apache.hadoop.yarn.proto.YarnProtos;
-import org.apache.hadoop.yarn.service.AbstractService;
import org.apache.tajo.ExecutionBlockId;
import org.apache.tajo.QueryId;
import org.apache.tajo.TajoIdProtos;
http://git-wip-us.apache.org/repos/asf/incubator-tajo/blob/ecc749e8/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/master/YarnContainerProxy.java
----------------------------------------------------------------------
diff --git a/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/master/YarnContainerProxy.java b/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/master/YarnContainerProxy.java
index 4dea476..dbec652 100644
--- a/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/master/YarnContainerProxy.java
+++ b/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/master/YarnContainerProxy.java
@@ -22,22 +22,21 @@ import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.*;
import org.apache.hadoop.fs.permission.FsPermission;
import org.apache.hadoop.security.UserGroupInformation;
-import org.apache.hadoop.security.token.Token;
import org.apache.hadoop.util.StringUtils;
import org.apache.hadoop.yarn.api.ApplicationConstants;
-import org.apache.hadoop.yarn.api.ContainerManager;
+import org.apache.hadoop.yarn.api.ContainerManagementProtocol;
import org.apache.hadoop.yarn.api.protocolrecords.StartContainerRequest;
-import org.apache.hadoop.yarn.api.protocolrecords.StartContainerResponse;
-import org.apache.hadoop.yarn.api.protocolrecords.StopContainerRequest;
+import org.apache.hadoop.yarn.api.protocolrecords.StartContainersRequest;
+import org.apache.hadoop.yarn.api.protocolrecords.StartContainersResponse;
+import org.apache.hadoop.yarn.api.protocolrecords.StopContainersRequest;
import org.apache.hadoop.yarn.api.records.*;
import org.apache.hadoop.yarn.conf.YarnConfiguration;
import org.apache.hadoop.yarn.factories.RecordFactory;
import org.apache.hadoop.yarn.factory.providers.RecordFactoryProvider;
import org.apache.hadoop.yarn.ipc.YarnRPC;
import org.apache.hadoop.yarn.security.ContainerTokenIdentifier;
-import org.apache.hadoop.yarn.util.BuilderUtils;
+import org.apache.hadoop.yarn.server.utils.BuilderUtils;
import org.apache.hadoop.yarn.util.ConverterUtils;
-import org.apache.hadoop.yarn.util.ProtoUtils;
import org.apache.hadoop.yarn.util.Records;
import org.apache.tajo.ExecutionBlockId;
import org.apache.tajo.TajoConstants;
@@ -57,7 +56,7 @@ public class YarnContainerProxy extends ContainerProxy {
protected final YarnRPC yarnRPC;
final protected String containerMgrAddress;
- protected ContainerToken containerToken;
+ protected Token containerToken;
public YarnContainerProxy(QueryMasterTask.QueryMasterTaskContext context, Configuration conf, YarnRPC yarnRPC,
Container container, ExecutionBlockId executionBlockId) {
@@ -69,9 +68,9 @@ public class YarnContainerProxy extends ContainerProxy {
this.containerToken = container.getContainerToken();
}
- protected ContainerManager getCMProxy(ContainerId containerID,
- final String containerManagerBindAddr,
- ContainerToken containerToken)
+ protected ContainerManagementProtocol getCMProxy(ContainerId containerID,
+ final String containerManagerBindAddr,
+ Token containerToken)
throws IOException {
String [] hosts = containerManagerBindAddr.split(":");
final InetSocketAddress cmAddr =
@@ -79,17 +78,17 @@ public class YarnContainerProxy extends ContainerProxy {
UserGroupInformation user = UserGroupInformation.getCurrentUser();
if (UserGroupInformation.isSecurityEnabled()) {
- Token<ContainerTokenIdentifier> token =
- ProtoUtils.convertFromProtoFormat(containerToken, cmAddr);
+ org.apache.hadoop.security.token.Token<ContainerTokenIdentifier> token =
+ ConverterUtils.convertFromYarn(containerToken, cmAddr);
// the user in createRemoteUser in this context has to be ContainerID
user = UserGroupInformation.createRemoteUser(containerID.toString());
user.addToken(token);
}
- ContainerManager proxy = user.doAs(new PrivilegedAction<ContainerManager>() {
+ ContainerManagementProtocol proxy = user.doAs(new PrivilegedAction<ContainerManagementProtocol>() {
@Override
- public ContainerManager run() {
- return (ContainerManager) yarnRPC.getProxy(ContainerManager.class,
+ public ContainerManagementProtocol run() {
+ return (ContainerManagementProtocol) yarnRPC.getProxy(ContainerManagementProtocol.class,
cmAddr, conf);
}
});
@@ -107,7 +106,7 @@ public class YarnContainerProxy extends ContainerProxy {
return;
}
- ContainerManager proxy = null;
+ ContainerManagementProtocol proxy = null;
try {
proxy = getCMProxy(containerID, containerMgrAddress,
@@ -117,13 +116,16 @@ public class YarnContainerProxy extends ContainerProxy {
ContainerLaunchContext containerLaunchContext = createContainerLaunchContext(commonContainerLaunchContext);
// Now launch the actual container
+ List<StartContainerRequest> startRequestList = new ArrayList<StartContainerRequest>();
StartContainerRequest startRequest = Records
.newRecord(StartContainerRequest.class);
startRequest.setContainerLaunchContext(containerLaunchContext);
- StartContainerResponse response = proxy.startContainer(startRequest);
+ startRequestList.add(startRequest);
+ StartContainersRequest startRequests = Records.newRecord(StartContainersRequest.class);
+ startRequests.setStartContainerRequests(startRequestList);
+ StartContainersResponse response = proxy.startContainers(startRequests);
- ByteBuffer portInfo = response
- .getServiceResponse(PullServerAuxService.PULLSERVER_SERVICEID);
+ ByteBuffer portInfo = response.getAllServicesMetaData().get(PullServerAuxService.PULLSERVER_SERVICEID);
if(portInfo != null) {
port = PullServerAuxService.deserializeMetaData(portInfo);
@@ -206,9 +208,12 @@ public class YarnContainerProxy extends ContainerProxy {
List<String> commands = new ArrayList<String>();
commands.add(command.toString());
- return BuilderUtils.newContainerLaunchContext(containerID, commonContainerLaunchContext.getUser(),
- container.getResource(), commonContainerLaunchContext.getLocalResources(), myEnv, commands,
- myServiceData, null, new HashMap<ApplicationAccessType, String>());
+ return BuilderUtils.newContainerLaunchContext(commonContainerLaunchContext.getLocalResources(),
+ myEnv,
+ commands,
+ myServiceData,
+ null,
+ new HashMap<ApplicationAccessType, String>());
}
public static ContainerLaunchContext createCommonContainerLaunchContext(Configuration config,
@@ -216,8 +221,10 @@ public class YarnContainerProxy extends ContainerProxy {
TajoConf conf = (TajoConf)config;
ContainerLaunchContext ctx = Records.newRecord(ContainerLaunchContext.class);
+
try {
- ctx.setUser(UserGroupInformation.getCurrentUser().getShortUserName());
+ ByteBuffer userToken = ByteBuffer.wrap(UserGroupInformation.getCurrentUser().getShortUserName().getBytes());
+ ctx.setTokens(userToken);
} catch (IOException e) {
e.printStackTrace();
}
@@ -362,16 +369,17 @@ public class YarnContainerProxy extends ContainerProxy {
} else {
LOG.info("KILLING " + containerID);
- ContainerManager proxy = null;
+ ContainerManagementProtocol proxy = null;
try {
proxy = getCMProxy(this.containerID, this.containerMgrAddress,
this.containerToken);
// kill the remote container if already launched
- StopContainerRequest stopRequest = Records
- .newRecord(StopContainerRequest.class);
- stopRequest.setContainerId(this.containerID);
- proxy.stopContainer(stopRequest);
+ List<ContainerId> willBeStopedIds = new ArrayList<ContainerId>();
+ willBeStopedIds.add(this.containerID);
+ StopContainersRequest stopRequests = Records.newRecord(StopContainersRequest.class);
+ stopRequests.setContainerIds(willBeStopedIds);
+ proxy.stopContainers(stopRequests);
// If stopContainer returns without an error, assuming the stop made
// it over to the NodeManager.
// context.getEventHandler().handle(
http://git-wip-us.apache.org/repos/asf/incubator-tajo/blob/ecc749e8/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/master/YarnTaskRunnerLauncherImpl.java
----------------------------------------------------------------------
diff --git a/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/master/YarnTaskRunnerLauncherImpl.java b/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/master/YarnTaskRunnerLauncherImpl.java
index c8792ca..7c38e7a 100644
--- a/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/master/YarnTaskRunnerLauncherImpl.java
+++ b/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/master/YarnTaskRunnerLauncherImpl.java
@@ -22,6 +22,7 @@ import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.permission.FsPermission;
+import org.apache.hadoop.service.AbstractService;
import org.apache.hadoop.yarn.api.ApplicationConstants.Environment;
import org.apache.hadoop.yarn.api.records.Container;
import org.apache.hadoop.yarn.api.records.ContainerId;
@@ -29,7 +30,6 @@ import org.apache.hadoop.yarn.api.records.ContainerLaunchContext;
import org.apache.hadoop.yarn.factories.RecordFactory;
import org.apache.hadoop.yarn.factory.providers.RecordFactoryProvider;
import org.apache.hadoop.yarn.ipc.YarnRPC;
-import org.apache.hadoop.yarn.service.AbstractService;
import org.apache.tajo.ExecutionBlockId;
import org.apache.tajo.conf.TajoConf;
import org.apache.tajo.master.TaskRunnerGroupEvent.EventType;
http://git-wip-us.apache.org/repos/asf/incubator-tajo/blob/ecc749e8/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/master/querymaster/Query.java
----------------------------------------------------------------------
diff --git a/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/master/querymaster/Query.java b/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/master/querymaster/Query.java
index 245c897..79ae34d 100644
--- a/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/master/querymaster/Query.java
+++ b/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/master/querymaster/Query.java
@@ -25,9 +25,9 @@ import org.apache.commons.logging.LogFactory;
import org.apache.hadoop.fs.ContentSummary;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
-import org.apache.hadoop.yarn.Clock;
import org.apache.hadoop.yarn.event.EventHandler;
import org.apache.hadoop.yarn.state.*;
+import org.apache.hadoop.yarn.util.Clock;
import org.apache.tajo.ExecutionBlockId;
import org.apache.tajo.QueryId;
import org.apache.tajo.TajoConstants;
http://git-wip-us.apache.org/repos/asf/incubator-tajo/blob/ecc749e8/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/master/querymaster/QueryInProgress.java
----------------------------------------------------------------------
diff --git a/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/master/querymaster/QueryInProgress.java b/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/master/querymaster/QueryInProgress.java
index 0eb9120..92ba2e2 100644
--- a/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/master/querymaster/QueryInProgress.java
+++ b/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/master/querymaster/QueryInProgress.java
@@ -22,15 +22,15 @@ import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.net.NetUtils;
+import org.apache.hadoop.service.CompositeService;
import org.apache.hadoop.util.StringUtils;
import org.apache.hadoop.yarn.event.EventHandler;
-import org.apache.hadoop.yarn.service.CompositeService;
import org.apache.tajo.QueryId;
import org.apache.tajo.TajoProtos;
import org.apache.tajo.engine.planner.logical.LogicalRootNode;
-import org.apache.tajo.ipc.QueryMasterProtocol;
-import org.apache.tajo.ipc.QueryMasterProtocol.*;
import org.apache.tajo.engine.query.QueryContext;
+import org.apache.tajo.ipc.QueryMasterProtocol;
+import org.apache.tajo.ipc.QueryMasterProtocol.QueryMasterProtocolService;
import org.apache.tajo.ipc.TajoWorkerProtocol;
import org.apache.tajo.master.TajoAsyncDispatcher;
import org.apache.tajo.master.TajoMaster;
http://git-wip-us.apache.org/repos/asf/incubator-tajo/blob/ecc749e8/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/master/querymaster/QueryJobManager.java
----------------------------------------------------------------------
diff --git a/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/master/querymaster/QueryJobManager.java b/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/master/querymaster/QueryJobManager.java
index 41df7c2..09d1d26 100644
--- a/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/master/querymaster/QueryJobManager.java
+++ b/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/master/querymaster/QueryJobManager.java
@@ -21,14 +21,14 @@ package org.apache.tajo.master.querymaster;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.service.CompositeService;
import org.apache.hadoop.yarn.event.AsyncDispatcher;
import org.apache.hadoop.yarn.event.EventHandler;
-import org.apache.hadoop.yarn.service.CompositeService;
import org.apache.tajo.QueryId;
import org.apache.tajo.QueryIdFactory;
import org.apache.tajo.engine.planner.logical.LogicalRootNode;
-import org.apache.tajo.ipc.TajoMasterProtocol;
import org.apache.tajo.engine.query.QueryContext;
+import org.apache.tajo.ipc.TajoMasterProtocol;
import org.apache.tajo.master.TajoMaster;
import org.apache.tajo.master.rm.WorkerResource;
http://git-wip-us.apache.org/repos/asf/incubator-tajo/blob/ecc749e8/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/master/querymaster/QueryMaster.java
----------------------------------------------------------------------
diff --git a/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/master/querymaster/QueryMaster.java b/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/master/querymaster/QueryMaster.java
index 02d3d81..ecf6cb2 100644
--- a/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/master/querymaster/QueryMaster.java
+++ b/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/master/querymaster/QueryMaster.java
@@ -22,12 +22,11 @@ import com.google.common.collect.Maps;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.hadoop.conf.Configuration;
-import org.apache.hadoop.yarn.Clock;
-import org.apache.hadoop.yarn.SystemClock;
+import org.apache.hadoop.service.CompositeService;
import org.apache.hadoop.yarn.event.Event;
import org.apache.hadoop.yarn.event.EventHandler;
-import org.apache.hadoop.yarn.service.CompositeService;
-import org.apache.hadoop.yarn.service.Service;
+import org.apache.hadoop.yarn.util.Clock;
+import org.apache.hadoop.yarn.util.SystemClock;
import org.apache.tajo.QueryId;
import org.apache.tajo.TajoProtos;
import org.apache.tajo.conf.TajoConf;
@@ -247,12 +246,6 @@ public class QueryMaster extends CompositeService implements EventHandler {
}
}
- protected void addIfService(Object object) {
- if (object instanceof Service) {
- addService((Service) object);
- }
- }
-
@Override
public void handle(Event event) {
dispatcher.getEventHandler().handle(event);
http://git-wip-us.apache.org/repos/asf/incubator-tajo/blob/ecc749e8/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/master/querymaster/QueryMasterManagerService.java
----------------------------------------------------------------------
diff --git a/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/master/querymaster/QueryMasterManagerService.java b/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/master/querymaster/QueryMasterManagerService.java
index 78c417e..a75548b 100644
--- a/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/master/querymaster/QueryMasterManagerService.java
+++ b/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/master/querymaster/QueryMasterManagerService.java
@@ -23,8 +23,8 @@ import com.google.protobuf.RpcController;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.service.CompositeService;
import org.apache.hadoop.yarn.api.records.ContainerId;
-import org.apache.hadoop.yarn.service.CompositeService;
import org.apache.tajo.ExecutionBlockId;
import org.apache.tajo.QueryId;
import org.apache.tajo.QueryUnitAttemptId;
http://git-wip-us.apache.org/repos/asf/incubator-tajo/blob/ecc749e8/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/master/querymaster/QueryMasterRunner.java
----------------------------------------------------------------------
diff --git a/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/master/querymaster/QueryMasterRunner.java b/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/master/querymaster/QueryMasterRunner.java
index ae1c1e5..56dd789 100644
--- a/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/master/querymaster/QueryMasterRunner.java
+++ b/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/master/querymaster/QueryMasterRunner.java
@@ -23,7 +23,7 @@ import org.apache.commons.logging.LogFactory;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.security.UserGroupInformation;
-import org.apache.hadoop.yarn.service.AbstractService;
+import org.apache.hadoop.service.AbstractService;
import org.apache.hadoop.yarn.util.RackResolver;
import org.apache.tajo.QueryId;
import org.apache.tajo.TajoConstants;
http://git-wip-us.apache.org/repos/asf/incubator-tajo/blob/ecc749e8/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/master/querymaster/QueryMasterTask.java
----------------------------------------------------------------------
diff --git a/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/master/querymaster/QueryMasterTask.java b/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/master/querymaster/QueryMasterTask.java
index 5d717ee..e57ea74 100644
--- a/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/master/querymaster/QueryMasterTask.java
+++ b/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/master/querymaster/QueryMasterTask.java
@@ -26,9 +26,9 @@ import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.fs.permission.FsPermission;
import org.apache.hadoop.security.UserGroupInformation;
-import org.apache.hadoop.yarn.Clock;
+import org.apache.hadoop.service.CompositeService;
import org.apache.hadoop.yarn.event.EventHandler;
-import org.apache.hadoop.yarn.service.CompositeService;
+import org.apache.hadoop.yarn.util.Clock;
import org.apache.tajo.*;
import org.apache.tajo.algebra.Expr;
import org.apache.tajo.catalog.CatalogService;
http://git-wip-us.apache.org/repos/asf/incubator-tajo/blob/ecc749e8/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/master/rm/TajoWorkerContainer.java
----------------------------------------------------------------------
diff --git a/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/master/rm/TajoWorkerContainer.java b/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/master/rm/TajoWorkerContainer.java
index 9f7aeb1..5ac6e39 100644
--- a/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/master/rm/TajoWorkerContainer.java
+++ b/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/master/rm/TajoWorkerContainer.java
@@ -20,7 +20,7 @@ package org.apache.tajo.master.rm;
import org.apache.hadoop.yarn.api.records.*;
-public class TajoWorkerContainer implements Container {
+public class TajoWorkerContainer extends Container {
ContainerId id;
NodeId nodeId;
WorkerResource workerResource;
@@ -84,32 +84,12 @@ public class TajoWorkerContainer implements Container {
}
@Override
- public ContainerState getState() {
+ public Token getContainerToken() {
return null; //To change body of implemented methods use File | Settings | File Templates.
}
@Override
- public void setState(ContainerState state) {
- //To change body of implemented methods use File | Settings | File Templates.
- }
-
- @Override
- public ContainerToken getContainerToken() {
- return null; //To change body of implemented methods use File | Settings | File Templates.
- }
-
- @Override
- public void setContainerToken(ContainerToken containerToken) {
- //To change body of implemented methods use File | Settings | File Templates.
- }
-
- @Override
- public ContainerStatus getContainerStatus() {
- return null; //To change body of implemented methods use File | Settings | File Templates.
- }
-
- @Override
- public void setContainerStatus(ContainerStatus containerStatus) {
+ public void setContainerToken(Token containerToken) {
//To change body of implemented methods use File | Settings | File Templates.
}
http://git-wip-us.apache.org/repos/asf/incubator-tajo/blob/ecc749e8/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/master/rm/TajoWorkerContainerId.java
----------------------------------------------------------------------
diff --git a/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/master/rm/TajoWorkerContainerId.java b/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/master/rm/TajoWorkerContainerId.java
index 2d7c1c3..634ad2b 100644
--- a/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/master/rm/TajoWorkerContainerId.java
+++ b/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/master/rm/TajoWorkerContainerId.java
@@ -85,4 +85,9 @@ public class TajoWorkerContainerId extends ContainerId {
.build();
}
}
+
+ @Override
+ protected void build() {
+
+ }
}
http://git-wip-us.apache.org/repos/asf/incubator-tajo/blob/ecc749e8/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/master/rm/TajoWorkerResourceManager.java
----------------------------------------------------------------------
diff --git a/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/master/rm/TajoWorkerResourceManager.java b/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/master/rm/TajoWorkerResourceManager.java
index 1485ffe..b39901e 100644
--- a/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/master/rm/TajoWorkerResourceManager.java
+++ b/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/master/rm/TajoWorkerResourceManager.java
@@ -22,7 +22,7 @@ import com.google.protobuf.RpcCallback;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.hadoop.conf.Configuration;
-import org.apache.hadoop.yarn.api.records.impl.pb.NodeIdPBImpl;
+import org.apache.hadoop.yarn.api.records.NodeId;
import org.apache.hadoop.yarn.proto.YarnProtos;
import org.apache.tajo.ExecutionBlockId;
import org.apache.tajo.QueryId;
@@ -375,10 +375,8 @@ public class TajoWorkerResourceManager implements WorkerResourceManager {
new ArrayList<TajoMasterProtocol.WorkerAllocatedResource>();
for(AllocatedWorkerResource eachWorker: allocatedWorkerResources) {
- NodeIdPBImpl nodeId = new NodeIdPBImpl();
-
- nodeId.setHost(eachWorker.workerResource.getAllocatedHost());
- nodeId.setPort(eachWorker.workerResource.getPeerRpcPort());
+ NodeId nodeId = NodeId.newInstance(eachWorker.workerResource.getAllocatedHost(),
+ eachWorker.workerResource.getPeerRpcPort());
TajoWorkerContainerId containerId = new TajoWorkerContainerId();
http://git-wip-us.apache.org/repos/asf/incubator-tajo/blob/ecc749e8/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/master/rm/YarnRMContainerAllocator.java
----------------------------------------------------------------------
diff --git a/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/master/rm/YarnRMContainerAllocator.java b/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/master/rm/YarnRMContainerAllocator.java
index 2202183..b9e132b 100644
--- a/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/master/rm/YarnRMContainerAllocator.java
+++ b/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/master/rm/YarnRMContainerAllocator.java
@@ -22,16 +22,15 @@ import com.google.common.collect.Lists;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.hadoop.conf.Configuration;
-import org.apache.hadoop.yarn.YarnException;
import org.apache.hadoop.yarn.api.protocolrecords.AllocateResponse;
import org.apache.hadoop.yarn.api.protocolrecords.RegisterApplicationMasterResponse;
-import org.apache.hadoop.yarn.api.records.AMResponse;
+import org.apache.hadoop.yarn.api.records.ApplicationAttemptId;
import org.apache.hadoop.yarn.api.records.Container;
import org.apache.hadoop.yarn.api.records.FinalApplicationStatus;
import org.apache.hadoop.yarn.api.records.Priority;
-import org.apache.hadoop.yarn.client.AMRMClientImpl;
+import org.apache.hadoop.yarn.client.api.impl.AMRMClientImpl;
import org.apache.hadoop.yarn.event.EventHandler;
-import org.apache.hadoop.yarn.exceptions.YarnRemoteException;
+import org.apache.hadoop.yarn.exceptions.YarnException;
import org.apache.tajo.ExecutionBlockId;
import org.apache.tajo.TajoProtos;
import org.apache.tajo.master.event.ContainerAllocationEvent;
@@ -43,6 +42,7 @@ import org.apache.tajo.master.querymaster.SubQuery;
import org.apache.tajo.master.querymaster.SubQueryState;
import org.apache.tajo.util.ApplicationIdUtils;
+import java.io.IOException;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
@@ -58,11 +58,13 @@ public class YarnRMContainerAllocator extends AMRMClientImpl
class.getName());
private QueryMasterTask.QueryMasterTaskContext context;
+ private ApplicationAttemptId appAttemptId;
private final EventHandler eventHandler;
public YarnRMContainerAllocator(QueryMasterTask.QueryMasterTaskContext context) {
- super(ApplicationIdUtils.createApplicationAttemptId(context.getQueryId()));
+ super();
this.context = context;
+ this.appAttemptId = ApplicationIdUtils.createApplicationAttemptId(context.getQueryId());
this.eventHandler = context.getDispatcher().getEventHandler();
}
@@ -90,7 +92,9 @@ public class YarnRMContainerAllocator extends AMRMClientImpl
}
}
context.getQueryMasterContext().getWorkerContext().setNumClusterNodes(allocateResponse.getNumClusterNodes());
- } catch (YarnRemoteException e) {
+ } catch (IOException e) {
+ LOG.error(e);
+ } catch (YarnException e) {
LOG.error(e);
}
@@ -173,28 +177,23 @@ public class YarnRMContainerAllocator extends AMRMClientImpl
public void heartbeat() throws Exception {
AllocateResponse allocateResponse = allocate(context.getProgress());
- AMResponse response = allocateResponse.getAMResponse();
- if(response == null) {
- LOG.warn("AM Response is null");
- return;
- }
- List<Container> allocatedContainers = response.getAllocatedContainers();
+
+ List<Container> allocatedContainers = allocateResponse.getAllocatedContainers();
long currentTime = System.currentTimeMillis();
if ((currentTime - prevReportTime.longValue()) >= reportInterval) {
LOG.debug("Available Cluster Nodes: " + allocateResponse.getNumClusterNodes());
LOG.debug("Num of Allocated Containers: " + allocatedContainers.size());
- LOG.info("Available Resource: " + response.getAvailableResources());
+ LOG.info("Available Resource: " + allocateResponse.getAvailableResources());
prevReportTime.set(currentTime);
}
if (allocatedContainers.size() > 0) {
LOG.info("================================================================");
- for (Container container : response.getAllocatedContainers()) {
+ for (Container container : allocateResponse.getAllocatedContainers()) {
LOG.info("> Container Id: " + container.getId());
LOG.info("> Node Id: " + container.getNodeId());
LOG.info("> Resource (Mem): " + container.getResource().getMemory());
- LOG.info("> State : " + container.getState());
LOG.info("> Priority: " + container.getPriority());
}
LOG.info("================================================================");
@@ -227,7 +226,7 @@ public class YarnRMContainerAllocator extends AMRMClientImpl
LOG.info(event);
subQueryMap.put(event.getPriority(), event.getExecutionBlockId());
addContainerRequest(new ContainerRequest(event.getCapability(), null, null,
- event.getPriority(), event.getRequiredNum()));
+ event.getPriority()));
} else if (event.getType() == ContainerAllocatorEventType.CONTAINER_DEALLOCATE) {
LOG.info(event);
http://git-wip-us.apache.org/repos/asf/incubator-tajo/blob/ecc749e8/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/master/rm/YarnTajoResourceManager.java
----------------------------------------------------------------------
diff --git a/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/master/rm/YarnTajoResourceManager.java b/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/master/rm/YarnTajoResourceManager.java
index 8b72cf9..72ada9b 100644
--- a/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/master/rm/YarnTajoResourceManager.java
+++ b/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/master/rm/YarnTajoResourceManager.java
@@ -23,21 +23,21 @@ import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.security.UserGroupInformation;
-import org.apache.hadoop.yarn.YarnException;
-import org.apache.hadoop.yarn.api.AMRMProtocol;
import org.apache.hadoop.yarn.api.ApplicationConstants;
+import org.apache.hadoop.yarn.api.ApplicationMasterProtocol;
import org.apache.hadoop.yarn.api.protocolrecords.FinishApplicationMasterRequest;
-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.client.api.YarnClient;
+import org.apache.hadoop.yarn.client.api.YarnClientApplication;
+import org.apache.hadoop.yarn.client.api.impl.YarnClientImpl;
import org.apache.hadoop.yarn.conf.YarnConfiguration;
-import org.apache.hadoop.yarn.exceptions.YarnRemoteException;
+import org.apache.hadoop.yarn.exceptions.YarnException;
+import org.apache.hadoop.yarn.exceptions.YarnRuntimeException;
import org.apache.hadoop.yarn.factories.RecordFactory;
import org.apache.hadoop.yarn.factory.providers.RecordFactoryProvider;
import org.apache.hadoop.yarn.ipc.YarnRPC;
import org.apache.hadoop.yarn.proto.YarnProtos;
-import org.apache.hadoop.yarn.util.BuilderUtils;
+import org.apache.hadoop.yarn.server.utils.BuilderUtils;
import org.apache.hadoop.yarn.util.Records;
import org.apache.tajo.ExecutionBlockId;
import org.apache.tajo.QueryId;
@@ -61,7 +61,7 @@ public class YarnTajoResourceManager implements WorkerResourceManager {
private static final Log LOG = LogFactory.getLog(YarnTajoResourceManager.class);
private YarnClient yarnClient;
- private AMRMProtocol rmClient;
+ private ApplicationMasterProtocol rmClient;
private final RecordFactory recordFactory = RecordFactoryProvider.getRecordFactory(null);
private Configuration conf;
private TajoMaster.MasterContext masterContext;
@@ -127,10 +127,10 @@ public class YarnTajoResourceManager implements WorkerResourceManager {
queryInProgress.getEventHandler().handle(
new QueryJobEvent(QueryJobEvent.Type.QUERY_JOB_START, queryInProgress.getQueryInfo()));
-
- } catch (YarnRemoteException e) {
- LOG.error(e.getMessage(), e);
- //TODO set QueryState(fail)
+ } catch (IOException e) {
+ LOG.error(e);
+ } catch (YarnException e) {
+ LOG.error(e);
}
}
@@ -150,13 +150,13 @@ public class YarnTajoResourceManager implements WorkerResourceManager {
try {
currentUser = UserGroupInformation.getCurrentUser();
} catch (IOException e) {
- throw new YarnException(e);
+ throw new YarnRuntimeException(e);
}
- rmClient = currentUser.doAs(new PrivilegedAction<AMRMProtocol>() {
+ rmClient = currentUser.doAs(new PrivilegedAction<ApplicationMasterProtocol>() {
@Override
- public AMRMProtocol run() {
- return (AMRMProtocol) rpc.getProxy(AMRMProtocol.class, rmAddress, yarnConf);
+ public ApplicationMasterProtocol run() {
+ return (ApplicationMasterProtocol) rpc.getProxy(ApplicationMasterProtocol.class, rmAddress, yarnConf);
}
});
}
@@ -164,10 +164,8 @@ public class YarnTajoResourceManager implements WorkerResourceManager {
@Override
public String getSeedQueryId() throws IOException {
try {
- GetNewApplicationResponse newApp = yarnClient.getNewApplication();
- ApplicationId appId = newApp.getApplicationId();
-
- return appId.toString();
+ YarnClientApplication app = yarnClient.createApplication();
+ return app.getApplicationSubmissionContext().getApplicationId().toString();
} catch (Exception e) {
LOG.error(e.getMessage(), e);
@@ -193,8 +191,7 @@ public class YarnTajoResourceManager implements WorkerResourceManager {
}
FinishApplicationMasterRequest request = recordFactory
.newRecordInstance(FinishApplicationMasterRequest.class);
- request.setAppAttemptId(ApplicationIdUtils.createApplicationAttemptId(queryId));
- request.setFinishApplicationStatus(appStatus);
+ request.setFinalApplicationStatus(appStatus);
request.setDiagnostics("QueryMaster shutdown by TajoMaster.");
rmClient.finishApplicationMaster(request);
} catch (Exception e) {
@@ -208,8 +205,7 @@ public class YarnTajoResourceManager implements WorkerResourceManager {
this.yarnClient.start();
}
- private ApplicationAttemptId allocateAndLaunchQueryMaster(QueryInProgress queryInProgress)
- throws YarnRemoteException {
+ private ApplicationAttemptId allocateAndLaunchQueryMaster(QueryInProgress queryInProgress) throws IOException, YarnException {
QueryId queryId = queryInProgress.getQueryId();
ApplicationId appId = ApplicationIdUtils.queryIdToAppId(queryId);
@@ -285,9 +281,13 @@ public class YarnTajoResourceManager implements WorkerResourceManager {
Map<String, ByteBuffer> myServiceData = new HashMap<String, ByteBuffer>();
ContainerLaunchContext masterContainerContext = BuilderUtils.newContainerLaunchContext(
- null, commonContainerLaunchContext.getUser(),
- resource, commonContainerLaunchContext.getLocalResources(), myEnv, commands,
- myServiceData, null, new HashMap<ApplicationAccessType, String>(2));
+ commonContainerLaunchContext.getLocalResources(),
+ myEnv,
+ commands,
+ myServiceData,
+ null,
+ new HashMap<ApplicationAccessType, String>(2)
+ );
appContext.setAMContainerSpec(masterContainerContext);
@@ -303,7 +303,7 @@ public class YarnTajoResourceManager implements WorkerResourceManager {
}
private ApplicationReport monitorApplication(ApplicationId appId,
- Set<YarnApplicationState> finalState) throws YarnRemoteException {
+ Set<YarnApplicationState> finalState) throws IOException, YarnException {
long sleepTime = 100;
int count = 1;
@@ -314,7 +314,7 @@ public class YarnTajoResourceManager implements WorkerResourceManager {
LOG.info("Got application report from ASM for" + ", appId="
+ appId.getId() + ", appAttemptId="
+ report.getCurrentApplicationAttemptId() + ", clientToken="
- + report.getClientToken() + ", appDiagnostics="
+ + report.getClientToAMToken() + ", appDiagnostics="
+ report.getDiagnostics() + ", appMasterHost=" + report.getHost()
+ ", appQueue=" + report.getQueue() + ", appMasterRpcPort="
+ report.getRpcPort() + ", appStartTime=" + report.getStartTime()
@@ -348,7 +348,10 @@ public class YarnTajoResourceManager implements WorkerResourceManager {
YarnApplicationState.FINISHED,
YarnApplicationState.KILLED,
YarnApplicationState.FAILED).contains(state);
- } catch (YarnRemoteException e) {
+ } catch (YarnException e) {
+ LOG.error(e.getMessage(), e);
+ return false;
+ } catch (IOException e) {
LOG.error(e.getMessage(), e);
return false;
}
http://git-wip-us.apache.org/repos/asf/incubator-tajo/blob/ecc749e8/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/util/ApplicationIdUtils.java
----------------------------------------------------------------------
diff --git a/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/util/ApplicationIdUtils.java b/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/util/ApplicationIdUtils.java
index cf68145..1db0c3b 100644
--- a/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/util/ApplicationIdUtils.java
+++ b/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/util/ApplicationIdUtils.java
@@ -21,7 +21,7 @@ package org.apache.tajo.util;
import org.apache.hadoop.yarn.api.records.ApplicationAttemptId;
import org.apache.hadoop.yarn.api.records.ApplicationId;
import org.apache.hadoop.yarn.proto.YarnProtos;
-import org.apache.hadoop.yarn.util.BuilderUtils;
+import org.apache.hadoop.yarn.server.utils.BuilderUtils;
import org.apache.tajo.QueryId;
import org.apache.tajo.QueryIdFactory;
http://git-wip-us.apache.org/repos/asf/incubator-tajo/blob/ecc749e8/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/worker/AbstractResourceAllocator.java
----------------------------------------------------------------------
diff --git a/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/worker/AbstractResourceAllocator.java b/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/worker/AbstractResourceAllocator.java
index c901e29..55aa8c4 100644
--- a/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/worker/AbstractResourceAllocator.java
+++ b/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/worker/AbstractResourceAllocator.java
@@ -19,8 +19,8 @@
package org.apache.tajo.worker;
import com.google.common.collect.Maps;
+import org.apache.hadoop.service.CompositeService;
import org.apache.hadoop.yarn.api.records.ContainerId;
-import org.apache.hadoop.yarn.service.CompositeService;
import org.apache.tajo.master.ContainerProxy;
import java.util.Map;
http://git-wip-us.apache.org/repos/asf/incubator-tajo/blob/ecc749e8/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/worker/TajoResourceAllocator.java
----------------------------------------------------------------------
diff --git a/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/worker/TajoResourceAllocator.java b/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/worker/TajoResourceAllocator.java
index ca6ea41..6d01f55 100644
--- a/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/worker/TajoResourceAllocator.java
+++ b/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/worker/TajoResourceAllocator.java
@@ -28,8 +28,8 @@ import org.apache.hadoop.fs.permission.FsPermission;
import org.apache.hadoop.yarn.api.records.ApplicationAttemptId;
import org.apache.hadoop.yarn.api.records.Container;
import org.apache.hadoop.yarn.api.records.ContainerId;
+import org.apache.hadoop.yarn.api.records.NodeId;
import org.apache.hadoop.yarn.api.records.impl.pb.ApplicationAttemptIdPBImpl;
-import org.apache.hadoop.yarn.api.records.impl.pb.NodeIdPBImpl;
import org.apache.hadoop.yarn.event.EventHandler;
import org.apache.hadoop.yarn.proto.YarnProtos;
import org.apache.tajo.ExecutionBlockId;
@@ -286,10 +286,8 @@ public class TajoResourceAllocator extends AbstractResourceAllocator {
List<Container> containers = new ArrayList<Container>();
for(TajoMasterProtocol.WorkerAllocatedResource eachAllocatedResource: allocatedResources) {
TajoWorkerContainer container = new TajoWorkerContainer();
- NodeIdPBImpl nodeId = new NodeIdPBImpl();
-
- nodeId.setHost(eachAllocatedResource.getWorkerHost());
- nodeId.setPort(eachAllocatedResource.getPeerRpcPort());
+ NodeId nodeId = NodeId.newInstance(eachAllocatedResource.getWorkerHost(),
+ eachAllocatedResource.getPeerRpcPort());
TajoWorkerContainerId containerId = new TajoWorkerContainerId();
http://git-wip-us.apache.org/repos/asf/incubator-tajo/blob/ecc749e8/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/worker/TajoWorker.java
----------------------------------------------------------------------
diff --git a/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/worker/TajoWorker.java b/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/worker/TajoWorker.java
index 986f453..456bd95 100644
--- a/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/worker/TajoWorker.java
+++ b/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/worker/TajoWorker.java
@@ -27,8 +27,8 @@ import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.LocalDirAllocator;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.fs.shell.PathData;
+import org.apache.hadoop.service.CompositeService;
import org.apache.hadoop.util.StringUtils;
-import org.apache.hadoop.yarn.service.CompositeService;
import org.apache.hadoop.yarn.util.RackResolver;
import org.apache.tajo.QueryId;
import org.apache.tajo.TajoConstants;
http://git-wip-us.apache.org/repos/asf/incubator-tajo/blob/ecc749e8/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/worker/TajoWorkerClientService.java
----------------------------------------------------------------------
diff --git a/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/worker/TajoWorkerClientService.java b/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/worker/TajoWorkerClientService.java
index 55bc746..9bcdf3a 100644
--- a/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/worker/TajoWorkerClientService.java
+++ b/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/worker/TajoWorkerClientService.java
@@ -24,7 +24,7 @@ import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.security.UserGroupInformation;
-import org.apache.hadoop.yarn.service.AbstractService;
+import org.apache.hadoop.service.AbstractService;
import org.apache.tajo.QueryId;
import org.apache.tajo.QueryIdFactory;
import org.apache.tajo.TajoIdProtos;
http://git-wip-us.apache.org/repos/asf/incubator-tajo/blob/ecc749e8/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/worker/TajoWorkerManagerService.java
----------------------------------------------------------------------
diff --git a/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/worker/TajoWorkerManagerService.java b/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/worker/TajoWorkerManagerService.java
index bd2dc4d..60b0903 100644
--- a/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/worker/TajoWorkerManagerService.java
+++ b/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/worker/TajoWorkerManagerService.java
@@ -23,7 +23,7 @@ import com.google.protobuf.RpcController;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.hadoop.conf.Configuration;
-import org.apache.hadoop.yarn.service.CompositeService;
+import org.apache.hadoop.service.CompositeService;
import org.apache.tajo.QueryId;
import org.apache.tajo.TajoIdProtos;
import org.apache.tajo.conf.TajoConf;
http://git-wip-us.apache.org/repos/asf/incubator-tajo/blob/ecc749e8/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/worker/TaskRunner.java
----------------------------------------------------------------------
diff --git a/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/worker/TaskRunner.java b/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/worker/TaskRunner.java
index 18c312f..febb671 100644
--- a/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/worker/TaskRunner.java
+++ b/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/worker/TaskRunner.java
@@ -26,10 +26,10 @@ import org.apache.hadoop.fs.LocalDirAllocator;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.net.NetUtils;
import org.apache.hadoop.security.UserGroupInformation;
+import org.apache.hadoop.service.AbstractService;
import org.apache.hadoop.yarn.api.records.ContainerId;
import org.apache.hadoop.yarn.api.records.NodeId;
import org.apache.hadoop.yarn.api.records.impl.pb.ContainerIdPBImpl;
-import org.apache.hadoop.yarn.service.AbstractService;
import org.apache.hadoop.yarn.util.ConverterUtils;
import org.apache.tajo.ExecutionBlockId;
import org.apache.tajo.QueryId;
http://git-wip-us.apache.org/repos/asf/incubator-tajo/blob/ecc749e8/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/worker/TaskRunnerManager.java
----------------------------------------------------------------------
diff --git a/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/worker/TaskRunnerManager.java b/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/worker/TaskRunnerManager.java
index 6523a4f..c7e159e 100644
--- a/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/worker/TaskRunnerManager.java
+++ b/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/worker/TaskRunnerManager.java
@@ -21,7 +21,7 @@ package org.apache.tajo.worker;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.hadoop.conf.Configuration;
-import org.apache.hadoop.yarn.service.CompositeService;
+import org.apache.hadoop.service.CompositeService;
import org.apache.tajo.conf.TajoConf;
import java.util.*;
http://git-wip-us.apache.org/repos/asf/incubator-tajo/blob/ecc749e8/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/worker/YarnResourceAllocator.java
----------------------------------------------------------------------
diff --git a/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/worker/YarnResourceAllocator.java b/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/worker/YarnResourceAllocator.java
index e74a09d..1771255 100644
--- a/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/worker/YarnResourceAllocator.java
+++ b/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/worker/YarnResourceAllocator.java
@@ -21,10 +21,11 @@ package org.apache.tajo.worker;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.service.Service;
import org.apache.hadoop.yarn.api.records.ContainerId;
import org.apache.hadoop.yarn.api.records.impl.pb.ContainerIdPBImpl;
-import org.apache.hadoop.yarn.client.YarnClient;
-import org.apache.hadoop.yarn.client.YarnClientImpl;
+import org.apache.hadoop.yarn.client.api.YarnClient;
+import org.apache.hadoop.yarn.client.api.impl.YarnClientImpl;
import org.apache.hadoop.yarn.ipc.YarnRPC;
import org.apache.hadoop.yarn.proto.YarnProtos;
import org.apache.tajo.conf.TajoConf;
@@ -83,7 +84,7 @@ public class YarnResourceAllocator extends AbstractResourceAllocator {
connectYarnClient();
taskRunnerLauncher = new YarnTaskRunnerLauncherImpl(queryTaskContext, yarnRPC);
- addService((org.apache.hadoop.yarn.service.Service) taskRunnerLauncher);
+ addService((Service) taskRunnerLauncher);
queryTaskContext.getDispatcher().register(TaskRunnerGroupEvent.EventType.class, taskRunnerLauncher);
rmAllocator = new YarnRMContainerAllocator(queryTaskContext);
http://git-wip-us.apache.org/repos/asf/incubator-tajo/blob/ecc749e8/tajo-core/tajo-core-backend/src/main/proto/InternalTypes.proto
----------------------------------------------------------------------
diff --git a/tajo-core/tajo-core-backend/src/main/proto/InternalTypes.proto b/tajo-core/tajo-core-backend/src/main/proto/InternalTypes.proto
index 5a9afc7..1a62bc2 100644
--- a/tajo-core/tajo-core-backend/src/main/proto/InternalTypes.proto
+++ b/tajo-core/tajo-core-backend/src/main/proto/InternalTypes.proto
@@ -29,4 +29,4 @@ message AvgLongProto {
message AvgDoubleProto {
required double sum = 1;
required int64 count = 2;
-}
\ No newline at end of file
+}
http://git-wip-us.apache.org/repos/asf/incubator-tajo/blob/ecc749e8/tajo-core/tajo-core-backend/src/main/proto/TajoMasterProtocol.proto
----------------------------------------------------------------------
diff --git a/tajo-core/tajo-core-backend/src/main/proto/TajoMasterProtocol.proto b/tajo-core/tajo-core-backend/src/main/proto/TajoMasterProtocol.proto
index 1280ab2..e0c09d2 100644
--- a/tajo-core/tajo-core-backend/src/main/proto/TajoMasterProtocol.proto
+++ b/tajo-core/tajo-core-backend/src/main/proto/TajoMasterProtocol.proto
@@ -127,11 +127,11 @@ message WorkerResourcesRequest {
message WorkerResourceReleaseRequest {
required ExecutionBlockIdProto executionBlockId = 1;
- repeated ContainerIdProto containerIds = 2;
+ repeated hadoop.yarn.ContainerIdProto containerIds = 2;
}
message WorkerAllocatedResource {
- required ContainerIdProto containerId = 1;
+ required hadoop.yarn.ContainerIdProto containerId = 1;
required string nodeId = 2;
required string workerHost = 3;
required int32 peerRpcPort = 4;
http://git-wip-us.apache.org/repos/asf/incubator-tajo/blob/ecc749e8/tajo-core/tajo-core-backend/src/main/proto/TajoWorkerProtocol.proto
----------------------------------------------------------------------
diff --git a/tajo-core/tajo-core-backend/src/main/proto/TajoWorkerProtocol.proto b/tajo-core/tajo-core-backend/src/main/proto/TajoWorkerProtocol.proto
index 9cf90a5..c5e7ad0 100644
--- a/tajo-core/tajo-core-backend/src/main/proto/TajoWorkerProtocol.proto
+++ b/tajo-core/tajo-core-backend/src/main/proto/TajoWorkerProtocol.proto
@@ -113,7 +113,7 @@ message QueryExecutionRequestProto {
}
message GetTaskRequestProto {
- required ContainerIdProto containerId = 1;
+ required hadoop.yarn.ContainerIdProto containerId = 1;
required ExecutionBlockIdProto executionBlockId = 2;
}
http://git-wip-us.apache.org/repos/asf/incubator-tajo/blob/ecc749e8/tajo-core/tajo-core-backend/src/test/java/org/apache/tajo/MiniTajoYarnCluster.java
----------------------------------------------------------------------
diff --git a/tajo-core/tajo-core-backend/src/test/java/org/apache/tajo/MiniTajoYarnCluster.java b/tajo-core/tajo-core-backend/src/test/java/org/apache/tajo/MiniTajoYarnCluster.java
index be0169d..88d913d 100644
--- a/tajo-core/tajo-core-backend/src/test/java/org/apache/tajo/MiniTajoYarnCluster.java
+++ b/tajo-core/tajo-core-backend/src/test/java/org/apache/tajo/MiniTajoYarnCluster.java
@@ -29,14 +29,14 @@ import org.apache.hadoop.mapreduce.MRConfig;
import org.apache.hadoop.mapreduce.MRJobConfig;
import org.apache.hadoop.mapreduce.v2.jobhistory.JHAdminConfig;
import org.apache.hadoop.mapreduce.v2.jobhistory.JobHistoryUtils;
+import org.apache.hadoop.service.AbstractService;
+import org.apache.hadoop.service.Service;
import org.apache.hadoop.util.JarFinder;
-import org.apache.hadoop.yarn.YarnException;
import org.apache.hadoop.yarn.conf.YarnConfiguration;
+import org.apache.hadoop.yarn.exceptions.YarnRuntimeException;
import org.apache.hadoop.yarn.server.MiniYARNCluster;
import org.apache.hadoop.yarn.server.nodemanager.ContainerExecutor;
import org.apache.hadoop.yarn.server.nodemanager.DefaultContainerExecutor;
-import org.apache.hadoop.yarn.service.AbstractService;
-import org.apache.hadoop.yarn.service.Service;
import org.apache.tajo.conf.TajoConf;
import org.apache.tajo.pullserver.PullServerAuxService;
@@ -93,7 +93,7 @@ public class MiniTajoYarnCluster extends MiniYARNCluster {
Path doneDirPath = fc.makeQualified(new Path(doneDir));
fc.mkdir(doneDirPath, null, true);
} catch (IOException e) {
- throw new YarnException("Could not create staging directory. ", e);
+ throw new YarnRuntimeException("Could not create staging directory. ", e);
}
conf.set(MRConfig.MASTER_ADDRESS, "test"); // The default is local because of
// which shuffle doesn't happen
@@ -147,7 +147,7 @@ public class MiniTajoYarnCluster extends MiniYARNCluster {
}
super.start();
} catch (Throwable t) {
- throw new YarnException(t);
+ throw new YarnRuntimeException(t);
}
LOG.info("MiniMRYARN ResourceManager address: " +
http://git-wip-us.apache.org/repos/asf/incubator-tajo/blob/ecc749e8/tajo-core/tajo-core-backend/src/test/java/org/apache/tajo/TajoTestingCluster.java
----------------------------------------------------------------------
diff --git a/tajo-core/tajo-core-backend/src/test/java/org/apache/tajo/TajoTestingCluster.java b/tajo-core/tajo-core-backend/src/test/java/org/apache/tajo/TajoTestingCluster.java
index b7c8512..6e73df8 100644
--- a/tajo-core/tajo-core-backend/src/test/java/org/apache/tajo/TajoTestingCluster.java
+++ b/tajo-core/tajo-core-backend/src/test/java/org/apache/tajo/TajoTestingCluster.java
@@ -184,6 +184,7 @@ public class TajoTestingCluster {
this.clusterTestBuildDir.toString());
conf.setInt(DFSConfigKeys.DFS_REPLICATION_KEY, 1);
+ conf.setBoolean(DFSConfigKeys.DFS_CLIENT_READ_SHORTCIRCUIT_KEY, false);
MiniDFSCluster.Builder builder = new MiniDFSCluster.Builder(new HdfsConfiguration(conf));
builder.hosts(hosts);
builder.numDataNodes(servers);
@@ -411,7 +412,7 @@ public class TajoTestingCluster {
conf.setInt(YarnConfiguration.RM_SCHEDULER_MINIMUM_ALLOCATION_MB, 384);
conf.setInt(YarnConfiguration.RM_SCHEDULER_MAXIMUM_ALLOCATION_MB, 3000);
conf.setInt(YarnConfiguration.RM_SCHEDULER_MINIMUM_ALLOCATION_VCORES, 1);
- conf.setInt(YarnConfiguration.RM_SCHEDULER_MAXIMUM_ALLOCATION_CORES, 2);
+ conf.setInt(YarnConfiguration.RM_SCHEDULER_MAXIMUM_ALLOCATION_VCORES, 2);
if (yarnCluster == null) {
yarnCluster = new MiniTajoYarnCluster(TajoTestingCluster.class.getName(), 3);
http://git-wip-us.apache.org/repos/asf/incubator-tajo/blob/ecc749e8/tajo-core/tajo-core-backend/src/test/java/org/apache/tajo/TestTajoIds.java
----------------------------------------------------------------------
diff --git a/tajo-core/tajo-core-backend/src/test/java/org/apache/tajo/TestTajoIds.java b/tajo-core/tajo-core-backend/src/test/java/org/apache/tajo/TestTajoIds.java
index 1929fb3..60b6c22 100644
--- a/tajo-core/tajo-core-backend/src/test/java/org/apache/tajo/TestTajoIds.java
+++ b/tajo-core/tajo-core-backend/src/test/java/org/apache/tajo/TestTajoIds.java
@@ -19,7 +19,7 @@
package org.apache.tajo;
import org.apache.hadoop.yarn.api.records.ApplicationId;
-import org.apache.hadoop.yarn.util.BuilderUtils;
+import org.apache.hadoop.yarn.server.utils.BuilderUtils;
import org.apache.tajo.engine.planner.global.MasterPlan;
import org.apache.tajo.util.TajoIdUtils;
import org.junit.Test;
http://git-wip-us.apache.org/repos/asf/incubator-tajo/blob/ecc749e8/tajo-core/tajo-core-pullserver/src/main/java/org/apache/tajo/pullserver/FadvisedChunkedFile.java
----------------------------------------------------------------------
diff --git a/tajo-core/tajo-core-pullserver/src/main/java/org/apache/tajo/pullserver/FadvisedChunkedFile.java b/tajo-core/tajo-core-pullserver/src/main/java/org/apache/tajo/pullserver/FadvisedChunkedFile.java
index 932d62e..ec34f00 100644
--- a/tajo-core/tajo-core-pullserver/src/main/java/org/apache/tajo/pullserver/FadvisedChunkedFile.java
+++ b/tajo-core/tajo-core-pullserver/src/main/java/org/apache/tajo/pullserver/FadvisedChunkedFile.java
@@ -22,7 +22,7 @@ import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.hadoop.io.ReadaheadPool;
import org.apache.hadoop.io.ReadaheadPool.ReadaheadRequest;
-import org.apache.hadoop.io.nativeio.NativeIO;
+import org.apache.hadoop.io.nativeio.NativeIO.POSIX;
import org.jboss.netty.handler.stream.ChunkedFile;
import java.io.FileDescriptor;
@@ -69,8 +69,8 @@ public class FadvisedChunkedFile extends ChunkedFile {
}
if (manageOsCache && getEndOffset() - getStartOffset() > 0) {
try {
- NativeIO.posixFadviseIfPossible(fd, getStartOffset(), getEndOffset()
- - getStartOffset(), NativeIO.POSIX_FADV_DONTNEED);
+ POSIX.posixFadviseIfPossible(identifier, fd, getStartOffset(), getEndOffset()
+ - getStartOffset(), POSIX.POSIX_FADV_DONTNEED);
} catch (Throwable t) {
LOG.warn("Failed to manage OS cache for " + identifier, t);
}
http://git-wip-us.apache.org/repos/asf/incubator-tajo/blob/ecc749e8/tajo-core/tajo-core-pullserver/src/main/java/org/apache/tajo/pullserver/FadvisedFileRegion.java
----------------------------------------------------------------------
diff --git a/tajo-core/tajo-core-pullserver/src/main/java/org/apache/tajo/pullserver/FadvisedFileRegion.java b/tajo-core/tajo-core-pullserver/src/main/java/org/apache/tajo/pullserver/FadvisedFileRegion.java
index 39c2938..81e0bf1 100644
--- a/tajo-core/tajo-core-pullserver/src/main/java/org/apache/tajo/pullserver/FadvisedFileRegion.java
+++ b/tajo-core/tajo-core-pullserver/src/main/java/org/apache/tajo/pullserver/FadvisedFileRegion.java
@@ -22,7 +22,7 @@ import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.hadoop.io.ReadaheadPool;
import org.apache.hadoop.io.ReadaheadPool.ReadaheadRequest;
-import org.apache.hadoop.io.nativeio.NativeIO;
+import org.apache.hadoop.io.nativeio.NativeIO.POSIX;
import org.jboss.netty.channel.DefaultFileRegion;
import java.io.FileDescriptor;
@@ -71,8 +71,8 @@ public class FadvisedFileRegion extends DefaultFileRegion {
}
if (manageOsCache && getCount() > 0) {
try {
- NativeIO.posixFadviseIfPossible(fd, getPosition(), getCount(),
- NativeIO.POSIX_FADV_DONTNEED);
+ POSIX.posixFadviseIfPossible(identifier, fd, getPosition(), getCount(),
+ POSIX.POSIX_FADV_DONTNEED);
} catch (Throwable t) {
LOG.warn("Failed to manage OS cache for " + identifier, t);
}
http://git-wip-us.apache.org/repos/asf/incubator-tajo/blob/ecc749e8/tajo-core/tajo-core-pullserver/src/main/java/org/apache/tajo/pullserver/PullServerAuxService.java
----------------------------------------------------------------------
diff --git a/tajo-core/tajo-core-pullserver/src/main/java/org/apache/tajo/pullserver/PullServerAuxService.java b/tajo-core/tajo-core-pullserver/src/main/java/org/apache/tajo/pullserver/PullServerAuxService.java
index 9574fc8..1531fad 100644
--- a/tajo-core/tajo-core-pullserver/src/main/java/org/apache/tajo/pullserver/PullServerAuxService.java
+++ b/tajo-core/tajo-core-pullserver/src/main/java/org/apache/tajo/pullserver/PullServerAuxService.java
@@ -40,8 +40,9 @@ import org.apache.hadoop.metrics2.lib.MutableCounterLong;
import org.apache.hadoop.metrics2.lib.MutableGaugeInt;
import org.apache.hadoop.security.ssl.SSLFactory;
import org.apache.hadoop.yarn.api.records.ApplicationId;
-import org.apache.hadoop.yarn.server.nodemanager.containermanager.AuxServices;
-import org.apache.hadoop.yarn.service.AbstractService;
+import org.apache.hadoop.yarn.server.api.ApplicationInitializationContext;
+import org.apache.hadoop.yarn.server.api.ApplicationTerminationContext;
+import org.apache.hadoop.yarn.server.api.AuxiliaryService;
import org.apache.tajo.QueryId;
import org.apache.tajo.catalog.Schema;
import org.apache.tajo.conf.TajoConf;
@@ -88,8 +89,7 @@ import static org.jboss.netty.handler.codec.http.HttpMethod.GET;
import static org.jboss.netty.handler.codec.http.HttpResponseStatus.*;
import static org.jboss.netty.handler.codec.http.HttpVersion.HTTP_1_1;
-public class PullServerAuxService extends AbstractService
- implements AuxServices.AuxiliaryService {
+public class PullServerAuxService extends AuxiliaryService {
private static final Log LOG = LogFactory.getLog(PullServerAuxService.class);
@@ -188,18 +188,18 @@ public class PullServerAuxService extends AbstractService
}
@Override
- public void initApp(String user, ApplicationId appId, ByteBuffer secret) {
+ public void initializeApplication(ApplicationInitializationContext appInitContext) {
// TODO these bytes should be versioned
// TODO: Once SHuffle is out of NM, this can use MR APIs
- this.appId = appId;
+ this.appId = appInitContext.getApplicationId();
this.queryId = TajoIdUtils.parseQueryId(appId.toString());
- this.userName = user;
- userRsrc.put(appId.toString(), user);
+ this.userName = appInitContext.getUser();
+ userRsrc.put(this.appId.toString(), this.userName);
}
@Override
- public void stopApp(ApplicationId appId) {
- userRsrc.remove(appId.toString());
+ public void stopApplication(ApplicationTerminationContext appStopContext) {
+ userRsrc.remove(appStopContext.getApplicationId().toString());
}
@Override
@@ -275,7 +275,7 @@ public class PullServerAuxService extends AbstractService
}
@Override
- public synchronized ByteBuffer getMeta() {
+ public synchronized ByteBuffer getMetaData() {
try {
return serializeMetaData(port);
} catch (IOException e) {
http://git-wip-us.apache.org/repos/asf/incubator-tajo/blob/ecc749e8/tajo-core/tajo-core-pullserver/src/main/java/org/apache/tajo/pullserver/TajoPullServerService.java
----------------------------------------------------------------------
diff --git a/tajo-core/tajo-core-pullserver/src/main/java/org/apache/tajo/pullserver/TajoPullServerService.java b/tajo-core/tajo-core-pullserver/src/main/java/org/apache/tajo/pullserver/TajoPullServerService.java
index d67efc1..0f96e2c 100644
--- a/tajo-core/tajo-core-pullserver/src/main/java/org/apache/tajo/pullserver/TajoPullServerService.java
+++ b/tajo-core/tajo-core-pullserver/src/main/java/org/apache/tajo/pullserver/TajoPullServerService.java
@@ -39,8 +39,8 @@ import org.apache.hadoop.metrics2.lib.MutableCounterInt;
import org.apache.hadoop.metrics2.lib.MutableCounterLong;
import org.apache.hadoop.metrics2.lib.MutableGaugeInt;
import org.apache.hadoop.security.ssl.SSLFactory;
+import org.apache.hadoop.service.AbstractService;
import org.apache.hadoop.yarn.api.records.ApplicationId;
-import org.apache.hadoop.yarn.service.AbstractService;
import org.apache.tajo.catalog.Schema;
import org.apache.tajo.conf.TajoConf;
import org.apache.tajo.conf.TajoConf.ConfVars;