You are viewing a plain text version of this content. The canonical link for it is here.
Posted to common-commits@hadoop.apache.org by ju...@apache.org on 2015/03/19 19:33:31 UTC
[4/4] hadoop git commit: YARN-3333. Rename TimelineAggregator etc. to
TimelineCollector. Contributed by Sangjin Lee
YARN-3333. Rename TimelineAggregator etc. to TimelineCollector. Contributed by Sangjin Lee
Project: http://git-wip-us.apache.org/repos/asf/hadoop/repo
Commit: http://git-wip-us.apache.org/repos/asf/hadoop/commit/dda84085
Tree: http://git-wip-us.apache.org/repos/asf/hadoop/tree/dda84085
Diff: http://git-wip-us.apache.org/repos/asf/hadoop/diff/dda84085
Branch: refs/heads/YARN-2928
Commit: dda84085cabd8fdf143b380e54e1730802fd9912
Parents: 8a63791
Author: Junping Du <ju...@apache.org>
Authored: Thu Mar 19 11:49:07 2015 -0700
Committer: Junping Du <ju...@apache.org>
Committed: Thu Mar 19 11:49:07 2015 -0700
----------------------------------------------------------------------
hadoop-yarn-project/CHANGES.txt | 5 +-
.../hadoop-yarn/hadoop-yarn-api/pom.xml | 4 +
.../api/protocolrecords/AllocateResponse.java | 20 +-
.../timelineservice/TimelineWriteResponse.java | 20 +-
.../hadoop/yarn/conf/YarnConfiguration.java | 20 +-
.../src/main/proto/yarn_service_protos.proto | 2 +-
.../pom.xml | 10 +
.../distributedshell/ApplicationMaster.java | 62 ++---
.../applications/distributedshell/Client.java | 8 +-
.../distributedshell/TestDistributedShell.java | 10 +-
.../hadoop/yarn/client/api/AMRMClient.java | 6 +-
.../yarn/client/api/async/AMRMClientAsync.java | 4 +-
.../api/async/impl/AMRMClientAsyncImpl.java | 20 +-
.../impl/pb/AllocateResponsePBImpl.java | 16 +-
.../hadoop/yarn/client/api/TimelineClient.java | 2 +-
.../client/api/impl/TimelineClientImpl.java | 32 +--
.../src/main/resources/yarn-default.xml | 14 +-
.../hadoop/yarn/TestContainerLaunchRPC.java | 2 +-
.../hadoop/yarn/api/TestAllocateResponse.java | 12 +-
.../hadoop-yarn-server-common/pom.xml | 2 +-
.../api/AggregatorNodemanagerProtocol.java | 56 ----
.../api/AggregatorNodemanagerProtocolPB.java | 33 ---
.../api/CollectorNodemanagerProtocol.java | 57 ++++
.../api/CollectorNodemanagerProtocolPB.java | 33 +++
...gregatorNodemanagerProtocolPBClientImpl.java | 94 -------
...ollectorNodemanagerProtocolPBClientImpl.java | 94 +++++++
...regatorNodemanagerProtocolPBServiceImpl.java | 61 ----
...llectorNodemanagerProtocolPBServiceImpl.java | 59 ++++
.../protocolrecords/NodeHeartbeatRequest.java | 16 +-
.../protocolrecords/NodeHeartbeatResponse.java | 8 +-
.../ReportNewAggregatorsInfoRequest.java | 53 ----
.../ReportNewAggregatorsInfoResponse.java | 32 ---
.../ReportNewCollectorInfoRequest.java | 53 ++++
.../ReportNewCollectorInfoResponse.java | 32 +++
.../impl/pb/NodeHeartbeatRequestPBImpl.java | 62 ++---
.../impl/pb/NodeHeartbeatResponsePBImpl.java | 60 ++--
.../ReportNewAggregatorsInfoRequestPBImpl.java | 142 ----------
.../ReportNewAggregatorsInfoResponsePBImpl.java | 74 -----
.../pb/ReportNewCollectorInfoRequestPBImpl.java | 142 ++++++++++
.../ReportNewCollectorInfoResponsePBImpl.java | 74 +++++
.../server/api/records/AppAggregatorsMap.java | 33 ---
.../server/api/records/AppCollectorsMap.java | 46 +++
.../impl/pb/AppAggregatorsMapPBImpl.java | 151 ----------
.../records/impl/pb/AppCollectorsMapPBImpl.java | 151 ++++++++++
.../proto/aggregatornodemanager_protocol.proto | 29 --
.../proto/collectornodemanager_protocol.proto | 29 ++
.../yarn_server_common_service_protos.proto | 18 +-
.../java/org/apache/hadoop/yarn/TestRPC.java | 116 ++++----
.../hadoop/yarn/TestYarnServerApiClasses.java | 24 +-
.../hadoop/yarn/server/nodemanager/Context.java | 14 +-
.../yarn/server/nodemanager/NodeManager.java | 60 ++--
.../nodemanager/NodeStatusUpdaterImpl.java | 11 +-
.../aggregatormanager/NMAggregatorService.java | 113 --------
.../collectormanager/NMCollectorService.java | 110 ++++++++
.../application/ApplicationImpl.java | 9 +-
.../ApplicationMasterService.java | 12 +-
.../resourcemanager/ResourceTrackerService.java | 72 ++---
.../server/resourcemanager/rmapp/RMApp.java | 22 +-
.../rmapp/RMAppAggregatorUpdateEvent.java | 36 ---
.../rmapp/RMAppCollectorUpdateEvent.java | 37 +++
.../resourcemanager/rmapp/RMAppEventType.java | 4 +-
.../server/resourcemanager/rmapp/RMAppImpl.java | 60 ++--
.../applicationsmanager/MockAsm.java | 6 +-
.../server/resourcemanager/rmapp/MockRMApp.java | 8 +-
.../hadoop-yarn-server-tests/pom.xml | 5 +
.../TestTimelineServiceClientIntegration.java | 52 +++-
.../hadoop-yarn-server-timelineservice/pom.xml | 10 +
.../aggregator/AppLevelTimelineAggregator.java | 57 ----
.../PerNodeTimelineAggregatorsAuxService.java | 211 --------------
.../aggregator/TimelineAggregator.java | 122 --------
.../TimelineAggregatorWebService.java | 180 ------------
.../TimelineAggregatorsCollection.java | 271 ------------------
.../collector/AppLevelTimelineCollector.java | 57 ++++
.../PerNodeTimelineCollectorsAuxService.java | 214 ++++++++++++++
.../collector/TimelineCollector.java | 122 ++++++++
.../collector/TimelineCollectorManager.java | 278 +++++++++++++++++++
.../collector/TimelineCollectorWebService.java | 183 ++++++++++++
.../storage/FileSystemTimelineWriterImpl.java | 6 +-
.../TestAppLevelTimelineAggregator.java | 23 --
...estPerNodeTimelineAggregatorsAuxService.java | 150 ----------
.../TestTimelineAggregatorsCollection.java | 109 --------
.../TestAppLevelTimelineCollector.java | 23 ++
...TestPerNodeTimelineCollectorsAuxService.java | 164 +++++++++++
.../collector/TestTimelineCollectorManager.java | 118 ++++++++
.../TestFileSystemTimelineWriterImpl.java | 43 ++-
85 files changed, 2597 insertions(+), 2478 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/hadoop/blob/dda84085/hadoop-yarn-project/CHANGES.txt
----------------------------------------------------------------------
diff --git a/hadoop-yarn-project/CHANGES.txt b/hadoop-yarn-project/CHANGES.txt
index 47351c6..216ae77 100644
--- a/hadoop-yarn-project/CHANGES.txt
+++ b/hadoop-yarn-project/CHANGES.txt
@@ -32,13 +32,16 @@ Branch YARN-2928: Timeline Server Next Generation: Phase 1
YARN-3039. Implemented the app-level timeline aggregator discovery service.
(Junping Du via zjshen)
+ YARN-3333. Rename TimelineAggregator etc. to TimelineCollector. (Sangjin Lee
+ via junping_du)
+
IMPROVEMENTS
OPTIMIZATIONS
BUG FIXES
-Trunk - Unreleased
+Trunk - Unreleased
INCOMPATIBLE CHANGES
http://git-wip-us.apache.org/repos/asf/hadoop/blob/dda84085/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-api/pom.xml
----------------------------------------------------------------------
diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-api/pom.xml b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-api/pom.xml
index a763d39..edb42a3 100644
--- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-api/pom.xml
+++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-api/pom.xml
@@ -45,6 +45,10 @@
<groupId>commons-logging</groupId>
<artifactId>commons-logging</artifactId>
</dependency>
+ <dependency>
+ <groupId>javax.xml.bind</groupId>
+ <artifactId>jaxb-api</artifactId>
+ </dependency>
<!-- 'mvn dependency:analyze' fails to detect use of this dependency -->
<dependency>
http://git-wip-us.apache.org/repos/asf/hadoop/blob/dda84085/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-api/src/main/java/org/apache/hadoop/yarn/api/protocolrecords/AllocateResponse.java
----------------------------------------------------------------------
diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-api/src/main/java/org/apache/hadoop/yarn/api/protocolrecords/AllocateResponse.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-api/src/main/java/org/apache/hadoop/yarn/api/protocolrecords/AllocateResponse.java
index 421c2a0..6703249 100644
--- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-api/src/main/java/org/apache/hadoop/yarn/api/protocolrecords/AllocateResponse.java
+++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-api/src/main/java/org/apache/hadoop/yarn/api/protocolrecords/AllocateResponse.java
@@ -120,7 +120,7 @@ public abstract class AllocateResponse {
response.setAMRMToken(amRMToken);
return response;
}
-
+
@Public
@Unstable
public static AllocateResponse newInstance(int responseId,
@@ -130,13 +130,13 @@ public abstract class AllocateResponse {
PreemptionMessage preempt, List<NMToken> nmTokens, Token amRMToken,
List<ContainerResourceIncrease> increasedContainers,
List<ContainerResourceDecrease> decreasedContainers,
- String aggregatorAddr) {
+ String collectorAddr) {
AllocateResponse response =
newInstance(responseId, completedContainers, allocatedContainers,
updatedNodes, availResources, command, numClusterNodes, preempt,
nmTokens, increasedContainers, decreasedContainers);
response.setAMRMToken(amRMToken);
- response.setAggregatorAddr(aggregatorAddr);
+ response.setCollectorAddr(collectorAddr);
return response;
}
@@ -323,18 +323,18 @@ public abstract class AllocateResponse {
@Private
@Unstable
public abstract void setAMRMToken(Token amRMToken);
-
+
/**
- * The address of aggregator that belong to this app
+ * The address of collector that belong to this app
*
- * @return The address of aggregator that belong to this attempt
+ * @return The address of collector that belong to this attempt
*/
@Public
@Unstable
- public abstract String getAggregatorAddr();
-
+ public abstract String getCollectorAddr();
+
@Private
@Unstable
- public abstract void setAggregatorAddr(String aggregatorAddr);
-
+ public abstract void setCollectorAddr(String collectorAddr);
+
}
http://git-wip-us.apache.org/repos/asf/hadoop/blob/dda84085/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-api/src/main/java/org/apache/hadoop/yarn/api/records/timelineservice/TimelineWriteResponse.java
----------------------------------------------------------------------
diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-api/src/main/java/org/apache/hadoop/yarn/api/records/timelineservice/TimelineWriteResponse.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-api/src/main/java/org/apache/hadoop/yarn/api/records/timelineservice/TimelineWriteResponse.java
index 82ecdbd..4739d8f 100644
--- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-api/src/main/java/org/apache/hadoop/yarn/api/records/timelineservice/TimelineWriteResponse.java
+++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-api/src/main/java/org/apache/hadoop/yarn/api/records/timelineservice/TimelineWriteResponse.java
@@ -47,7 +47,7 @@ public class TimelineWriteResponse {
/**
* Get a list of {@link TimelineWriteError} instances
- *
+ *
* @return a list of {@link TimelineWriteError} instances
*/
@XmlElement(name = "errors")
@@ -57,7 +57,7 @@ public class TimelineWriteResponse {
/**
* Add a single {@link TimelineWriteError} instance into the existing list
- *
+ *
* @param error
* a single {@link TimelineWriteError} instance
*/
@@ -67,7 +67,7 @@ public class TimelineWriteResponse {
/**
* Add a list of {@link TimelineWriteError} instances into the existing list
- *
+ *
* @param errors
* a list of {@link TimelineWriteError} instances
*/
@@ -77,7 +77,7 @@ public class TimelineWriteResponse {
/**
* Set the list to the given list of {@link TimelineWriteError} instances
- *
+ *
* @param errors
* a list of {@link TimelineWriteError} instances
*/
@@ -107,7 +107,7 @@ public class TimelineWriteResponse {
/**
* Get the entity Id
- *
+ *
* @return the entity Id
*/
@XmlElement(name = "entity")
@@ -117,7 +117,7 @@ public class TimelineWriteResponse {
/**
* Set the entity Id
- *
+ *
* @param entityId
* the entity Id
*/
@@ -127,7 +127,7 @@ public class TimelineWriteResponse {
/**
* Get the entity type
- *
+ *
* @return the entity type
*/
@XmlElement(name = "entitytype")
@@ -137,7 +137,7 @@ public class TimelineWriteResponse {
/**
* Set the entity type
- *
+ *
* @param entityType
* the entity type
*/
@@ -147,7 +147,7 @@ public class TimelineWriteResponse {
/**
* Get the error code
- *
+ *
* @return an error code
*/
@XmlElement(name = "errorcode")
@@ -157,7 +157,7 @@ public class TimelineWriteResponse {
/**
* Set the error code to the given error code
- *
+ *
* @param errorCode
* an error code
*/
http://git-wip-us.apache.org/repos/asf/hadoop/blob/dda84085/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-api/src/main/java/org/apache/hadoop/yarn/conf/YarnConfiguration.java
----------------------------------------------------------------------
diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-api/src/main/java/org/apache/hadoop/yarn/conf/YarnConfiguration.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-api/src/main/java/org/apache/hadoop/yarn/conf/YarnConfiguration.java
index a987044..73f11b8 100644
--- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-api/src/main/java/org/apache/hadoop/yarn/conf/YarnConfiguration.java
+++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-api/src/main/java/org/apache/hadoop/yarn/conf/YarnConfiguration.java
@@ -644,10 +644,10 @@ public class YarnConfiguration extends Configuration {
public static final int DEFAULT_NM_CONTAINER_MGR_THREAD_COUNT = 20;
/** Number of threads container manager uses.*/
- public static final String NM_AGGREGATOR_SERVICE_THREAD_COUNT =
- NM_PREFIX + "aggregator-service.thread-count";
- public static final int DEFAULT_NM_AGGREGATOR_SERVICE_THREAD_COUNT = 5;
-
+ public static final String NM_COLLECTOR_SERVICE_THREAD_COUNT =
+ NM_PREFIX + "collector-service.thread-count";
+ public static final int DEFAULT_NM_COLLECTOR_SERVICE_THREAD_COUNT = 5;
+
/** Number of threads used in cleanup.*/
public static final String NM_DELETE_THREAD_COUNT =
NM_PREFIX + "delete.thread-count";
@@ -675,13 +675,13 @@ public class YarnConfiguration extends Configuration {
public static final String DEFAULT_NM_LOCALIZER_ADDRESS = "0.0.0.0:" +
DEFAULT_NM_LOCALIZER_PORT;
- /** Address where the aggregator service IPC is.*/
- public static final String NM_AGGREGATOR_SERVICE_ADDRESS =
- NM_PREFIX + "aggregator-service.address";
- public static final int DEFAULT_NM_AGGREGATOR_SERVICE_PORT = 8048;
- public static final String DEFAULT_NM_AGGREGATOR_SERVICE_ADDRESS =
+ /** Address where the collector service IPC is.*/
+ public static final String NM_COLLECTOR_SERVICE_ADDRESS =
+ NM_PREFIX + "collector-service.address";
+ public static final int DEFAULT_NM_COLLECTOR_SERVICE_PORT = 8048;
+ public static final String DEFAULT_NM_COLLECTOR_SERVICE_ADDRESS =
"0.0.0.0:" + DEFAULT_NM_LOCALIZER_PORT;
-
+
/** Interval in between cache cleanups.*/
public static final String NM_LOCALIZER_CACHE_CLEANUP_INTERVAL_MS =
NM_PREFIX + "localizer.cache.cleanup.interval-ms";
http://git-wip-us.apache.org/repos/asf/hadoop/blob/dda84085/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-api/src/main/proto/yarn_service_protos.proto
----------------------------------------------------------------------
diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-api/src/main/proto/yarn_service_protos.proto b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-api/src/main/proto/yarn_service_protos.proto
index 4ae4806..95aca85 100644
--- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-api/src/main/proto/yarn_service_protos.proto
+++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-api/src/main/proto/yarn_service_protos.proto
@@ -87,7 +87,7 @@ message AllocateResponseProto {
repeated ContainerResourceIncreaseProto increased_containers = 10;
repeated ContainerResourceDecreaseProto decreased_containers = 11;
optional hadoop.common.TokenProto am_rm_token = 12;
- optional string aggregator_addr = 13;
+ optional string collector_addr = 13;
}
enum SchedulerResourceTypes {
http://git-wip-us.apache.org/repos/asf/hadoop/blob/dda84085/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-applications/hadoop-yarn-applications-distributedshell/pom.xml
----------------------------------------------------------------------
diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-applications/hadoop-yarn-applications-distributedshell/pom.xml b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-applications/hadoop-yarn-applications-distributedshell/pom.xml
index d91c67b..a0087dc 100644
--- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-applications/hadoop-yarn-applications-distributedshell/pom.xml
+++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-applications/hadoop-yarn-applications-distributedshell/pom.xml
@@ -70,6 +70,16 @@
<groupId>commons-io</groupId>
<artifactId>commons-io</artifactId>
</dependency>
+ <dependency>
+ <groupId>org.apache.hadoop</groupId>
+ <artifactId>hadoop-yarn-server-applicationhistoryservice</artifactId>
+ <scope>test</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.hadoop</groupId>
+ <artifactId>hadoop-yarn-server-timelineservice</artifactId>
+ <scope>test</scope>
+ </dependency>
<!-- 'mvn dependency:analyze' fails to detect use of this dependency -->
<dependency>
<groupId>org.apache.hadoop</groupId>
http://git-wip-us.apache.org/repos/asf/hadoop/blob/dda84085/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-applications/hadoop-yarn-applications-distributedshell/src/main/java/org/apache/hadoop/yarn/applications/distributedshell/ApplicationMaster.java
----------------------------------------------------------------------
diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-applications/hadoop-yarn-applications-distributedshell/src/main/java/org/apache/hadoop/yarn/applications/distributedshell/ApplicationMaster.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-applications/hadoop-yarn-applications-distributedshell/src/main/java/org/apache/hadoop/yarn/applications/distributedshell/ApplicationMaster.java
index 3a19ac2..fe35a8c 100644
--- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-applications/hadoop-yarn-applications-distributedshell/src/main/java/org/apache/hadoop/yarn/applications/distributedshell/ApplicationMaster.java
+++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-applications/hadoop-yarn-applications-distributedshell/src/main/java/org/apache/hadoop/yarn/applications/distributedshell/ApplicationMaster.java
@@ -211,12 +211,12 @@ public class ApplicationMaster {
private int appMasterRpcPort = -1;
// Tracking url to which app master publishes info for clients to monitor
private String appMasterTrackingUrl = "";
-
+
private boolean newTimelineService = false;
-
+
// For posting entities in new timeline service in a non-blocking way
// TODO replace with event loop in TimelineClient.
- private static ExecutorService threadPool =
+ private static ExecutorService threadPool =
Executors.newCachedThreadPool(
new ThreadFactoryBuilder().setNameFormat("TimelineService #%d")
.build());
@@ -302,9 +302,9 @@ public class ApplicationMaster {
}
appMaster.run();
result = appMaster.finish();
-
+
threadPool.shutdown();
-
+
while (!threadPool.isTerminated()) { // wait for all posting thread to finish
try {
if (!threadPool.awaitTermination(30, TimeUnit.SECONDS)) {
@@ -386,7 +386,7 @@ public class ApplicationMaster {
"No. of containers on which the shell command needs to be executed");
opts.addOption("priority", true, "Application Priority. Default 0");
opts.addOption("debug", false, "Dump out debug information");
- opts.addOption("timeline_service_version", true,
+ opts.addOption("timeline_service_version", true,
"Version for timeline service");
opts.addOption("help", false, "Print usage");
CommandLine cliParser = new GnuParser().parse(opts, args);
@@ -527,7 +527,7 @@ public class ApplicationMaster {
if (conf.getBoolean(YarnConfiguration.TIMELINE_SERVICE_ENABLED,
YarnConfiguration.DEFAULT_TIMELINE_SERVICE_ENABLED)) {
if (cliParser.hasOption("timeline_service_version")) {
- String timelineServiceVersion =
+ String timelineServiceVersion =
cliParser.getOptionValue("timeline_service_version", "v1");
if (timelineServiceVersion.trim().equalsIgnoreCase("v1")) {
newTimelineService = false;
@@ -605,8 +605,8 @@ public class ApplicationMaster {
if(timelineClient != null) {
if (newTimelineService) {
- publishApplicationAttemptEventOnNewTimelineService(timelineClient,
- appAttemptID.toString(), DSEvent.DS_APP_ATTEMPT_START, domainId,
+ publishApplicationAttemptEventOnNewTimelineService(timelineClient,
+ appAttemptID.toString(), DSEvent.DS_APP_ATTEMPT_START, domainId,
appSubmitterUgi);
} else {
publishApplicationAttemptEvent(timelineClient, appAttemptID.toString(),
@@ -620,7 +620,7 @@ public class ApplicationMaster {
// need to bind timelineClient before start.
amRMClient.registerTimelineClient(timelineClient);
amRMClient.start();
-
+
containerListener = createNMCallbackHandler();
nmClientAsync = new NMClientAsyncImpl(containerListener);
nmClientAsync.init(conf);
@@ -683,8 +683,8 @@ public class ApplicationMaster {
if(timelineClient != null) {
if (newTimelineService) {
- publishApplicationAttemptEventOnNewTimelineService(timelineClient,
- appAttemptID.toString(), DSEvent.DS_APP_ATTEMPT_START, domainId,
+ publishApplicationAttemptEventOnNewTimelineService(timelineClient,
+ appAttemptID.toString(), DSEvent.DS_APP_ATTEMPT_START, domainId,
appSubmitterUgi);
} else {
publishApplicationAttemptEvent(timelineClient, appAttemptID.toString(),
@@ -757,7 +757,7 @@ public class ApplicationMaster {
if(timelineClient != null) {
timelineClient.stop();
}
-
+
return success;
}
@@ -1220,13 +1220,13 @@ public class ApplicationMaster {
e instanceof UndeclaredThrowableException ? e.getCause() : e);
}
}
-
+
private static void publishContainerStartEventOnNewTimelineService(
- final TimelineClient timelineClient, final Container container,
+ final TimelineClient timelineClient, final Container container,
final String domainId, final UserGroupInformation ugi) {
Runnable publishWrapper = new Runnable() {
public void run() {
- publishContainerStartEventOnNewTimelineServiceBase(timelineClient,
+ publishContainerStartEventOnNewTimelineServiceBase(timelineClient,
container, domainId, ugi);
}
};
@@ -1236,14 +1236,14 @@ public class ApplicationMaster {
private static void publishContainerStartEventOnNewTimelineServiceBase(
final TimelineClient timelineClient, Container container, String domainId,
UserGroupInformation ugi) {
- final org.apache.hadoop.yarn.api.records.timelineservice.TimelineEntity entity =
+ final org.apache.hadoop.yarn.api.records.timelineservice.TimelineEntity entity =
new org.apache.hadoop.yarn.api.records.timelineservice.TimelineEntity();
entity.setId(container.getId().toString());
entity.setType(DSEntity.DS_CONTAINER.toString());
//entity.setDomainId(domainId);
entity.addInfo("user", ugi.getShortUserName());
-
- org.apache.hadoop.yarn.api.records.timelineservice.TimelineEvent event =
+
+ org.apache.hadoop.yarn.api.records.timelineservice.TimelineEvent event =
new org.apache.hadoop.yarn.api.records.timelineservice.TimelineEvent();
event.setTimestamp(System.currentTimeMillis());
event.setId(DSEvent.DS_CONTAINER_START.toString());
@@ -1265,29 +1265,29 @@ public class ApplicationMaster {
e instanceof UndeclaredThrowableException ? e.getCause() : e);
}
}
-
+
private static void publishContainerEndEventOnNewTimelineService(
final TimelineClient timelineClient, final ContainerStatus container,
final String domainId, final UserGroupInformation ugi) {
Runnable publishWrapper = new Runnable() {
public void run() {
- publishContainerEndEventOnNewTimelineServiceBase(timelineClient,
+ publishContainerEndEventOnNewTimelineServiceBase(timelineClient,
container, domainId, ugi);
}
};
threadPool.execute(publishWrapper);
}
-
+
private static void publishContainerEndEventOnNewTimelineServiceBase(
final TimelineClient timelineClient, final ContainerStatus container,
final String domainId, final UserGroupInformation ugi) {
- final org.apache.hadoop.yarn.api.records.timelineservice.TimelineEntity entity =
+ final org.apache.hadoop.yarn.api.records.timelineservice.TimelineEntity entity =
new org.apache.hadoop.yarn.api.records.timelineservice.TimelineEntity();
entity.setId(container.getContainerId().toString());
entity.setType(DSEntity.DS_CONTAINER.toString());
//entity.setDomainId(domainId);
entity.addInfo("user", ugi.getShortUserName());
- org.apache.hadoop.yarn.api.records.timelineservice.TimelineEvent event =
+ org.apache.hadoop.yarn.api.records.timelineservice.TimelineEvent event =
new org.apache.hadoop.yarn.api.records.timelineservice.TimelineEvent();
event.setTimestamp(System.currentTimeMillis());
event.setId(DSEvent.DS_CONTAINER_END.toString());
@@ -1312,28 +1312,28 @@ public class ApplicationMaster {
private static void publishApplicationAttemptEventOnNewTimelineService(
final TimelineClient timelineClient, final String appAttemptId,
- final DSEvent appEvent, final String domainId,
+ final DSEvent appEvent, final String domainId,
final UserGroupInformation ugi) {
-
+
Runnable publishWrapper = new Runnable() {
public void run() {
- publishApplicationAttemptEventOnNewTimelineServiceBase(timelineClient,
+ publishApplicationAttemptEventOnNewTimelineServiceBase(timelineClient,
appAttemptId, appEvent, domainId, ugi);
}
};
threadPool.execute(publishWrapper);
}
-
+
private static void publishApplicationAttemptEventOnNewTimelineServiceBase(
final TimelineClient timelineClient, String appAttemptId,
DSEvent appEvent, String domainId, UserGroupInformation ugi) {
- final org.apache.hadoop.yarn.api.records.timelineservice.TimelineEntity entity =
+ final org.apache.hadoop.yarn.api.records.timelineservice.TimelineEntity entity =
new org.apache.hadoop.yarn.api.records.timelineservice.TimelineEntity();
entity.setId(appAttemptId);
entity.setType(DSEntity.DS_APP_ATTEMPT.toString());
//entity.setDomainId(domainId);
entity.addInfo("user", ugi.getShortUserName());
- org.apache.hadoop.yarn.api.records.timelineservice.TimelineEvent event =
+ org.apache.hadoop.yarn.api.records.timelineservice.TimelineEvent event =
new org.apache.hadoop.yarn.api.records.timelineservice.TimelineEvent();
event.setId(appEvent.toString());
event.setTimestamp(System.currentTimeMillis());
@@ -1355,5 +1355,5 @@ public class ApplicationMaster {
e instanceof UndeclaredThrowableException ? e.getCause() : e);
}
}
-
+
}
http://git-wip-us.apache.org/repos/asf/hadoop/blob/dda84085/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-applications/hadoop-yarn-applications-distributedshell/src/main/java/org/apache/hadoop/yarn/applications/distributedshell/Client.java
----------------------------------------------------------------------
diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-applications/hadoop-yarn-applications-distributedshell/src/main/java/org/apache/hadoop/yarn/applications/distributedshell/Client.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-applications/hadoop-yarn-applications-distributedshell/src/main/java/org/apache/hadoop/yarn/applications/distributedshell/Client.java
index 02a627e..934515e 100644
--- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-applications/hadoop-yarn-applications-distributedshell/src/main/java/org/apache/hadoop/yarn/applications/distributedshell/Client.java
+++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-applications/hadoop-yarn-applications-distributedshell/src/main/java/org/apache/hadoop/yarn/applications/distributedshell/Client.java
@@ -185,7 +185,7 @@ public class Client {
// Command line options
private Options opts;
-
+
private String timelineServiceVersion;
private static final String shellCommandPath = "shellCommands";
@@ -357,11 +357,11 @@ public class Client {
throw new IllegalArgumentException("Invalid virtual cores specified for application master, exiting."
+ " Specified virtual cores=" + amVCores);
}
-
+
if (cliParser.hasOption("timeline_service_version")) {
- timelineServiceVersion =
+ timelineServiceVersion =
cliParser.getOptionValue("timeline_service_version", "v1");
- if (! (timelineServiceVersion.trim().equalsIgnoreCase("v1") ||
+ if (! (timelineServiceVersion.trim().equalsIgnoreCase("v1") ||
timelineServiceVersion.trim().equalsIgnoreCase("v2"))) {
throw new IllegalArgumentException(
"timeline_service_version is not set properly, should be 'v1' or 'v2'");
http://git-wip-us.apache.org/repos/asf/hadoop/blob/dda84085/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-applications/hadoop-yarn-applications-distributedshell/src/test/java/org/apache/hadoop/yarn/applications/distributedshell/TestDistributedShell.java
----------------------------------------------------------------------
diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-applications/hadoop-yarn-applications-distributedshell/src/test/java/org/apache/hadoop/yarn/applications/distributedshell/TestDistributedShell.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-applications/hadoop-yarn-applications-distributedshell/src/test/java/org/apache/hadoop/yarn/applications/distributedshell/TestDistributedShell.java
index 97d9168..0af050c 100644
--- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-applications/hadoop-yarn-applications-distributedshell/src/test/java/org/apache/hadoop/yarn/applications/distributedshell/TestDistributedShell.java
+++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-applications/hadoop-yarn-applications-distributedshell/src/test/java/org/apache/hadoop/yarn/applications/distributedshell/TestDistributedShell.java
@@ -50,7 +50,7 @@ import org.apache.hadoop.yarn.client.api.YarnClient;
import org.apache.hadoop.yarn.conf.YarnConfiguration;
import org.apache.hadoop.yarn.server.MiniYARNCluster;
import org.apache.hadoop.yarn.server.resourcemanager.scheduler.capacity.CapacityScheduler;
-import org.apache.hadoop.yarn.server.timelineservice.aggregator.PerNodeTimelineAggregatorsAuxService;
+import org.apache.hadoop.yarn.server.timelineservice.collector.PerNodeTimelineCollectorsAuxService;
import org.apache.hadoop.yarn.server.timelineservice.storage.FileSystemTimelineWriterImpl;
import org.junit.After;
import org.junit.Assert;
@@ -67,7 +67,7 @@ public class TestDistributedShell {
protected MiniYARNCluster yarnCluster = null;
protected YarnConfiguration conf = null;
private static final int NUM_NMS = 1;
- private static final String TIMELINE_AUX_SERVICE_NAME = "timeline_aggregator";
+ private static final String TIMELINE_AUX_SERVICE_NAME = "timeline_collector";
protected final static String APPMASTER_JAR =
JarFinder.getJar(ApplicationMaster.class);
@@ -91,14 +91,14 @@ public class TestDistributedShell {
// mark if we need to launch the v1 timeline server
boolean enableATSV1 = false;
if (!currTestName.getMethodName().toLowerCase().contains("v2")) {
- // disable aux-service based timeline aggregators
+ // disable aux-service based timeline collectors
conf.set(YarnConfiguration.NM_AUX_SERVICES, "");
enableATSV1 = true;
} else {
- // enable aux-service based timeline aggregators
+ // enable aux-service based timeline collectors
conf.set(YarnConfiguration.NM_AUX_SERVICES, TIMELINE_AUX_SERVICE_NAME);
conf.set(YarnConfiguration.NM_AUX_SERVICES + "." + TIMELINE_AUX_SERVICE_NAME
- + ".class", PerNodeTimelineAggregatorsAuxService.class.getName());
+ + ".class", PerNodeTimelineCollectorsAuxService.class.getName());
}
conf.set(YarnConfiguration.RM_SCHEDULER, CapacityScheduler.class.getName());
conf.setBoolean(YarnConfiguration.NODE_LABELS_ENABLED, true);
http://git-wip-us.apache.org/repos/asf/hadoop/blob/dda84085/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-client/src/main/java/org/apache/hadoop/yarn/client/api/AMRMClient.java
----------------------------------------------------------------------
diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-client/src/main/java/org/apache/hadoop/yarn/client/api/AMRMClient.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-client/src/main/java/org/apache/hadoop/yarn/client/api/AMRMClient.java
index 56f2b10..afb2e09 100644
--- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-client/src/main/java/org/apache/hadoop/yarn/client/api/AMRMClient.java
+++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-client/src/main/java/org/apache/hadoop/yarn/client/api/AMRMClient.java
@@ -47,7 +47,7 @@ import com.google.common.collect.ImmutableList;
public abstract class AMRMClient<T extends AMRMClient.ContainerRequest> extends
AbstractService {
private static final Log LOG = LogFactory.getLog(AMRMClient.class);
-
+
private TimelineClient timelineClient;
/**
@@ -382,7 +382,7 @@ public abstract class AMRMClient<T extends AMRMClient.ContainerRequest> extends
public void registerTimelineClient(TimelineClient timelineClient) {
this.timelineClient = timelineClient;
}
-
+
/**
* Get registered timeline client.
* @return
@@ -390,7 +390,7 @@ public abstract class AMRMClient<T extends AMRMClient.ContainerRequest> extends
public TimelineClient getRegisteredTimeineClient() {
return this.timelineClient;
}
-
+
/**
* Wait for <code>check</code> to return true for each 1000 ms.
* See also {@link #waitFor(com.google.common.base.Supplier, int)}
http://git-wip-us.apache.org/repos/asf/hadoop/blob/dda84085/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-client/src/main/java/org/apache/hadoop/yarn/client/api/async/AMRMClientAsync.java
----------------------------------------------------------------------
diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-client/src/main/java/org/apache/hadoop/yarn/client/api/async/AMRMClientAsync.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-client/src/main/java/org/apache/hadoop/yarn/client/api/async/AMRMClientAsync.java
index be5610e..1a5c257 100644
--- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-client/src/main/java/org/apache/hadoop/yarn/client/api/async/AMRMClientAsync.java
+++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-client/src/main/java/org/apache/hadoop/yarn/client/api/async/AMRMClientAsync.java
@@ -194,7 +194,7 @@ extends AbstractService {
* @return Current number of nodes in the cluster
*/
public abstract int getClusterNodeCount();
-
+
/**
* Register TimelineClient to AMRMClient.
* @param timelineClient
@@ -202,7 +202,7 @@ extends AbstractService {
public void registerTimelineClient(TimelineClient timelineClient) {
client.registerTimelineClient(timelineClient);
}
-
+
/**
* Get registered timeline client.
* @return
http://git-wip-us.apache.org/repos/asf/hadoop/blob/dda84085/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-client/src/main/java/org/apache/hadoop/yarn/client/api/async/impl/AMRMClientAsyncImpl.java
----------------------------------------------------------------------
diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-client/src/main/java/org/apache/hadoop/yarn/client/api/async/impl/AMRMClientAsyncImpl.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-client/src/main/java/org/apache/hadoop/yarn/client/api/async/impl/AMRMClientAsyncImpl.java
index f0f0bc9..5351ef5 100644
--- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-client/src/main/java/org/apache/hadoop/yarn/client/api/async/impl/AMRMClientAsyncImpl.java
+++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-client/src/main/java/org/apache/hadoop/yarn/client/api/async/impl/AMRMClientAsyncImpl.java
@@ -66,8 +66,8 @@ extends AMRMClientAsync<T> {
private volatile boolean keepRunning;
private volatile float progress;
- private volatile String aggregatorAddr;
-
+ private volatile String collectorAddr;
+
private volatile Throwable savedException;
public AMRMClientAsyncImpl(int intervalMs, CallbackHandler callbackHandler) {
@@ -307,15 +307,15 @@ extends AMRMClientAsync<T> {
if (!allocated.isEmpty()) {
handler.onContainersAllocated(allocated);
}
-
- String aggregatorAddress = response.getAggregatorAddr();
+
+ String collectorAddress = response.getCollectorAddr();
TimelineClient timelineClient = client.getRegisteredTimeineClient();
- if (timelineClient != null && aggregatorAddress != null
- && !aggregatorAddress.isEmpty()) {
- if (aggregatorAddr == null ||
- !aggregatorAddr.equals(aggregatorAddress)) {
- aggregatorAddr = aggregatorAddress;
- timelineClient.setTimelineServiceAddress(aggregatorAddress);
+ if (timelineClient != null && collectorAddress != null
+ && !collectorAddress.isEmpty()) {
+ if (collectorAddr == null ||
+ !collectorAddr.equals(collectorAddress)) {
+ collectorAddr = collectorAddress;
+ timelineClient.setTimelineServiceAddress(collectorAddress);
}
}
progress = handler.getProgress();
http://git-wip-us.apache.org/repos/asf/hadoop/blob/dda84085/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-common/src/main/java/org/apache/hadoop/yarn/api/protocolrecords/impl/pb/AllocateResponsePBImpl.java
----------------------------------------------------------------------
diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-common/src/main/java/org/apache/hadoop/yarn/api/protocolrecords/impl/pb/AllocateResponsePBImpl.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-common/src/main/java/org/apache/hadoop/yarn/api/protocolrecords/impl/pb/AllocateResponsePBImpl.java
index 605c29a..8dc473c 100644
--- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-common/src/main/java/org/apache/hadoop/yarn/api/protocolrecords/impl/pb/AllocateResponsePBImpl.java
+++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-common/src/main/java/org/apache/hadoop/yarn/api/protocolrecords/impl/pb/AllocateResponsePBImpl.java
@@ -384,22 +384,22 @@ public class AllocateResponsePBImpl extends AllocateResponse {
}
this.amrmToken = amRMToken;
}
-
+
@Override
- public String getAggregatorAddr() {
+ public String getCollectorAddr() {
AllocateResponseProtoOrBuilder p = viaProto ? proto : builder;
- return p.getAggregatorAddr();
+ return p.getCollectorAddr();
}
-
+
@Override
- public void setAggregatorAddr(String aggregatorAddr) {
+ public void setCollectorAddr(String collectorAddr) {
maybeInitBuilder();
- if (aggregatorAddr == null) {
- builder.clearAggregatorAddr();
+ if (collectorAddr == null) {
+ builder.clearCollectorAddr();
return;
}
- builder.setAggregatorAddr(aggregatorAddr);
+ builder.setCollectorAddr(collectorAddr);
}
private synchronized void initLocalIncreasedContainerList() {
http://git-wip-us.apache.org/repos/asf/hadoop/blob/dda84085/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-common/src/main/java/org/apache/hadoop/yarn/client/api/TimelineClient.java
----------------------------------------------------------------------
diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-common/src/main/java/org/apache/hadoop/yarn/client/api/TimelineClient.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-common/src/main/java/org/apache/hadoop/yarn/client/api/TimelineClient.java
index 5db347e..b4c3980 100644
--- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-common/src/main/java/org/apache/hadoop/yarn/client/api/TimelineClient.java
+++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-common/src/main/java/org/apache/hadoop/yarn/client/api/TimelineClient.java
@@ -185,5 +185,5 @@ public abstract class TimelineClient extends AbstractService {
* the timeline service address
*/
public abstract void setTimelineServiceAddress(String address);
-
+
}
http://git-wip-us.apache.org/repos/asf/hadoop/blob/dda84085/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-common/src/main/java/org/apache/hadoop/yarn/client/api/impl/TimelineClientImpl.java
----------------------------------------------------------------------
diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-common/src/main/java/org/apache/hadoop/yarn/client/api/impl/TimelineClientImpl.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-common/src/main/java/org/apache/hadoop/yarn/client/api/impl/TimelineClientImpl.java
index 407682d..f70cf48 100644
--- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-common/src/main/java/org/apache/hadoop/yarn/client/api/impl/TimelineClientImpl.java
+++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-common/src/main/java/org/apache/hadoop/yarn/client/api/impl/TimelineClientImpl.java
@@ -115,14 +115,14 @@ public class TimelineClientImpl extends TimelineClient {
private DelegationTokenAuthenticatedURL.Token token;
private UserGroupInformation authUgi;
private String doAsUser;
-
+
private volatile String timelineServiceAddress;
-
+
// Retry parameters for identifying new timeline service
// TODO consider to merge with connection retry
private int maxServiceRetries;
private long serviceRetryInterval;
-
+
private boolean newTimelineService = false;
@Private
@@ -322,7 +322,7 @@ public class TimelineClientImpl extends TimelineClient {
YarnConfiguration.DEFAULT_TIMELINE_SERVICE_WEBAPP_ADDRESS));
}
LOG.info("Timeline service address: " + getTimelineServiceAddress());
- }
+ }
super.serviceInit(conf);
}
@@ -373,16 +373,16 @@ public class TimelineClientImpl extends TimelineClient {
YarnException {
doPosting(domain, "domain");
}
-
+
// Used for new timeline service only
@Private
- public void putObjects(String path, MultivaluedMap<String, String> params,
+ public void putObjects(String path, MultivaluedMap<String, String> params,
Object obj) throws IOException, YarnException {
-
- // timelineServiceAddress could haven't be initialized yet
+
+ // timelineServiceAddress could haven't be initialized yet
// or stale (only for new timeline service)
int retries = pollTimelineServiceAddress(this.maxServiceRetries);
-
+
// timelineServiceAddress could be stale, add retry logic here.
boolean needRetry = true;
while (needRetry) {
@@ -399,13 +399,13 @@ public class TimelineClientImpl extends TimelineClient {
}
}
}
-
+
/**
* Check if reaching to maximum of retries.
* @param retries
* @param e
*/
- private void checkRetryWithSleep(int retries, Exception e) throws
+ private void checkRetryWithSleep(int retries, Exception e) throws
YarnException, IOException {
if (retries > 0) {
try {
@@ -415,8 +415,8 @@ public class TimelineClientImpl extends TimelineClient {
}
} else {
LOG.error(
- "TimelineClient has reached to max retry times :" +
- this.maxServiceRetries + " for service address: " +
+ "TimelineClient has reached to max retry times :" +
+ this.maxServiceRetries + " for service address: " +
timelineServiceAddress);
if (e instanceof YarnException) {
throw (YarnException)e;
@@ -487,12 +487,12 @@ public class TimelineClientImpl extends TimelineClient {
}
return resp;
}
-
+
@Override
public void setTimelineServiceAddress(String address) {
this.timelineServiceAddress = address;
}
-
+
private String getTimelineServiceAddress() {
return this.timelineServiceAddress;
}
@@ -629,7 +629,7 @@ public class TimelineClientImpl extends TimelineClient {
throw new YarnRuntimeException("Unknown resource type");
}
}
-
+
/**
* Poll TimelineServiceAddress for maximum of retries times if it is null
* @param retries
http://git-wip-us.apache.org/repos/asf/hadoop/blob/dda84085/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-common/src/main/resources/yarn-default.xml
----------------------------------------------------------------------
diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-common/src/main/resources/yarn-default.xml b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-common/src/main/resources/yarn-default.xml
index 226d8ce..9ac54ce 100644
--- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-common/src/main/resources/yarn-default.xml
+++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-common/src/main/resources/yarn-default.xml
@@ -714,10 +714,10 @@
<name>yarn.nodemanager.container-manager.thread-count</name>
<value>20</value>
</property>
-
+
<property>
- <description>Number of threads aggregator service uses.</description>
- <name>yarn.nodemanager.aggregator-service.thread-count</name>
+ <description>Number of threads collector service uses.</description>
+ <name>yarn.nodemanager.collector-service.thread-count</name>
<value>5</value>
</property>
@@ -788,11 +788,11 @@
<name>yarn.nodemanager.localizer.address</name>
<value>${yarn.nodemanager.hostname}:8040</value>
</property>
-
-
+
+
<property>
- <description>Address where the aggregator service IPC is.</description>
- <name>yarn.nodemanager.aggregator-service.address</name>
+ <description>Address where the collector service IPC is.</description>
+ <name>yarn.nodemanager.collector-service.address</name>
<value>${yarn.nodemanager.hostname}:8048</value>
</property>
http://git-wip-us.apache.org/repos/asf/hadoop/blob/dda84085/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-common/src/test/java/org/apache/hadoop/yarn/TestContainerLaunchRPC.java
----------------------------------------------------------------------
diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-common/src/test/java/org/apache/hadoop/yarn/TestContainerLaunchRPC.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-common/src/test/java/org/apache/hadoop/yarn/TestContainerLaunchRPC.java
index 26d6d04..515adc8 100644
--- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-common/src/test/java/org/apache/hadoop/yarn/TestContainerLaunchRPC.java
+++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-common/src/test/java/org/apache/hadoop/yarn/TestContainerLaunchRPC.java
@@ -131,7 +131,7 @@ public class TestContainerLaunchRPC {
Assert.fail("timeout exception should have occurred!");
}
-
+
public static Token newContainerToken(NodeId nodeId, byte[] password,
ContainerTokenIdentifier tokenIdentifier) {
// RPC layer client expects ip:port as service for tokens
http://git-wip-us.apache.org/repos/asf/hadoop/blob/dda84085/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-common/src/test/java/org/apache/hadoop/yarn/api/TestAllocateResponse.java
----------------------------------------------------------------------
diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-common/src/test/java/org/apache/hadoop/yarn/api/TestAllocateResponse.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-common/src/test/java/org/apache/hadoop/yarn/api/TestAllocateResponse.java
index ef0bdcc..5af0f11 100644
--- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-common/src/test/java/org/apache/hadoop/yarn/api/TestAllocateResponse.java
+++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-common/src/test/java/org/apache/hadoop/yarn/api/TestAllocateResponse.java
@@ -111,21 +111,21 @@ public class TestAllocateResponse {
Assert.assertEquals(0, r.getIncreasedContainers().size());
Assert.assertEquals(0, r.getDecreasedContainers().size());
}
-
+
@SuppressWarnings("deprecation")
@Test
- public void testAllocateResponseWithAggregatorAddress() {
- final String aggregatorAddr = "localhost:0";
+ public void testAllocateResponseWithCollectorAddress() {
+ final String collectorAddr = "localhost:0";
AllocateResponse r =
AllocateResponse.newInstance(3, new ArrayList<ContainerStatus>(),
new ArrayList<Container>(), new ArrayList<NodeReport>(), null,
- AMCommand.AM_RESYNC, 3, null, new ArrayList<NMToken>(), null,
- null, null, aggregatorAddr);
+ AMCommand.AM_RESYNC, 3, null, new ArrayList<NMToken>(), null,
+ null, null, collectorAddr);
AllocateResponseProto p = ((AllocateResponsePBImpl) r).getProto();
r = new AllocateResponsePBImpl(p);
// check value
- Assert.assertEquals(aggregatorAddr, r.getAggregatorAddr());
+ Assert.assertEquals(collectorAddr, r.getCollectorAddr());
}
}
http://git-wip-us.apache.org/repos/asf/hadoop/blob/dda84085/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-common/pom.xml
----------------------------------------------------------------------
diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-common/pom.xml b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-common/pom.xml
index d1e4acb..4fea04c 100644
--- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-common/pom.xml
+++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-common/pom.xml
@@ -142,7 +142,7 @@
<include>yarn_server_common_service_protos.proto</include>
<include>ResourceTracker.proto</include>
<include>SCMUploader.proto</include>
- <include>aggregatornodemanager_protocol.proto</include>
+ <include>collectornodemanager_protocol.proto</include>
</includes>
</source>
<output>${project.build.directory}/generated-sources/java</output>
http://git-wip-us.apache.org/repos/asf/hadoop/blob/dda84085/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-common/src/main/java/org/apache/hadoop/yarn/server/api/AggregatorNodemanagerProtocol.java
----------------------------------------------------------------------
diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-common/src/main/java/org/apache/hadoop/yarn/server/api/AggregatorNodemanagerProtocol.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-common/src/main/java/org/apache/hadoop/yarn/server/api/AggregatorNodemanagerProtocol.java
deleted file mode 100644
index 53bdb4e..0000000
--- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-common/src/main/java/org/apache/hadoop/yarn/server/api/AggregatorNodemanagerProtocol.java
+++ /dev/null
@@ -1,56 +0,0 @@
-/**
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-package org.apache.hadoop.yarn.server.api;
-
-import java.io.IOException;
-
-import org.apache.hadoop.classification.InterfaceAudience.Private;
-import org.apache.hadoop.yarn.exceptions.YarnException;
-import org.apache.hadoop.yarn.server.api.protocolrecords.ReportNewAggregatorsInfoRequest;
-import org.apache.hadoop.yarn.server.api.protocolrecords.ReportNewAggregatorsInfoResponse;
-
-/**
- * <p>The protocol between an <code>TimelineAggregatorsCollection</code> and a
- * <code>NodeManager</code> to report a new application aggregator get launched.
- * </p>
- *
- */
-@Private
-public interface AggregatorNodemanagerProtocol {
-
- /**
- *
- * <p>
- * The <code>TimelineAggregatorsCollection</code> provides a list of mapping
- * between application and aggregator's address in
- * {@link ReportNewAggregatorsInfoRequest} to a <code>NodeManager</code> to
- * <em>register</em> aggregator's info, include: applicationId and REST URI to
- * access aggregator. NodeManager will add them into registered aggregators
- * and register them into <code>ResourceManager</code> afterwards.
- * </p>
- *
- * @param request the request of registering a new aggregator or a list of aggregators
- * @return
- * @throws YarnException
- * @throws IOException
- */
- ReportNewAggregatorsInfoResponse reportNewAggregatorInfo(
- ReportNewAggregatorsInfoRequest request)
- throws YarnException, IOException;
-
-}
http://git-wip-us.apache.org/repos/asf/hadoop/blob/dda84085/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-common/src/main/java/org/apache/hadoop/yarn/server/api/AggregatorNodemanagerProtocolPB.java
----------------------------------------------------------------------
diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-common/src/main/java/org/apache/hadoop/yarn/server/api/AggregatorNodemanagerProtocolPB.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-common/src/main/java/org/apache/hadoop/yarn/server/api/AggregatorNodemanagerProtocolPB.java
deleted file mode 100644
index 4df80a5..0000000
--- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-common/src/main/java/org/apache/hadoop/yarn/server/api/AggregatorNodemanagerProtocolPB.java
+++ /dev/null
@@ -1,33 +0,0 @@
-/**
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-package org.apache.hadoop.yarn.server.api;
-
-import org.apache.hadoop.classification.InterfaceAudience.Private;
-import org.apache.hadoop.classification.InterfaceStability.Unstable;
-import org.apache.hadoop.ipc.ProtocolInfo;
-import org.apache.hadoop.yarn.proto.AggregatorNodemanagerProtocol.AggregatorNodemanagerProtocolService;
-
-@Private
-@Unstable
-@ProtocolInfo(
- protocolName = "org.apache.hadoop.yarn.server.api.AggregatorNodemanagerProtocolPB",
- protocolVersion = 1)
-public interface AggregatorNodemanagerProtocolPB extends
- AggregatorNodemanagerProtocolService.BlockingInterface {
-
-}
http://git-wip-us.apache.org/repos/asf/hadoop/blob/dda84085/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-common/src/main/java/org/apache/hadoop/yarn/server/api/CollectorNodemanagerProtocol.java
----------------------------------------------------------------------
diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-common/src/main/java/org/apache/hadoop/yarn/server/api/CollectorNodemanagerProtocol.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-common/src/main/java/org/apache/hadoop/yarn/server/api/CollectorNodemanagerProtocol.java
new file mode 100644
index 0000000..26c121a
--- /dev/null
+++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-common/src/main/java/org/apache/hadoop/yarn/server/api/CollectorNodemanagerProtocol.java
@@ -0,0 +1,57 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.hadoop.yarn.server.api;
+
+import java.io.IOException;
+
+import org.apache.hadoop.classification.InterfaceAudience.Private;
+import org.apache.hadoop.yarn.exceptions.YarnException;
+import org.apache.hadoop.yarn.server.api.protocolrecords.ReportNewCollectorInfoRequest;
+import org.apache.hadoop.yarn.server.api.protocolrecords.ReportNewCollectorInfoResponse;
+
+/**
+ * <p>The protocol between an <code>TimelineCollectorManager</code> and a
+ * <code>NodeManager</code> to report a new application collector get launched.
+ * </p>
+ *
+ */
+@Private
+public interface CollectorNodemanagerProtocol {
+
+ /**
+ *
+ * <p>
+ * The <code>TimelineCollectorManager</code> provides a list of mapping
+ * between application and collector's address in
+ * {@link ReportNewCollectorInfoRequest} to a <code>NodeManager</code> to
+ * <em>register</em> collector's info, include: applicationId and REST URI to
+ * access collector. NodeManager will add them into registered collectors
+ * and register them into <code>ResourceManager</code> afterwards.
+ * </p>
+ *
+ * @param request the request of registering a new collector or a list of
+ * collectors
+ * @return
+ * @throws YarnException
+ * @throws IOException
+ */
+ ReportNewCollectorInfoResponse reportNewCollectorInfo(
+ ReportNewCollectorInfoRequest request)
+ throws YarnException, IOException;
+
+}
http://git-wip-us.apache.org/repos/asf/hadoop/blob/dda84085/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-common/src/main/java/org/apache/hadoop/yarn/server/api/CollectorNodemanagerProtocolPB.java
----------------------------------------------------------------------
diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-common/src/main/java/org/apache/hadoop/yarn/server/api/CollectorNodemanagerProtocolPB.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-common/src/main/java/org/apache/hadoop/yarn/server/api/CollectorNodemanagerProtocolPB.java
new file mode 100644
index 0000000..655e989
--- /dev/null
+++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-common/src/main/java/org/apache/hadoop/yarn/server/api/CollectorNodemanagerProtocolPB.java
@@ -0,0 +1,33 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.hadoop.yarn.server.api;
+
+import org.apache.hadoop.classification.InterfaceAudience.Private;
+import org.apache.hadoop.classification.InterfaceStability.Unstable;
+import org.apache.hadoop.ipc.ProtocolInfo;
+import org.apache.hadoop.yarn.proto.CollectorNodemanagerProtocol.CollectorNodemanagerProtocolService;
+
+@Private
+@Unstable
+@ProtocolInfo(
+ protocolName = "org.apache.hadoop.yarn.server.api.CollectorNodemanagerProtocolPB",
+ protocolVersion = 1)
+public interface CollectorNodemanagerProtocolPB extends
+ CollectorNodemanagerProtocolService.BlockingInterface {
+
+}
http://git-wip-us.apache.org/repos/asf/hadoop/blob/dda84085/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-common/src/main/java/org/apache/hadoop/yarn/server/api/impl/pb/client/AggregatorNodemanagerProtocolPBClientImpl.java
----------------------------------------------------------------------
diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-common/src/main/java/org/apache/hadoop/yarn/server/api/impl/pb/client/AggregatorNodemanagerProtocolPBClientImpl.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-common/src/main/java/org/apache/hadoop/yarn/server/api/impl/pb/client/AggregatorNodemanagerProtocolPBClientImpl.java
deleted file mode 100644
index 6e777e7..0000000
--- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-common/src/main/java/org/apache/hadoop/yarn/server/api/impl/pb/client/AggregatorNodemanagerProtocolPBClientImpl.java
+++ /dev/null
@@ -1,94 +0,0 @@
-/**
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-package org.apache.hadoop.yarn.server.api.impl.pb.client;
-
-import java.io.Closeable;
-import java.io.IOException;
-import java.net.InetSocketAddress;
-
-import org.apache.hadoop.classification.InterfaceAudience.Private;
-import org.apache.hadoop.conf.Configuration;
-import org.apache.hadoop.ipc.ProtobufRpcEngine;
-import org.apache.hadoop.ipc.RPC;
-import org.apache.hadoop.net.NetUtils;
-import org.apache.hadoop.security.UserGroupInformation;
-import org.apache.hadoop.yarn.conf.YarnConfiguration;
-import org.apache.hadoop.yarn.exceptions.YarnException;
-import org.apache.hadoop.yarn.ipc.RPCUtil;
-import org.apache.hadoop.yarn.proto.YarnServerCommonServiceProtos.ReportNewAggregatorsInfoRequestProto;
-import org.apache.hadoop.yarn.server.api.AggregatorNodemanagerProtocol;
-import org.apache.hadoop.yarn.server.api.AggregatorNodemanagerProtocolPB;
-import org.apache.hadoop.yarn.server.api.protocolrecords.ReportNewAggregatorsInfoRequest;
-import org.apache.hadoop.yarn.server.api.protocolrecords.ReportNewAggregatorsInfoResponse;
-import org.apache.hadoop.yarn.server.api.protocolrecords.impl.pb.ReportNewAggregatorsInfoRequestPBImpl;
-import org.apache.hadoop.yarn.server.api.protocolrecords.impl.pb.ReportNewAggregatorsInfoResponsePBImpl;
-
-import com.google.protobuf.ServiceException;
-
-public class AggregatorNodemanagerProtocolPBClientImpl implements
- AggregatorNodemanagerProtocol, Closeable {
-
- // Not a documented config. Only used for tests internally
- static final String NM_COMMAND_TIMEOUT = YarnConfiguration.YARN_PREFIX
- + "rpc.nm-command-timeout";
-
- /**
- * Maximum of 1 minute timeout for a Node to react to the command
- */
- static final int DEFAULT_COMMAND_TIMEOUT = 60000;
-
- private AggregatorNodemanagerProtocolPB proxy;
-
- @Private
- public AggregatorNodemanagerProtocolPBClientImpl(long clientVersion,
- InetSocketAddress addr, Configuration conf) throws IOException {
- RPC.setProtocolEngine(conf, AggregatorNodemanagerProtocolPB.class,
- ProtobufRpcEngine.class);
- UserGroupInformation ugi = UserGroupInformation.getCurrentUser();
-
- int expireIntvl = conf.getInt(NM_COMMAND_TIMEOUT, DEFAULT_COMMAND_TIMEOUT);
- proxy =
- (AggregatorNodemanagerProtocolPB) RPC.getProxy(
- AggregatorNodemanagerProtocolPB.class,
- clientVersion, addr, ugi, conf,
- NetUtils.getDefaultSocketFactory(conf), expireIntvl);
- }
-
- @Override
- public ReportNewAggregatorsInfoResponse reportNewAggregatorInfo(
- ReportNewAggregatorsInfoRequest request) throws YarnException, IOException {
-
- ReportNewAggregatorsInfoRequestProto requestProto =
- ((ReportNewAggregatorsInfoRequestPBImpl) request).getProto();
- try {
- return new ReportNewAggregatorsInfoResponsePBImpl(
- proxy.reportNewAggregatorInfo(null, requestProto));
- } catch (ServiceException e) {
- RPCUtil.unwrapAndThrowException(e);
- return null;
- }
- }
-
- @Override
- public void close() {
- if (this.proxy != null) {
- RPC.stopProxy(this.proxy);
- }
- }
-
-}
http://git-wip-us.apache.org/repos/asf/hadoop/blob/dda84085/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-common/src/main/java/org/apache/hadoop/yarn/server/api/impl/pb/client/CollectorNodemanagerProtocolPBClientImpl.java
----------------------------------------------------------------------
diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-common/src/main/java/org/apache/hadoop/yarn/server/api/impl/pb/client/CollectorNodemanagerProtocolPBClientImpl.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-common/src/main/java/org/apache/hadoop/yarn/server/api/impl/pb/client/CollectorNodemanagerProtocolPBClientImpl.java
new file mode 100644
index 0000000..276a540
--- /dev/null
+++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-common/src/main/java/org/apache/hadoop/yarn/server/api/impl/pb/client/CollectorNodemanagerProtocolPBClientImpl.java
@@ -0,0 +1,94 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.hadoop.yarn.server.api.impl.pb.client;
+
+import java.io.Closeable;
+import java.io.IOException;
+import java.net.InetSocketAddress;
+
+import org.apache.hadoop.classification.InterfaceAudience.Private;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.ipc.ProtobufRpcEngine;
+import org.apache.hadoop.ipc.RPC;
+import org.apache.hadoop.net.NetUtils;
+import org.apache.hadoop.security.UserGroupInformation;
+import org.apache.hadoop.yarn.conf.YarnConfiguration;
+import org.apache.hadoop.yarn.exceptions.YarnException;
+import org.apache.hadoop.yarn.ipc.RPCUtil;
+import org.apache.hadoop.yarn.proto.YarnServerCommonServiceProtos.ReportNewCollectorInfoRequestProto;
+import org.apache.hadoop.yarn.server.api.CollectorNodemanagerProtocol;
+import org.apache.hadoop.yarn.server.api.CollectorNodemanagerProtocolPB;
+import org.apache.hadoop.yarn.server.api.protocolrecords.ReportNewCollectorInfoRequest;
+import org.apache.hadoop.yarn.server.api.protocolrecords.ReportNewCollectorInfoResponse;
+import org.apache.hadoop.yarn.server.api.protocolrecords.impl.pb.ReportNewCollectorInfoRequestPBImpl;
+import org.apache.hadoop.yarn.server.api.protocolrecords.impl.pb.ReportNewCollectorInfoResponsePBImpl;
+
+import com.google.protobuf.ServiceException;
+
+public class CollectorNodemanagerProtocolPBClientImpl implements
+ CollectorNodemanagerProtocol, Closeable {
+
+ // Not a documented config. Only used for tests internally
+ static final String NM_COMMAND_TIMEOUT = YarnConfiguration.YARN_PREFIX
+ + "rpc.nm-command-timeout";
+
+ /**
+ * Maximum of 1 minute timeout for a Node to react to the command
+ */
+ static final int DEFAULT_COMMAND_TIMEOUT = 60000;
+
+ private CollectorNodemanagerProtocolPB proxy;
+
+ @Private
+ public CollectorNodemanagerProtocolPBClientImpl(long clientVersion,
+ InetSocketAddress addr, Configuration conf) throws IOException {
+ RPC.setProtocolEngine(conf, CollectorNodemanagerProtocolPB.class,
+ ProtobufRpcEngine.class);
+ UserGroupInformation ugi = UserGroupInformation.getCurrentUser();
+
+ int expireIntvl = conf.getInt(NM_COMMAND_TIMEOUT, DEFAULT_COMMAND_TIMEOUT);
+ proxy =
+ (CollectorNodemanagerProtocolPB) RPC.getProxy(
+ CollectorNodemanagerProtocolPB.class,
+ clientVersion, addr, ugi, conf,
+ NetUtils.getDefaultSocketFactory(conf), expireIntvl);
+ }
+
+ @Override
+ public ReportNewCollectorInfoResponse reportNewCollectorInfo(
+ ReportNewCollectorInfoRequest request) throws YarnException, IOException {
+
+ ReportNewCollectorInfoRequestProto requestProto =
+ ((ReportNewCollectorInfoRequestPBImpl) request).getProto();
+ try {
+ return new ReportNewCollectorInfoResponsePBImpl(
+ proxy.reportNewCollectorInfo(null, requestProto));
+ } catch (ServiceException e) {
+ RPCUtil.unwrapAndThrowException(e);
+ return null;
+ }
+ }
+
+ @Override
+ public void close() {
+ if (this.proxy != null) {
+ RPC.stopProxy(this.proxy);
+ }
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/hadoop/blob/dda84085/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-common/src/main/java/org/apache/hadoop/yarn/server/api/impl/pb/service/AggregatorNodemanagerProtocolPBServiceImpl.java
----------------------------------------------------------------------
diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-common/src/main/java/org/apache/hadoop/yarn/server/api/impl/pb/service/AggregatorNodemanagerProtocolPBServiceImpl.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-common/src/main/java/org/apache/hadoop/yarn/server/api/impl/pb/service/AggregatorNodemanagerProtocolPBServiceImpl.java
deleted file mode 100644
index 87bce16..0000000
--- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-common/src/main/java/org/apache/hadoop/yarn/server/api/impl/pb/service/AggregatorNodemanagerProtocolPBServiceImpl.java
+++ /dev/null
@@ -1,61 +0,0 @@
-/**
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-package org.apache.hadoop.yarn.server.api.impl.pb.service;
-
-import java.io.IOException;
-
-import org.apache.hadoop.classification.InterfaceAudience.Private;
-import org.apache.hadoop.yarn.exceptions.YarnException;
-import org.apache.hadoop.yarn.proto.YarnServerCommonServiceProtos.ReportNewAggregatorsInfoRequestProto;
-import org.apache.hadoop.yarn.proto.YarnServerCommonServiceProtos.ReportNewAggregatorsInfoResponseProto;
-import org.apache.hadoop.yarn.server.api.AggregatorNodemanagerProtocol;
-import org.apache.hadoop.yarn.server.api.AggregatorNodemanagerProtocolPB;
-import org.apache.hadoop.yarn.server.api.protocolrecords.ReportNewAggregatorsInfoRequest;
-import org.apache.hadoop.yarn.server.api.protocolrecords.ReportNewAggregatorsInfoResponse;
-import org.apache.hadoop.yarn.server.api.protocolrecords.impl.pb.ReportNewAggregatorsInfoRequestPBImpl;
-import org.apache.hadoop.yarn.server.api.protocolrecords.impl.pb.ReportNewAggregatorsInfoResponsePBImpl;
-
-import com.google.protobuf.RpcController;
-import com.google.protobuf.ServiceException;
-
-public class AggregatorNodemanagerProtocolPBServiceImpl implements
- AggregatorNodemanagerProtocolPB {
-
- private AggregatorNodemanagerProtocol real;
-
- public AggregatorNodemanagerProtocolPBServiceImpl(AggregatorNodemanagerProtocol impl) {
- this.real = impl;
- }
-
- @Override
- public ReportNewAggregatorsInfoResponseProto reportNewAggregatorInfo(
- RpcController arg0, ReportNewAggregatorsInfoRequestProto proto)
- throws ServiceException {
- ReportNewAggregatorsInfoRequestPBImpl request =
- new ReportNewAggregatorsInfoRequestPBImpl(proto);
- try {
- ReportNewAggregatorsInfoResponse response = real.reportNewAggregatorInfo(request);
- return ((ReportNewAggregatorsInfoResponsePBImpl)response).getProto();
- } catch (YarnException e) {
- throw new ServiceException(e);
- } catch (IOException e) {
- throw new ServiceException(e);
- }
- }
-
-}
http://git-wip-us.apache.org/repos/asf/hadoop/blob/dda84085/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-common/src/main/java/org/apache/hadoop/yarn/server/api/impl/pb/service/CollectorNodemanagerProtocolPBServiceImpl.java
----------------------------------------------------------------------
diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-common/src/main/java/org/apache/hadoop/yarn/server/api/impl/pb/service/CollectorNodemanagerProtocolPBServiceImpl.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-common/src/main/java/org/apache/hadoop/yarn/server/api/impl/pb/service/CollectorNodemanagerProtocolPBServiceImpl.java
new file mode 100644
index 0000000..3f42732
--- /dev/null
+++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-common/src/main/java/org/apache/hadoop/yarn/server/api/impl/pb/service/CollectorNodemanagerProtocolPBServiceImpl.java
@@ -0,0 +1,59 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.hadoop.yarn.server.api.impl.pb.service;
+
+import java.io.IOException;
+
+import org.apache.hadoop.yarn.exceptions.YarnException;
+import org.apache.hadoop.yarn.proto.YarnServerCommonServiceProtos.ReportNewCollectorInfoRequestProto;
+import org.apache.hadoop.yarn.proto.YarnServerCommonServiceProtos.ReportNewCollectorInfoResponseProto;
+import org.apache.hadoop.yarn.server.api.CollectorNodemanagerProtocol;
+import org.apache.hadoop.yarn.server.api.CollectorNodemanagerProtocolPB;
+import org.apache.hadoop.yarn.server.api.protocolrecords.ReportNewCollectorInfoResponse;
+import org.apache.hadoop.yarn.server.api.protocolrecords.impl.pb.ReportNewCollectorInfoRequestPBImpl;
+import org.apache.hadoop.yarn.server.api.protocolrecords.impl.pb.ReportNewCollectorInfoResponsePBImpl;
+
+import com.google.protobuf.RpcController;
+import com.google.protobuf.ServiceException;
+
+public class CollectorNodemanagerProtocolPBServiceImpl implements
+ CollectorNodemanagerProtocolPB {
+
+ private CollectorNodemanagerProtocol real;
+
+ public CollectorNodemanagerProtocolPBServiceImpl(CollectorNodemanagerProtocol impl) {
+ this.real = impl;
+ }
+
+ @Override
+ public ReportNewCollectorInfoResponseProto reportNewCollectorInfo(
+ RpcController arg0, ReportNewCollectorInfoRequestProto proto)
+ throws ServiceException {
+ ReportNewCollectorInfoRequestPBImpl request =
+ new ReportNewCollectorInfoRequestPBImpl(proto);
+ try {
+ ReportNewCollectorInfoResponse response = real.reportNewCollectorInfo(request);
+ return ((ReportNewCollectorInfoResponsePBImpl)response).getProto();
+ } catch (YarnException e) {
+ throw new ServiceException(e);
+ } catch (IOException e) {
+ throw new ServiceException(e);
+ }
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/hadoop/blob/dda84085/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-common/src/main/java/org/apache/hadoop/yarn/server/api/protocolrecords/NodeHeartbeatRequest.java
----------------------------------------------------------------------
diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-common/src/main/java/org/apache/hadoop/yarn/server/api/protocolrecords/NodeHeartbeatRequest.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-common/src/main/java/org/apache/hadoop/yarn/server/api/protocolrecords/NodeHeartbeatRequest.java
index 0b020b7..2f080d2 100644
--- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-common/src/main/java/org/apache/hadoop/yarn/server/api/protocolrecords/NodeHeartbeatRequest.java
+++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-common/src/main/java/org/apache/hadoop/yarn/server/api/protocolrecords/NodeHeartbeatRequest.java
@@ -18,7 +18,6 @@
package org.apache.hadoop.yarn.server.api.protocolrecords;
-import java.util.List;
import java.util.Map;
import org.apache.hadoop.yarn.api.records.ApplicationId;
@@ -40,11 +39,11 @@ public abstract class NodeHeartbeatRequest {
.setLastKnownNMTokenMasterKey(lastKnownNMTokenMasterKey);
return nodeHeartbeatRequest;
}
-
+
public static NodeHeartbeatRequest newInstance(NodeStatus nodeStatus,
MasterKey lastKnownContainerTokenMasterKey,
MasterKey lastKnownNMTokenMasterKey,
- Map<ApplicationId, String> registeredAggregators) {
+ Map<ApplicationId, String> registeredCollectors) {
NodeHeartbeatRequest nodeHeartbeatRequest =
Records.newRecord(NodeHeartbeatRequest.class);
nodeHeartbeatRequest.setNodeStatus(nodeStatus);
@@ -52,7 +51,7 @@ public abstract class NodeHeartbeatRequest {
.setLastKnownContainerTokenMasterKey(lastKnownContainerTokenMasterKey);
nodeHeartbeatRequest
.setLastKnownNMTokenMasterKey(lastKnownNMTokenMasterKey);
- nodeHeartbeatRequest.setRegisteredAggregators(registeredAggregators);
+ nodeHeartbeatRequest.setRegisteredCollectors(registeredCollectors);
return nodeHeartbeatRequest;
}
@@ -64,8 +63,9 @@ public abstract class NodeHeartbeatRequest {
public abstract MasterKey getLastKnownNMTokenMasterKey();
public abstract void setLastKnownNMTokenMasterKey(MasterKey secretKey);
-
- // This tells RM registered aggregators' address info on this node
- public abstract Map<ApplicationId, String> getRegisteredAggregators();
- public abstract void setRegisteredAggregators(Map<ApplicationId, String> appAggregatorsMap);
+
+ // This tells RM registered collectors' address info on this node
+ public abstract Map<ApplicationId, String> getRegisteredCollectors();
+ public abstract void setRegisteredCollectors(Map<ApplicationId,
+ String> appCollectorsMap);
}
http://git-wip-us.apache.org/repos/asf/hadoop/blob/dda84085/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-common/src/main/java/org/apache/hadoop/yarn/server/api/protocolrecords/NodeHeartbeatResponse.java
----------------------------------------------------------------------
diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-common/src/main/java/org/apache/hadoop/yarn/server/api/protocolrecords/NodeHeartbeatResponse.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-common/src/main/java/org/apache/hadoop/yarn/server/api/protocolrecords/NodeHeartbeatResponse.java
index 262ca07..167a9cf 100644
--- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-common/src/main/java/org/apache/hadoop/yarn/server/api/protocolrecords/NodeHeartbeatResponse.java
+++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-common/src/main/java/org/apache/hadoop/yarn/server/api/protocolrecords/NodeHeartbeatResponse.java
@@ -35,10 +35,10 @@ public interface NodeHeartbeatResponse {
List<ContainerId> getContainersToBeRemovedFromNM();
List<ApplicationId> getApplicationsToCleanup();
-
- // This tells NM the aggregators' address info of related Apps
- Map<ApplicationId, String> getAppAggregatorsMap();
- void setAppAggregatorsMap(Map<ApplicationId, String> appAggregatorsMap);
+
+ // This tells NM the collectors' address info of related apps
+ Map<ApplicationId, String> getAppCollectorsMap();
+ void setAppCollectorsMap(Map<ApplicationId, String> appCollectorsMap);
void setResponseId(int responseId);
void setNodeAction(NodeAction action);
http://git-wip-us.apache.org/repos/asf/hadoop/blob/dda84085/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-common/src/main/java/org/apache/hadoop/yarn/server/api/protocolrecords/ReportNewAggregatorsInfoRequest.java
----------------------------------------------------------------------
diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-common/src/main/java/org/apache/hadoop/yarn/server/api/protocolrecords/ReportNewAggregatorsInfoRequest.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-common/src/main/java/org/apache/hadoop/yarn/server/api/protocolrecords/ReportNewAggregatorsInfoRequest.java
deleted file mode 100644
index ae538a2..0000000
--- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-common/src/main/java/org/apache/hadoop/yarn/server/api/protocolrecords/ReportNewAggregatorsInfoRequest.java
+++ /dev/null
@@ -1,53 +0,0 @@
-/**
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-package org.apache.hadoop.yarn.server.api.protocolrecords;
-
-import java.util.List;
-import java.util.Arrays;
-
-import org.apache.hadoop.classification.InterfaceAudience.Private;
-import org.apache.hadoop.yarn.api.records.ApplicationId;
-import org.apache.hadoop.yarn.server.api.records.AppAggregatorsMap;
-import org.apache.hadoop.yarn.util.Records;
-
-@Private
-public abstract class ReportNewAggregatorsInfoRequest {
-
- public static ReportNewAggregatorsInfoRequest newInstance(
- List<AppAggregatorsMap> appAggregatorsList) {
- ReportNewAggregatorsInfoRequest request =
- Records.newRecord(ReportNewAggregatorsInfoRequest.class);
- request.setAppAggregatorsList(appAggregatorsList);
- return request;
- }
-
- public static ReportNewAggregatorsInfoRequest newInstance(
- ApplicationId id, String aggregatorAddr) {
- ReportNewAggregatorsInfoRequest request =
- Records.newRecord(ReportNewAggregatorsInfoRequest.class);
- request.setAppAggregatorsList(
- Arrays.asList(AppAggregatorsMap.newInstance(id, aggregatorAddr)));
- return request;
- }
-
- public abstract List<AppAggregatorsMap> getAppAggregatorsList();
-
- public abstract void setAppAggregatorsList(
- List<AppAggregatorsMap> appAggregatorsList);
-
-}
http://git-wip-us.apache.org/repos/asf/hadoop/blob/dda84085/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-common/src/main/java/org/apache/hadoop/yarn/server/api/protocolrecords/ReportNewAggregatorsInfoResponse.java
----------------------------------------------------------------------
diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-common/src/main/java/org/apache/hadoop/yarn/server/api/protocolrecords/ReportNewAggregatorsInfoResponse.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-common/src/main/java/org/apache/hadoop/yarn/server/api/protocolrecords/ReportNewAggregatorsInfoResponse.java
deleted file mode 100644
index 3b847d6..0000000
--- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-common/src/main/java/org/apache/hadoop/yarn/server/api/protocolrecords/ReportNewAggregatorsInfoResponse.java
+++ /dev/null
@@ -1,32 +0,0 @@
-/**
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-package org.apache.hadoop.yarn.server.api.protocolrecords;
-
-import org.apache.hadoop.classification.InterfaceAudience.Private;
-import org.apache.hadoop.yarn.util.Records;
-
-public abstract class ReportNewAggregatorsInfoResponse {
-
- @Private
- public static ReportNewAggregatorsInfoResponse newInstance() {
- ReportNewAggregatorsInfoResponse response =
- Records.newRecord(ReportNewAggregatorsInfoResponse.class);
- return response;
- }
-
-}