MAPREDUCE-3141. Fix the broken MRAppMaster to work over YARN in security mode.(vinodkv)
git-svn-id: https://svn.apache.org/repos/asf/hadoop/common/trunk@1180007 13f79535-47bb-0310-9956-ffa450edef68
This commit is contained in:
parent
a56f3931d1
commit
fa2529c931
@ -1537,6 +1537,9 @@ Release 0.23.0 - Unreleased
|
|||||||
MAPREDUCE-2783. Fixing RM web-UI to show no tracking-URL when AM
|
MAPREDUCE-2783. Fixing RM web-UI to show no tracking-URL when AM
|
||||||
crashes. (Eric Payne via vinodkv)
|
crashes. (Eric Payne via vinodkv)
|
||||||
|
|
||||||
|
MAPREDUCE-3141. Fix the broken MRAppMaster to work over YARN in security
|
||||||
|
mode.(vinodkv)
|
||||||
|
|
||||||
Release 0.22.0 - Unreleased
|
Release 0.22.0 - Unreleased
|
||||||
|
|
||||||
INCOMPATIBLE CHANGES
|
INCOMPATIBLE CHANGES
|
||||||
|
@ -131,7 +131,9 @@ public class MRAppMaster extends CompositeService {
|
|||||||
private JobEventDispatcher jobEventDispatcher;
|
private JobEventDispatcher jobEventDispatcher;
|
||||||
|
|
||||||
private Job job;
|
private Job job;
|
||||||
|
private Credentials fsTokens = new Credentials(); // Filled during init
|
||||||
|
private UserGroupInformation currentUser; // Will be setup during init
|
||||||
|
|
||||||
public MRAppMaster(ApplicationAttemptId applicationAttemptId) {
|
public MRAppMaster(ApplicationAttemptId applicationAttemptId) {
|
||||||
this(applicationAttemptId, new SystemClock());
|
this(applicationAttemptId, new SystemClock());
|
||||||
}
|
}
|
||||||
@ -146,6 +148,9 @@ public MRAppMaster(ApplicationAttemptId applicationAttemptId, Clock clock) {
|
|||||||
|
|
||||||
@Override
|
@Override
|
||||||
public void init(final Configuration conf) {
|
public void init(final Configuration conf) {
|
||||||
|
|
||||||
|
downloadTokensAndSetupUGI(conf);
|
||||||
|
|
||||||
context = new RunningAppContext(conf);
|
context = new RunningAppContext(conf);
|
||||||
|
|
||||||
// Job name is the same as the app name util we support DAG of jobs
|
// Job name is the same as the app name util we support DAG of jobs
|
||||||
@ -226,37 +231,6 @@ public void init(final Configuration conf) {
|
|||||||
/** Create and initialize (but don't start) a single job. */
|
/** Create and initialize (but don't start) a single job. */
|
||||||
protected Job createJob(Configuration conf) {
|
protected Job createJob(Configuration conf) {
|
||||||
|
|
||||||
// ////////// Obtain the tokens needed by the job. //////////
|
|
||||||
Credentials fsTokens = new Credentials();
|
|
||||||
UserGroupInformation currentUser = null;
|
|
||||||
|
|
||||||
try {
|
|
||||||
currentUser = UserGroupInformation.getCurrentUser();
|
|
||||||
|
|
||||||
if (UserGroupInformation.isSecurityEnabled()) {
|
|
||||||
// Read the file-system tokens from the localized tokens-file.
|
|
||||||
Path jobSubmitDir =
|
|
||||||
FileContext.getLocalFSFileContext().makeQualified(
|
|
||||||
new Path(new File(MRJobConfig.JOB_SUBMIT_DIR)
|
|
||||||
.getAbsolutePath()));
|
|
||||||
Path jobTokenFile =
|
|
||||||
new Path(jobSubmitDir, MRJobConfig.APPLICATION_TOKENS_FILE);
|
|
||||||
fsTokens.addAll(Credentials.readTokenStorageFile(jobTokenFile, conf));
|
|
||||||
LOG.info("jobSubmitDir=" + jobSubmitDir + " jobTokenFile="
|
|
||||||
+ jobTokenFile);
|
|
||||||
|
|
||||||
for (Token<? extends TokenIdentifier> tk : fsTokens.getAllTokens()) {
|
|
||||||
LOG.info(" --- DEBUG: Token of kind " + tk.getKind()
|
|
||||||
+ "in current ugi in the AppMaster for service "
|
|
||||||
+ tk.getService());
|
|
||||||
currentUser.addToken(tk); // For use by AppMaster itself.
|
|
||||||
}
|
|
||||||
}
|
|
||||||
} catch (IOException e) {
|
|
||||||
throw new YarnException(e);
|
|
||||||
}
|
|
||||||
// ////////// End of obtaining the tokens needed by the job. //////////
|
|
||||||
|
|
||||||
// create single job
|
// create single job
|
||||||
Job newJob = new JobImpl(appAttemptID, conf, dispatcher.getEventHandler(),
|
Job newJob = new JobImpl(appAttemptID, conf, dispatcher.getEventHandler(),
|
||||||
taskAttemptListener, jobTokenSecretManager, fsTokens, clock,
|
taskAttemptListener, jobTokenSecretManager, fsTokens, clock,
|
||||||
@ -297,6 +271,42 @@ public void handle(JobFinishEvent event) {
|
|||||||
return newJob;
|
return newJob;
|
||||||
} // end createJob()
|
} // end createJob()
|
||||||
|
|
||||||
|
|
||||||
|
/**
|
||||||
|
* Obtain the tokens needed by the job and put them in the UGI
|
||||||
|
* @param conf
|
||||||
|
*/
|
||||||
|
protected void downloadTokensAndSetupUGI(Configuration conf) {
|
||||||
|
|
||||||
|
try {
|
||||||
|
this.currentUser = UserGroupInformation.getCurrentUser();
|
||||||
|
|
||||||
|
if (UserGroupInformation.isSecurityEnabled()) {
|
||||||
|
// Read the file-system tokens from the localized tokens-file.
|
||||||
|
Path jobSubmitDir =
|
||||||
|
FileContext.getLocalFSFileContext().makeQualified(
|
||||||
|
new Path(new File(MRJobConfig.JOB_SUBMIT_DIR)
|
||||||
|
.getAbsolutePath()));
|
||||||
|
Path jobTokenFile =
|
||||||
|
new Path(jobSubmitDir, MRJobConfig.APPLICATION_TOKENS_FILE);
|
||||||
|
fsTokens.addAll(Credentials.readTokenStorageFile(jobTokenFile, conf));
|
||||||
|
LOG.info("jobSubmitDir=" + jobSubmitDir + " jobTokenFile="
|
||||||
|
+ jobTokenFile);
|
||||||
|
|
||||||
|
for (Token<? extends TokenIdentifier> tk : fsTokens.getAllTokens()) {
|
||||||
|
if (LOG.isDebugEnabled()) {
|
||||||
|
LOG.debug("Token of kind " + tk.getKind()
|
||||||
|
+ "in current ugi in the AppMaster for service "
|
||||||
|
+ tk.getService());
|
||||||
|
}
|
||||||
|
currentUser.addToken(tk); // For use by AppMaster itself.
|
||||||
|
}
|
||||||
|
}
|
||||||
|
} catch (IOException e) {
|
||||||
|
throw new YarnException(e);
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
protected void addIfService(Object object) {
|
protected void addIfService(Object object) {
|
||||||
if (object instanceof Service) {
|
if (object instanceof Service) {
|
||||||
addService((Service) object);
|
addService((Service) object);
|
||||||
|
Loading…
Reference in New Issue
Block a user