MAPREDUCE-4942. mapreduce.Job has a bunch of methods that throw InterruptedException so its incompatible with MR1. (rkanter via tucu)

git-svn-id: https://svn.apache.org/repos/asf/hadoop/common/branches/branch-2@1480754 13f79535-47bb-0310-9956-ffa450edef68
This commit is contained in:
Alejandro Abdelnur 2013-05-09 20:11:33 +00:00
parent 6aa0edc793
commit c56c50df3f
4 changed files with 80 additions and 84 deletions

View File

@ -233,6 +233,9 @@ Release 2.0.5-beta - UNRELEASED
MAPREDUCE-5226. Handling YarnRemoteException separately from IOException in
MR App's use of AMRMProtocol after YARN-630. (Xuan Gong via vinodkv)
MAPREDUCE-4942. mapreduce.Job has a bunch of methods that throw
InterruptedException so its incompatible with MR1. (rkanter via tucu)
Release 2.0.4-alpha - 2013-04-25
INCOMPATIBLE CHANGES

View File

@ -209,11 +209,7 @@ public class JobClient extends CLI {
* completed.
*/
public float mapProgress() throws IOException {
try {
return job.mapProgress();
} catch (InterruptedException ie) {
throw new IOException(ie);
}
return job.mapProgress();
}
/**
@ -221,11 +217,7 @@ public class JobClient extends CLI {
* completed.
*/
public float reduceProgress() throws IOException {
try {
return job.reduceProgress();
} catch (InterruptedException ie) {
throw new IOException(ie);
}
return job.reduceProgress();
}
/**
@ -245,33 +237,21 @@ public class JobClient extends CLI {
* completed.
*/
public float setupProgress() throws IOException {
try {
return job.setupProgress();
} catch (InterruptedException ie) {
throw new IOException(ie);
}
return job.setupProgress();
}
/**
* Returns immediately whether the whole job is done yet or not.
*/
public synchronized boolean isComplete() throws IOException {
try {
return job.isComplete();
} catch (InterruptedException ie) {
throw new IOException(ie);
}
return job.isComplete();
}
/**
* True iff job completed successfully.
*/
public synchronized boolean isSuccessful() throws IOException {
try {
return job.isSuccessful();
} catch (InterruptedException ie) {
throw new IOException(ie);
}
return job.isSuccessful();
}
/**
@ -302,11 +282,7 @@ public class JobClient extends CLI {
* Tells the service to terminate the current job.
*/
public synchronized void killJob() throws IOException {
try {
job.killJob();
} catch (InterruptedException ie) {
throw new IOException(ie);
}
job.killJob();
}
@ -331,14 +307,10 @@ public class JobClient extends CLI {
*/
public synchronized void killTask(TaskAttemptID taskId,
boolean shouldFail) throws IOException {
try {
if (shouldFail) {
job.failTask(taskId);
} else {
job.killTask(taskId);
}
} catch (InterruptedException ie) {
throw new IOException(ie);
if (shouldFail) {
job.failTask(taskId);
} else {
job.killTask(taskId);
}
}
@ -378,16 +350,12 @@ public class JobClient extends CLI {
* Returns the counters for this job
*/
public Counters getCounters() throws IOException {
try {
Counters result = null;
org.apache.hadoop.mapreduce.Counters temp = job.getCounters();
if(temp != null) {
result = Counters.downgrade(temp);
}
return result;
} catch (InterruptedException ie) {
throw new IOException(ie);
Counters result = null;
org.apache.hadoop.mapreduce.Counters temp = job.getCounters();
if(temp != null) {
result = Counters.downgrade(temp);
}
return result;
}
@Override

View File

@ -296,7 +296,7 @@ public class Job extends JobContextImpl implements JobContext {
* it, if necessary
*/
synchronized void ensureFreshStatus()
throws IOException, InterruptedException {
throws IOException {
if (System.currentTimeMillis() - statustime > MAX_JOBSTATUS_AGE) {
updateStatus();
}
@ -306,13 +306,18 @@ public class Job extends JobContextImpl implements JobContext {
* immediately
* @throws IOException
*/
synchronized void updateStatus() throws IOException, InterruptedException {
this.status = ugi.doAs(new PrivilegedExceptionAction<JobStatus>() {
@Override
public JobStatus run() throws IOException, InterruptedException {
return cluster.getClient().getJobStatus(status.getJobID());
}
});
synchronized void updateStatus() throws IOException {
try {
this.status = ugi.doAs(new PrivilegedExceptionAction<JobStatus>() {
@Override
public JobStatus run() throws IOException, InterruptedException {
return cluster.getClient().getJobStatus(status.getJobID());
}
});
}
catch (InterruptedException ie) {
throw new IOException(ie);
}
if (this.status == null) {
throw new IOException("Job status not available ");
}
@ -537,7 +542,7 @@ public class Job extends JobContextImpl implements JobContext {
* @return the progress of the job's map-tasks.
* @throws IOException
*/
public float mapProgress() throws IOException, InterruptedException {
public float mapProgress() throws IOException {
ensureState(JobState.RUNNING);
ensureFreshStatus();
return status.getMapProgress();
@ -550,7 +555,7 @@ public class Job extends JobContextImpl implements JobContext {
* @return the progress of the job's reduce-tasks.
* @throws IOException
*/
public float reduceProgress() throws IOException, InterruptedException {
public float reduceProgress() throws IOException {
ensureState(JobState.RUNNING);
ensureFreshStatus();
return status.getReduceProgress();
@ -576,7 +581,7 @@ public class Job extends JobContextImpl implements JobContext {
* @return the progress of the job's setup-tasks.
* @throws IOException
*/
public float setupProgress() throws IOException, InterruptedException {
public float setupProgress() throws IOException {
ensureState(JobState.RUNNING);
ensureFreshStatus();
return status.getSetupProgress();
@ -589,7 +594,7 @@ public class Job extends JobContextImpl implements JobContext {
* @return <code>true</code> if the job is complete, else <code>false</code>.
* @throws IOException
*/
public boolean isComplete() throws IOException, InterruptedException {
public boolean isComplete() throws IOException {
ensureState(JobState.RUNNING);
updateStatus();
return status.isJobComplete();
@ -601,7 +606,7 @@ public class Job extends JobContextImpl implements JobContext {
* @return <code>true</code> if the job succeeded, else <code>false</code>.
* @throws IOException
*/
public boolean isSuccessful() throws IOException, InterruptedException {
public boolean isSuccessful() throws IOException {
ensureState(JobState.RUNNING);
updateStatus();
return status.getState() == JobStatus.State.SUCCEEDED;
@ -613,9 +618,14 @@ public class Job extends JobContextImpl implements JobContext {
*
* @throws IOException
*/
public void killJob() throws IOException, InterruptedException {
public void killJob() throws IOException {
ensureState(JobState.RUNNING);
cluster.getClient().killJob(getJobID());
try {
cluster.getClient().killJob(getJobID());
}
catch (InterruptedException ie) {
throw new IOException(ie);
}
}
/**
@ -673,7 +683,7 @@ public class Job extends JobContextImpl implements JobContext {
try {
return getTaskCompletionEvents(startFrom, 10);
} catch (InterruptedException ie) {
throw new RuntimeException(ie);
throw new IOException(ie);
}
}
@ -684,13 +694,18 @@ public class Job extends JobContextImpl implements JobContext {
* @throws IOException
*/
public boolean killTask(final TaskAttemptID taskId)
throws IOException, InterruptedException {
throws IOException {
ensureState(JobState.RUNNING);
return ugi.doAs(new PrivilegedExceptionAction<Boolean>() {
public Boolean run() throws IOException, InterruptedException {
return cluster.getClient().killTask(taskId, false);
}
});
try {
return ugi.doAs(new PrivilegedExceptionAction<Boolean>() {
public Boolean run() throws IOException, InterruptedException {
return cluster.getClient().killTask(taskId, false);
}
});
}
catch (InterruptedException ie) {
throw new IOException(ie);
}
}
/**
@ -700,14 +715,19 @@ public class Job extends JobContextImpl implements JobContext {
* @throws IOException
*/
public boolean failTask(final TaskAttemptID taskId)
throws IOException, InterruptedException {
throws IOException {
ensureState(JobState.RUNNING);
return ugi.doAs(new PrivilegedExceptionAction<Boolean>() {
@Override
public Boolean run() throws IOException, InterruptedException {
return cluster.getClient().killTask(taskId, true);
}
});
try {
return ugi.doAs(new PrivilegedExceptionAction<Boolean>() {
@Override
public Boolean run() throws IOException, InterruptedException {
return cluster.getClient().killTask(taskId, true);
}
});
}
catch (InterruptedException ie) {
throw new IOException(ie);
}
}
/**
@ -718,14 +738,19 @@ public class Job extends JobContextImpl implements JobContext {
* @throws IOException
*/
public Counters getCounters()
throws IOException, InterruptedException {
throws IOException {
ensureState(JobState.RUNNING);
return ugi.doAs(new PrivilegedExceptionAction<Counters>() {
@Override
public Counters run() throws IOException, InterruptedException {
return cluster.getClient().getJobCounters(getJobID());
}
});
try {
return ugi.doAs(new PrivilegedExceptionAction<Counters>() {
@Override
public Counters run() throws IOException, InterruptedException {
return cluster.getClient().getJobCounters(getJobID());
}
});
}
catch (InterruptedException ie) {
throw new IOException(ie);
}
}
/**

View File

@ -346,7 +346,7 @@ public class TestGridmixSummary {
};
@Override
public boolean isSuccessful() throws IOException, InterruptedException {
public boolean isSuccessful() throws IOException {
if (lost) {
throw new IOException("Test failure!");
}