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 jl...@apache.org on 2015/10/29 19:06:20 UTC
hadoop git commit: MAPREDUCE-6515. Update Application priority in AM
side from AM-RM heartbeat. Contributed by Sunil G (cherry picked from commit
cf953b6258b036fc482456b4591cfb98054f48f2)
Repository: hadoop
Updated Branches:
refs/heads/branch-2 c75d8b164 -> 145058003
MAPREDUCE-6515. Update Application priority in AM side from AM-RM heartbeat. Contributed by Sunil G
(cherry picked from commit cf953b6258b036fc482456b4591cfb98054f48f2)
Project: http://git-wip-us.apache.org/repos/asf/hadoop/repo
Commit: http://git-wip-us.apache.org/repos/asf/hadoop/commit/14505800
Tree: http://git-wip-us.apache.org/repos/asf/hadoop/tree/14505800
Diff: http://git-wip-us.apache.org/repos/asf/hadoop/diff/14505800
Branch: refs/heads/branch-2
Commit: 145058003aec7f99a24d474d765133bb3b503785
Parents: c75d8b1
Author: Jason Lowe <jl...@apache.org>
Authored: Thu Oct 29 18:05:01 2015 +0000
Committer: Jason Lowe <jl...@apache.org>
Committed: Thu Oct 29 18:05:56 2015 +0000
----------------------------------------------------------------------
hadoop-mapreduce-project/CHANGES.txt | 3 ++
.../apache/hadoop/mapreduce/v2/app/job/Job.java | 2 +
.../mapreduce/v2/app/job/impl/JobImpl.java | 13 +++++-
.../v2/app/rm/RMContainerAllocator.java | 9 ++++
.../hadoop/mapreduce/v2/app/MockJobs.java | 6 +++
.../mapreduce/v2/app/TestRuntimeEstimators.java | 6 +++
.../mapreduce/v2/app/job/impl/TestJobImpl.java | 35 ++++++++++++++++
.../v2/app/rm/TestRMContainerAllocator.java | 2 +
.../apache/hadoop/mapreduce/TypeConverter.java | 34 +++++++++++----
.../mapreduce/v2/api/records/JobReport.java | 4 ++
.../v2/api/records/impl/pb/JobReportPBImpl.java | 44 ++++++++++++++++++--
.../mapreduce/v2/util/MRBuilderUtils.java | 15 ++++++-
.../src/main/proto/mr_protos.proto | 1 +
.../hadoop/mapreduce/TestTypeConverter.java | 8 +++-
.../org/apache/hadoop/mapred/JobPriority.java | 5 ++-
.../apache/hadoop/mapreduce/JobPriority.java | 4 +-
.../hadoop/mapreduce/v2/hs/CompletedJob.java | 7 ++++
.../hadoop/mapreduce/v2/hs/PartialJob.java | 7 ++++
.../v2/hs/webapp/TestHsWebServicesAcls.java | 5 +++
19 files changed, 191 insertions(+), 19 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/hadoop/blob/14505800/hadoop-mapreduce-project/CHANGES.txt
----------------------------------------------------------------------
diff --git a/hadoop-mapreduce-project/CHANGES.txt b/hadoop-mapreduce-project/CHANGES.txt
index b48b9cd..b52fa13 100644
--- a/hadoop-mapreduce-project/CHANGES.txt
+++ b/hadoop-mapreduce-project/CHANGES.txt
@@ -339,6 +339,9 @@ Release 2.8.0 - UNRELEASED
MAPREDUCE-6508. TestNetworkedJob fails consistently due to delegation
token changes on RM. (Akira AJISAKA via junping_du)
+ MAPREDUCE-6515. Update Application priority in AM side from AM-RM heartbeat
+ (Sunil G via jlowe)
+
Release 2.7.2 - UNRELEASED
INCOMPATIBLE CHANGES
http://git-wip-us.apache.org/repos/asf/hadoop/blob/14505800/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-app/src/main/java/org/apache/hadoop/mapreduce/v2/app/job/Job.java
----------------------------------------------------------------------
diff --git a/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-app/src/main/java/org/apache/hadoop/mapreduce/v2/app/job/Job.java b/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-app/src/main/java/org/apache/hadoop/mapreduce/v2/app/job/Job.java
index a40e5d2..7738810 100644
--- a/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-app/src/main/java/org/apache/hadoop/mapreduce/v2/app/job/Job.java
+++ b/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-app/src/main/java/org/apache/hadoop/mapreduce/v2/app/job/Job.java
@@ -36,6 +36,7 @@ import org.apache.hadoop.mapreduce.v2.api.records.TaskId;
import org.apache.hadoop.mapreduce.v2.api.records.TaskType;
import org.apache.hadoop.security.UserGroupInformation;
import org.apache.hadoop.security.authorize.AccessControlList;
+import org.apache.hadoop.yarn.api.records.Priority;
/**
@@ -100,4 +101,5 @@ public interface Job {
boolean checkAccess(UserGroupInformation callerUGI, JobACL jobOperation);
public void setQueueName(String queueName);
+ public void setJobPriority(Priority priority);
}
http://git-wip-us.apache.org/repos/asf/hadoop/blob/14505800/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-app/src/main/java/org/apache/hadoop/mapreduce/v2/app/job/impl/JobImpl.java
----------------------------------------------------------------------
diff --git a/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-app/src/main/java/org/apache/hadoop/mapreduce/v2/app/job/impl/JobImpl.java b/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-app/src/main/java/org/apache/hadoop/mapreduce/v2/app/job/impl/JobImpl.java
index ab94960..b11745b 100644
--- a/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-app/src/main/java/org/apache/hadoop/mapreduce/v2/app/job/impl/JobImpl.java
+++ b/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-app/src/main/java/org/apache/hadoop/mapreduce/v2/app/job/impl/JobImpl.java
@@ -120,6 +120,7 @@ import org.apache.hadoop.yarn.api.records.ApplicationAttemptId;
import org.apache.hadoop.yarn.api.records.NodeId;
import org.apache.hadoop.yarn.api.records.NodeReport;
import org.apache.hadoop.yarn.api.records.NodeState;
+import org.apache.hadoop.yarn.api.records.Priority;
import org.apache.hadoop.yarn.event.EventHandler;
import org.apache.hadoop.yarn.exceptions.YarnRuntimeException;
import org.apache.hadoop.yarn.state.InvalidStateTransitionException;
@@ -653,6 +654,8 @@ public class JobImpl implements org.apache.hadoop.mapreduce.v2.app.job.Job,
private JobState lastNonFinalState = JobState.NEW;
+ private volatile Priority jobPriority = Priority.newInstance(0);
+
public JobImpl(JobId jobId, ApplicationAttemptId applicationAttemptId,
Configuration conf, EventHandler eventHandler,
TaskAttemptListener taskAttemptListener,
@@ -878,7 +881,8 @@ public class JobImpl implements org.apache.hadoop.mapreduce.v2.app.job.Job,
reporterUserName,
state, appSubmitTime, startTime, finishTime, setupProgress,
this.mapProgress, this.reduceProgress,
- cleanupProgress, jobFile, amInfos, isUber, diagsb.toString());
+ cleanupProgress, jobFile, amInfos, isUber, diagsb.toString(),
+ jobPriority);
return report;
} finally {
readLock.unlock();
@@ -2164,7 +2168,7 @@ public class JobImpl implements org.apache.hadoop.mapreduce.v2.app.job.Job,
}
}
}
-
+
private static class InternalTerminationTransition implements
SingleArcTransition<JobImpl, JobEvent> {
JobStateInternal terminationState = null;
@@ -2217,4 +2221,9 @@ public class JobImpl implements org.apache.hadoop.mapreduce.v2.app.job.Job,
public int getMaxFetchFailuresNotifications() {
return maxFetchFailuresNotifications;
}
+
+ @Override
+ public void setJobPriority(Priority priority) {
+ this.jobPriority = priority;
+ }
}
http://git-wip-us.apache.org/repos/asf/hadoop/blob/14505800/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-app/src/main/java/org/apache/hadoop/mapreduce/v2/app/rm/RMContainerAllocator.java
----------------------------------------------------------------------
diff --git a/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-app/src/main/java/org/apache/hadoop/mapreduce/v2/app/rm/RMContainerAllocator.java b/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-app/src/main/java/org/apache/hadoop/mapreduce/v2/app/rm/RMContainerAllocator.java
index fadab46..3713cfb 100644
--- a/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-app/src/main/java/org/apache/hadoop/mapreduce/v2/app/rm/RMContainerAllocator.java
+++ b/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-app/src/main/java/org/apache/hadoop/mapreduce/v2/app/rm/RMContainerAllocator.java
@@ -779,6 +779,7 @@ public class RMContainerAllocator extends RMContainerRequestor
computeIgnoreBlacklisting();
handleUpdatedNodes(response);
+ handleJobPriorityChange(response);
for (ContainerStatus cont : finishedContainers) {
LOG.info("Received completed container " + cont.getContainerId());
@@ -901,6 +902,14 @@ public class RMContainerAllocator extends RMContainerRequestor
}
}
+ private void handleJobPriorityChange(AllocateResponse response) {
+ Priority priorityFromResponse = Priority.newInstance(response
+ .getApplicationPriority().getPriority());
+
+ // Update the job priority to Job directly.
+ getJob().setJobPriority(priorityFromResponse);
+ }
+
@Private
public Resource getResourceLimit() {
Resource headRoom = getAvailableResources();
http://git-wip-us.apache.org/repos/asf/hadoop/blob/14505800/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-app/src/test/java/org/apache/hadoop/mapreduce/v2/app/MockJobs.java
----------------------------------------------------------------------
diff --git a/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-app/src/test/java/org/apache/hadoop/mapreduce/v2/app/MockJobs.java b/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-app/src/test/java/org/apache/hadoop/mapreduce/v2/app/MockJobs.java
index fd9c094..ccacf1c 100644
--- a/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-app/src/test/java/org/apache/hadoop/mapreduce/v2/app/MockJobs.java
+++ b/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-app/src/test/java/org/apache/hadoop/mapreduce/v2/app/MockJobs.java
@@ -66,6 +66,7 @@ import org.apache.hadoop.yarn.api.records.ApplicationAttemptId;
import org.apache.hadoop.yarn.api.records.ApplicationId;
import org.apache.hadoop.yarn.api.records.ContainerId;
import org.apache.hadoop.yarn.api.records.NodeId;
+import org.apache.hadoop.yarn.api.records.Priority;
import org.apache.hadoop.yarn.util.Records;
import com.google.common.collect.Iterators;
@@ -634,6 +635,11 @@ public class MockJobs extends MockApps {
public void setQueueName(String queueName) {
// do nothing
}
+
+ @Override
+ public void setJobPriority(Priority priority) {
+ // do nothing
+ }
};
}
http://git-wip-us.apache.org/repos/asf/hadoop/blob/14505800/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-app/src/test/java/org/apache/hadoop/mapreduce/v2/app/TestRuntimeEstimators.java
----------------------------------------------------------------------
diff --git a/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-app/src/test/java/org/apache/hadoop/mapreduce/v2/app/TestRuntimeEstimators.java b/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-app/src/test/java/org/apache/hadoop/mapreduce/v2/app/TestRuntimeEstimators.java
index 475cd1f..0b7d1b1 100644
--- a/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-app/src/test/java/org/apache/hadoop/mapreduce/v2/app/TestRuntimeEstimators.java
+++ b/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-app/src/test/java/org/apache/hadoop/mapreduce/v2/app/TestRuntimeEstimators.java
@@ -69,6 +69,7 @@ import org.apache.hadoop.yarn.api.records.ApplicationAttemptId;
import org.apache.hadoop.yarn.api.records.ApplicationId;
import org.apache.hadoop.yarn.api.records.ContainerId;
import org.apache.hadoop.yarn.api.records.NodeId;
+import org.apache.hadoop.yarn.api.records.Priority;
import org.apache.hadoop.yarn.event.AsyncDispatcher;
import org.apache.hadoop.yarn.event.EventHandler;
import org.apache.hadoop.yarn.factories.RecordFactory;
@@ -526,6 +527,11 @@ public class TestRuntimeEstimators {
public void setQueueName(String queueName) {
// do nothing
}
+
+ @Override
+ public void setJobPriority(Priority priority) {
+ // do nothing
+ }
}
/*
http://git-wip-us.apache.org/repos/asf/hadoop/blob/14505800/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-app/src/test/java/org/apache/hadoop/mapreduce/v2/app/job/impl/TestJobImpl.java
----------------------------------------------------------------------
diff --git a/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-app/src/test/java/org/apache/hadoop/mapreduce/v2/app/job/impl/TestJobImpl.java b/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-app/src/test/java/org/apache/hadoop/mapreduce/v2/app/job/impl/TestJobImpl.java
index 2af4380..9e18920 100644
--- a/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-app/src/test/java/org/apache/hadoop/mapreduce/v2/app/job/impl/TestJobImpl.java
+++ b/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-app/src/test/java/org/apache/hadoop/mapreduce/v2/app/job/impl/TestJobImpl.java
@@ -72,6 +72,7 @@ import org.apache.hadoop.mapreduce.v2.app.job.event.JobDiagnosticsUpdateEvent;
import org.apache.hadoop.mapreduce.v2.app.job.event.JobEvent;
import org.apache.hadoop.mapreduce.v2.app.job.event.JobEventType;
import org.apache.hadoop.mapreduce.v2.app.job.event.JobFinishEvent;
+import org.apache.hadoop.mapreduce.v2.app.job.event.JobSetupCompletedEvent;
import org.apache.hadoop.mapreduce.v2.app.job.event.JobStartEvent;
import org.apache.hadoop.mapreduce.v2.app.job.event.JobTaskAttemptCompletedEvent;
import org.apache.hadoop.mapreduce.v2.app.job.event.JobTaskEvent;
@@ -92,6 +93,7 @@ import org.apache.hadoop.yarn.api.records.ApplicationId;
import org.apache.hadoop.yarn.api.records.NodeId;
import org.apache.hadoop.yarn.api.records.NodeReport;
import org.apache.hadoop.yarn.api.records.NodeState;
+import org.apache.hadoop.yarn.api.records.Priority;
import org.apache.hadoop.yarn.event.AsyncDispatcher;
import org.apache.hadoop.yarn.event.Dispatcher;
import org.apache.hadoop.yarn.event.DrainDispatcher;
@@ -889,6 +891,39 @@ public class TestJobImpl {
job.getDiagnostics().toString().contains(EXCEPTIONMSG));
}
+ @Test
+ public void testJobPriorityUpdate() throws Exception {
+ Configuration conf = new Configuration();
+ AsyncDispatcher dispatcher = new AsyncDispatcher();
+ Priority submittedPriority = Priority.newInstance(5);
+
+ AppContext mockContext = mock(AppContext.class);
+ when(mockContext.hasSuccessfullyUnregistered()).thenReturn(false);
+ JobImpl job = createStubbedJob(conf, dispatcher, 2, mockContext);
+
+ JobId jobId = job.getID();
+ job.handle(new JobEvent(jobId, JobEventType.JOB_INIT));
+ assertJobState(job, JobStateInternal.INITED);
+ job.handle(new JobStartEvent(jobId));
+ assertJobState(job, JobStateInternal.SETUP);
+ // Update priority of job to 5, and it will be updated
+ job.setJobPriority(submittedPriority);
+ Assert.assertEquals(submittedPriority, job.getReport().getJobPriority());
+
+ job.handle(new JobSetupCompletedEvent(jobId));
+ assertJobState(job, JobStateInternal.RUNNING);
+
+ // Update priority of job to 8, and see whether its updated
+ Priority updatedPriority = Priority.newInstance(5);
+ job.setJobPriority(updatedPriority);
+ assertJobState(job, JobStateInternal.RUNNING);
+ Priority jobPriority = job.getReport().getJobPriority();
+ Assert.assertNotNull(jobPriority);
+
+ // Verify whether changed priority is same as what is set in Job.
+ Assert.assertEquals(updatedPriority, jobPriority);
+ }
+
private static CommitterEventHandler createCommitterEventHandler(
Dispatcher dispatcher, OutputCommitter committer) {
final SystemClock clock = new SystemClock();
http://git-wip-us.apache.org/repos/asf/hadoop/blob/14505800/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-app/src/test/java/org/apache/hadoop/mapreduce/v2/app/rm/TestRMContainerAllocator.java
----------------------------------------------------------------------
diff --git a/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-app/src/test/java/org/apache/hadoop/mapreduce/v2/app/rm/TestRMContainerAllocator.java b/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-app/src/test/java/org/apache/hadoop/mapreduce/v2/app/rm/TestRMContainerAllocator.java
index ab1e629..b5e33cd 100644
--- a/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-app/src/test/java/org/apache/hadoop/mapreduce/v2/app/rm/TestRMContainerAllocator.java
+++ b/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-app/src/test/java/org/apache/hadoop/mapreduce/v2/app/rm/TestRMContainerAllocator.java
@@ -2943,6 +2943,8 @@ public class TestRMContainerAllocator {
Collections.<NodeReport>emptyList(),
Resource.newInstance(512000, 1024), null, 10, null,
Collections.<NMToken>emptyList());
+ // RM will always ensure that a default priority is sent to AM
+ response.setApplicationPriority(Priority.newInstance(0));
containersToComplete.clear();
containersToAllocate.clear();
return response;
http://git-wip-us.apache.org/repos/asf/hadoop/blob/14505800/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-common/src/main/java/org/apache/hadoop/mapreduce/TypeConverter.java
----------------------------------------------------------------------
diff --git a/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-common/src/main/java/org/apache/hadoop/mapreduce/TypeConverter.java b/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-common/src/main/java/org/apache/hadoop/mapreduce/TypeConverter.java
index 5b8d3a7..88f61b0 100644
--- a/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-common/src/main/java/org/apache/hadoop/mapreduce/TypeConverter.java
+++ b/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-common/src/main/java/org/apache/hadoop/mapreduce/TypeConverter.java
@@ -296,21 +296,41 @@ public class TypeConverter {
}
return yCntrs;
}
-
+
public static JobStatus fromYarn(JobReport jobreport, String trackingUrl) {
- JobPriority jobPriority = JobPriority.NORMAL;
+ JobPriority jobPriority = (jobreport.getJobPriority() == null)
+ ? JobPriority.DEFAULT
+ : fromYarnPriority(jobreport.getJobPriority().getPriority());
JobStatus jobStatus = new org.apache.hadoop.mapred.JobStatus(
- fromYarn(jobreport.getJobId()), jobreport.getSetupProgress(), jobreport
- .getMapProgress(), jobreport.getReduceProgress(), jobreport
- .getCleanupProgress(), fromYarn(jobreport.getJobState()),
- jobPriority, jobreport.getUser(), jobreport.getJobName(), jobreport
- .getJobFile(), trackingUrl, jobreport.isUber());
+ fromYarn(jobreport.getJobId()), jobreport.getSetupProgress(),
+ jobreport.getMapProgress(), jobreport.getReduceProgress(),
+ jobreport.getCleanupProgress(), fromYarn(jobreport.getJobState()),
+ jobPriority, jobreport.getUser(), jobreport.getJobName(),
+ jobreport.getJobFile(), trackingUrl, jobreport.isUber());
jobStatus.setStartTime(jobreport.getStartTime());
jobStatus.setFinishTime(jobreport.getFinishTime());
jobStatus.setFailureInfo(jobreport.getDiagnostics());
return jobStatus;
}
+ private static JobPriority fromYarnPriority(int priority) {
+ switch (priority) {
+ case 5 :
+ return JobPriority.VERY_HIGH;
+ case 4 :
+ return JobPriority.HIGH;
+ case 3 :
+ return JobPriority.NORMAL;
+ case 2 :
+ return JobPriority.LOW;
+ case 1 :
+ return JobPriority.VERY_LOW;
+ case 0 :
+ return JobPriority.DEFAULT;
+ }
+ return JobPriority.UNDEFINED_PRIORITY;
+ }
+
public static org.apache.hadoop.mapreduce.QueueState fromYarn(
QueueState state) {
org.apache.hadoop.mapreduce.QueueState qState =
http://git-wip-us.apache.org/repos/asf/hadoop/blob/14505800/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-common/src/main/java/org/apache/hadoop/mapreduce/v2/api/records/JobReport.java
----------------------------------------------------------------------
diff --git a/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-common/src/main/java/org/apache/hadoop/mapreduce/v2/api/records/JobReport.java b/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-common/src/main/java/org/apache/hadoop/mapreduce/v2/api/records/JobReport.java
index b2f2cc1..38dfcae 100644
--- a/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-common/src/main/java/org/apache/hadoop/mapreduce/v2/api/records/JobReport.java
+++ b/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-common/src/main/java/org/apache/hadoop/mapreduce/v2/api/records/JobReport.java
@@ -20,6 +20,8 @@ package org.apache.hadoop.mapreduce.v2.api.records;
import java.util.List;
+import org.apache.hadoop.yarn.api.records.Priority;
+
public interface JobReport {
public abstract JobId getJobId();
public abstract JobState getJobState();
@@ -37,6 +39,7 @@ public interface JobReport {
public abstract String getJobFile();
public abstract List<AMInfo> getAMInfos();
public abstract boolean isUber();
+ public abstract Priority getJobPriority();
public abstract void setJobId(JobId jobId);
public abstract void setJobState(JobState jobState);
@@ -54,4 +57,5 @@ public interface JobReport {
public abstract void setJobFile(String jobFile);
public abstract void setAMInfos(List<AMInfo> amInfos);
public abstract void setIsUber(boolean isUber);
+ public abstract void setJobPriority(Priority priority);
}
http://git-wip-us.apache.org/repos/asf/hadoop/blob/14505800/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-common/src/main/java/org/apache/hadoop/mapreduce/v2/api/records/impl/pb/JobReportPBImpl.java
----------------------------------------------------------------------
diff --git a/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-common/src/main/java/org/apache/hadoop/mapreduce/v2/api/records/impl/pb/JobReportPBImpl.java b/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-common/src/main/java/org/apache/hadoop/mapreduce/v2/api/records/impl/pb/JobReportPBImpl.java
index 5c90942..f4cb0a6 100644
--- a/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-common/src/main/java/org/apache/hadoop/mapreduce/v2/api/records/impl/pb/JobReportPBImpl.java
+++ b/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-common/src/main/java/org/apache/hadoop/mapreduce/v2/api/records/impl/pb/JobReportPBImpl.java
@@ -32,7 +32,10 @@ import org.apache.hadoop.mapreduce.v2.proto.MRProtos.JobReportProto;
import org.apache.hadoop.mapreduce.v2.proto.MRProtos.JobReportProtoOrBuilder;
import org.apache.hadoop.mapreduce.v2.proto.MRProtos.JobStateProto;
import org.apache.hadoop.mapreduce.v2.util.MRProtoUtils;
+import org.apache.hadoop.yarn.api.records.Priority;
+import org.apache.hadoop.yarn.api.records.impl.pb.PriorityPBImpl;
import org.apache.hadoop.yarn.api.records.impl.pb.ProtoBase;
+import org.apache.hadoop.yarn.proto.YarnProtos.PriorityProto;
@@ -41,11 +44,11 @@ public class JobReportPBImpl extends ProtoBase<JobReportProto> implements
JobReportProto proto = JobReportProto.getDefaultInstance();
JobReportProto.Builder builder = null;
boolean viaProto = false;
-
+
private JobId jobId = null;
private List<AMInfo> amInfos = null;
-
-
+ private Priority jobPriority = null;
+
public JobReportPBImpl() {
builder = JobReportProto.newBuilder();
}
@@ -69,6 +72,9 @@ public class JobReportPBImpl extends ProtoBase<JobReportProto> implements
if (this.amInfos != null) {
addAMInfosToProto();
}
+ if (this.jobPriority != null) {
+ builder.setJobPriority(convertToProtoFormat(this.jobPriority));
+ }
}
private synchronized void mergeLocalToProto() {
@@ -333,6 +339,14 @@ public class JobReportPBImpl extends ProtoBase<JobReportProto> implements
return MRProtoUtils.convertFromProtoFormat(e);
}
+ private PriorityPBImpl convertFromProtoFormat(PriorityProto p) {
+ return new PriorityPBImpl(p);
+ }
+
+ private PriorityProto convertToProtoFormat(Priority t) {
+ return ((PriorityPBImpl)t).getProto();
+ }
+
@Override
public synchronized boolean isUber() {
JobReportProtoOrBuilder p = viaProto ? proto : builder;
@@ -344,4 +358,26 @@ public class JobReportPBImpl extends ProtoBase<JobReportProto> implements
maybeInitBuilder();
builder.setIsUber(isUber);
}
-}
+
+ @Override
+ public synchronized Priority getJobPriority() {
+ JobReportProtoOrBuilder p = viaProto ? proto : builder;
+ if (this.jobPriority != null) {
+ return this.jobPriority;
+ }
+ if (!p.hasJobPriority()) {
+ return null;
+ }
+ this.jobPriority = convertFromProtoFormat(p.getJobPriority());
+ return this.jobPriority;
+ }
+
+ @Override
+ public synchronized void setJobPriority(Priority priority) {
+ maybeInitBuilder();
+ if (priority == null) {
+ builder.clearJobPriority();
+ }
+ this.jobPriority = priority;
+ }
+}
http://git-wip-us.apache.org/repos/asf/hadoop/blob/14505800/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-common/src/main/java/org/apache/hadoop/mapreduce/v2/util/MRBuilderUtils.java
----------------------------------------------------------------------
diff --git a/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-common/src/main/java/org/apache/hadoop/mapreduce/v2/util/MRBuilderUtils.java b/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-common/src/main/java/org/apache/hadoop/mapreduce/v2/util/MRBuilderUtils.java
index 95e0083..893f76a 100644
--- a/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-common/src/main/java/org/apache/hadoop/mapreduce/v2/util/MRBuilderUtils.java
+++ b/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-common/src/main/java/org/apache/hadoop/mapreduce/v2/util/MRBuilderUtils.java
@@ -30,6 +30,7 @@ import org.apache.hadoop.mapreduce.v2.api.records.TaskType;
import org.apache.hadoop.yarn.api.records.ApplicationAttemptId;
import org.apache.hadoop.yarn.api.records.ApplicationId;
import org.apache.hadoop.yarn.api.records.ContainerId;
+import org.apache.hadoop.yarn.api.records.Priority;
import org.apache.hadoop.yarn.util.Records;
public class MRBuilderUtils {
@@ -63,10 +64,21 @@ public class MRBuilderUtils {
}
public static JobReport newJobReport(JobId jobId, String jobName,
+ String userName, JobState state, long submitTime, long startTime,
+ long finishTime, float setupProgress, float mapProgress,
+ float reduceProgress, float cleanupProgress, String jobFile,
+ List<AMInfo> amInfos, boolean isUber, String diagnostics) {
+ return newJobReport(jobId, jobName, userName, state, submitTime, startTime,
+ finishTime, setupProgress, mapProgress, reduceProgress,
+ cleanupProgress, jobFile, amInfos, isUber, diagnostics,
+ Priority.newInstance(0));
+ }
+
+ public static JobReport newJobReport(JobId jobId, String jobName,
String userName, JobState state, long submitTime, long startTime, long finishTime,
float setupProgress, float mapProgress, float reduceProgress,
float cleanupProgress, String jobFile, List<AMInfo> amInfos,
- boolean isUber, String diagnostics) {
+ boolean isUber, String diagnostics, Priority priority) {
JobReport report = Records.newRecord(JobReport.class);
report.setJobId(jobId);
report.setJobName(jobName);
@@ -83,6 +95,7 @@ public class MRBuilderUtils {
report.setAMInfos(amInfos);
report.setIsUber(isUber);
report.setDiagnostics(diagnostics);
+ report.setJobPriority(priority);
return report;
}
http://git-wip-us.apache.org/repos/asf/hadoop/blob/14505800/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-common/src/main/proto/mr_protos.proto
----------------------------------------------------------------------
diff --git a/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-common/src/main/proto/mr_protos.proto b/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-common/src/main/proto/mr_protos.proto
index b74eef6..5a4bac1 100644
--- a/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-common/src/main/proto/mr_protos.proto
+++ b/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-common/src/main/proto/mr_protos.proto
@@ -146,6 +146,7 @@ message JobReportProto {
repeated AMInfoProto am_infos = 14;
optional int64 submit_time = 15;
optional bool is_uber = 16 [default = false];
+ optional hadoop.yarn.PriorityProto jobPriority = 17;
}
message AMInfoProto {
http://git-wip-us.apache.org/repos/asf/hadoop/blob/14505800/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-common/src/test/java/org/apache/hadoop/mapreduce/TestTypeConverter.java
----------------------------------------------------------------------
diff --git a/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-common/src/test/java/org/apache/hadoop/mapreduce/TestTypeConverter.java b/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-common/src/test/java/org/apache/hadoop/mapreduce/TestTypeConverter.java
index e36efec..60ce170 100644
--- a/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-common/src/test/java/org/apache/hadoop/mapreduce/TestTypeConverter.java
+++ b/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-common/src/test/java/org/apache/hadoop/mapreduce/TestTypeConverter.java
@@ -18,6 +18,7 @@
package org.apache.hadoop.mapreduce;
import org.apache.hadoop.util.StringUtils;
+
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.when;
@@ -35,6 +36,7 @@ import org.apache.hadoop.yarn.api.records.ApplicationId;
import org.apache.hadoop.yarn.api.records.ApplicationReport;
import org.apache.hadoop.yarn.api.records.ApplicationResourceUsageReport;
import org.apache.hadoop.yarn.api.records.FinalApplicationStatus;
+import org.apache.hadoop.yarn.api.records.Priority;
import org.apache.hadoop.yarn.api.records.QueueState;
import org.apache.hadoop.yarn.api.records.Resource;
import org.apache.hadoop.yarn.api.records.YarnApplicationState;
@@ -201,10 +203,12 @@ public class TestTypeConverter {
jobReport.setJobState(state);
jobReport.setStartTime(jobStartTime);
jobReport.setFinishTime(jobFinishTime);
- jobReport.setUser("TestTypeConverter-user");
+ jobReport.setUser("TestTypeConverter-user");
+ jobReport.setJobPriority(Priority.newInstance(0));
JobStatus jobStatus = TypeConverter.fromYarn(jobReport, "dummy-jobfile");
Assert.assertEquals(jobStartTime, jobStatus.getStartTime());
Assert.assertEquals(jobFinishTime, jobStatus.getFinishTime());
Assert.assertEquals(state.toString(), jobStatus.getState().toString());
- }
+ Assert.assertEquals(JobPriority.DEFAULT, jobStatus.getPriority());
+ }
}
http://git-wip-us.apache.org/repos/asf/hadoop/blob/14505800/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-core/src/main/java/org/apache/hadoop/mapred/JobPriority.java
----------------------------------------------------------------------
diff --git a/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-core/src/main/java/org/apache/hadoop/mapred/JobPriority.java b/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-core/src/main/java/org/apache/hadoop/mapred/JobPriority.java
index 376d8a4..b76d46d 100644
--- a/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-core/src/main/java/org/apache/hadoop/mapred/JobPriority.java
+++ b/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-core/src/main/java/org/apache/hadoop/mapred/JobPriority.java
@@ -31,6 +31,7 @@ public enum JobPriority {
HIGH,
NORMAL,
LOW,
- VERY_LOW;
-
+ VERY_LOW,
+ DEFAULT,
+ UNDEFINED_PRIORITY;
}
http://git-wip-us.apache.org/repos/asf/hadoop/blob/14505800/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-core/src/main/java/org/apache/hadoop/mapreduce/JobPriority.java
----------------------------------------------------------------------
diff --git a/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-core/src/main/java/org/apache/hadoop/mapreduce/JobPriority.java b/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-core/src/main/java/org/apache/hadoop/mapreduce/JobPriority.java
index b0d232a..7178568 100644
--- a/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-core/src/main/java/org/apache/hadoop/mapreduce/JobPriority.java
+++ b/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-core/src/main/java/org/apache/hadoop/mapreduce/JobPriority.java
@@ -31,5 +31,7 @@ public enum JobPriority {
HIGH,
NORMAL,
LOW,
- VERY_LOW;
+ VERY_LOW,
+ DEFAULT,
+ UNDEFINED_PRIORITY;
}
http://git-wip-us.apache.org/repos/asf/hadoop/blob/14505800/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-hs/src/main/java/org/apache/hadoop/mapreduce/v2/hs/CompletedJob.java
----------------------------------------------------------------------
diff --git a/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-hs/src/main/java/org/apache/hadoop/mapreduce/v2/hs/CompletedJob.java b/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-hs/src/main/java/org/apache/hadoop/mapreduce/v2/hs/CompletedJob.java
index 0f1f391..3c12bdf 100644
--- a/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-hs/src/main/java/org/apache/hadoop/mapreduce/v2/hs/CompletedJob.java
+++ b/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-hs/src/main/java/org/apache/hadoop/mapreduce/v2/hs/CompletedJob.java
@@ -64,6 +64,7 @@ import org.apache.hadoop.mapreduce.v2.util.MRBuilderUtils;
import org.apache.hadoop.mapreduce.v2.util.MRWebAppUtil;
import org.apache.hadoop.security.UserGroupInformation;
import org.apache.hadoop.security.authorize.AccessControlList;
+import org.apache.hadoop.yarn.api.records.Priority;
import org.apache.hadoop.yarn.exceptions.YarnRuntimeException;
import org.apache.hadoop.yarn.util.Records;
@@ -481,4 +482,10 @@ public class CompletedJob implements org.apache.hadoop.mapreduce.v2.app.job.Job
public void setQueueName(String queueName) {
throw new UnsupportedOperationException("Can't set job's queue name in history");
}
+
+ @Override
+ public void setJobPriority(Priority priority) {
+ throw new UnsupportedOperationException(
+ "Can't set job's priority in history");
+ }
}
http://git-wip-us.apache.org/repos/asf/hadoop/blob/14505800/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-hs/src/main/java/org/apache/hadoop/mapreduce/v2/hs/PartialJob.java
----------------------------------------------------------------------
diff --git a/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-hs/src/main/java/org/apache/hadoop/mapreduce/v2/hs/PartialJob.java b/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-hs/src/main/java/org/apache/hadoop/mapreduce/v2/hs/PartialJob.java
index 0725f46..b3b181c 100644
--- a/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-hs/src/main/java/org/apache/hadoop/mapreduce/v2/hs/PartialJob.java
+++ b/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-hs/src/main/java/org/apache/hadoop/mapreduce/v2/hs/PartialJob.java
@@ -39,6 +39,7 @@ import org.apache.hadoop.mapreduce.v2.app.job.Task;
import org.apache.hadoop.mapreduce.v2.jobhistory.JobIndexInfo;
import org.apache.hadoop.security.UserGroupInformation;
import org.apache.hadoop.security.authorize.AccessControlList;
+import org.apache.hadoop.yarn.api.records.Priority;
import org.apache.hadoop.yarn.factory.providers.RecordFactoryProvider;
@@ -196,4 +197,10 @@ public class PartialJob implements org.apache.hadoop.mapreduce.v2.app.job.Job {
throw new UnsupportedOperationException("Can't set job's queue name in history");
}
+ @Override
+ public void setJobPriority(Priority priority) {
+ throw new UnsupportedOperationException(
+ "Can't set job's priority in history");
+ }
+
}
http://git-wip-us.apache.org/repos/asf/hadoop/blob/14505800/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-hs/src/test/java/org/apache/hadoop/mapreduce/v2/hs/webapp/TestHsWebServicesAcls.java
----------------------------------------------------------------------
diff --git a/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-hs/src/test/java/org/apache/hadoop/mapreduce/v2/hs/webapp/TestHsWebServicesAcls.java b/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-hs/src/test/java/org/apache/hadoop/mapreduce/v2/hs/webapp/TestHsWebServicesAcls.java
index fd87b94..14961d2 100644
--- a/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-hs/src/test/java/org/apache/hadoop/mapreduce/v2/hs/webapp/TestHsWebServicesAcls.java
+++ b/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-hs/src/test/java/org/apache/hadoop/mapreduce/v2/hs/webapp/TestHsWebServicesAcls.java
@@ -58,6 +58,7 @@ import org.apache.hadoop.security.GroupMappingServiceProvider;
import org.apache.hadoop.security.Groups;
import org.apache.hadoop.security.UserGroupInformation;
import org.apache.hadoop.security.authorize.AccessControlList;
+import org.apache.hadoop.yarn.api.records.Priority;
import org.apache.hadoop.yarn.webapp.WebApp;
import org.junit.Before;
import org.junit.Test;
@@ -419,5 +420,9 @@ public class TestHsWebServicesAcls {
@Override
public void setQueueName(String queueName) {
}
+
+ @Override
+ public void setJobPriority(Priority priority) {
+ }
}
}