package org.apache.hadoop.examples;import java.io.IOException;
import java.util.StringTokenizer;import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.util.GenericOptionsParser;public class WordCount {//继承泛型类Mapperpublic static class TokenizerMapper extends Mapper<Object, Text, Text, IntWritable>{//定义hadoop数据类型IntWritable实例one，并且赋值为1private final static IntWritable one = new IntWritable(1);//定义hadoop数据类型Text实例wordprivate Text word = new Text();//实现map函数    public void map(Object key, Text value, Context context) throws IOException, InterruptedException {//Java的字符串分解类，默认分隔符“空格”、“制表符(‘\t’)”、“换行符(‘\n’)”、“回车符(‘\r’)”StringTokenizer itr = new StringTokenizer(value.toString());//循环条件表示返回是否还有分隔符。while (itr.hasMoreTokens()) {/*nextToken()：返回从当前位置到下一个分隔符的字符串word.set()Java数据类型与hadoop数据类型转换*/word.set(itr.nextToken());//hadoop全局类context输出函数write;
        context.write(word, one);}}}//继承泛型类Reducerpublic static class IntSumReducer extends Reducer<Text,IntWritable,Text,IntWritable> {//实例化IntWritableprivate IntWritable result = new IntWritable();//实现reducepublic void reduce(Text key, Iterable<IntWritable> values, Context context) throws IOException, InterruptedException {int sum = 0;//循环values，并记录单词个数for (IntWritable val : values) {sum += val.get();}//Java数据类型sum，转换为hadoop数据类型result
      result.set(sum);//输出结果到hdfs
      context.write(key, result);}}public static void main(String[] args) throws Exception {//实例化ConfigurationConfiguration conf = new Configuration();/*GenericOptionsParser是hadoop框架中解析命令行参数的基本类。getRemainingArgs();返回数组【一组路径】*//*函数实现public String[] getRemainingArgs() {return (commandLine == null) ? new String[]{} : commandLine.getArgs();}*/String[] otherArgs = new GenericOptionsParser(conf, args).getRemainingArgs();//如果只有一个路径，则输出需要有输入路径和输出路径if (otherArgs.length < 2) {System.err.println("Usage: wordcount <in> [<in>...] <out>");System.exit(2);}//实例化jobJob job = Job.getInstance(conf, "word count");job.setJarByClass(WordCount.class);job.setMapperClass(TokenizerMapper.class);/*指定CombinerClass类这里很多人对CombinerClass不理解*/job.setCombinerClass(IntSumReducer.class);job.setReducerClass(IntSumReducer.class);//rduce输出Key的类型，是Textjob.setOutputKeyClass(Text.class);// rduce输出Value的类型job.setOutputValueClass(IntWritable.class);//添加输入路径for (int i = 0; i < otherArgs.length - 1; ++i) {FileInputFormat.addInputPath(job, new Path(otherArgs[i]));}//添加输出路径
    FileOutputFormat.setOutputPath(job,new Path(otherArgs[otherArgs.length - 1]));//提交jobSystem.exit(job.waitForCompletion(true) ? 0 : 1);}
}

     public class MyMapper<K extends WritableComparable, V extends Writable> extends MapReduceBase implements Mapper<K, V, K, V> {static enum MyCounters { NUM_RECORDS }private String mapTaskId;private String inputFile;private int noRecords = 0;public void configure(JobConf job) { mapTaskId = job.get(JobContext.TASK_ATTEMPT_ID); inputFile = job.get(JobContext.MAP_INPUT_FILE); } public void map(K key, V val, OutputCollector<K, V> output, Reporter reporter) throws IOException { // Process the <key, value> pair (assume this takes a while) // ... // ... // Let the framework know that we are alive, and kicking! // reporter.progress(); // Process some more // ... // ... // Increment the no. of <key, value> pairs processed ++noRecords; // Increment counters reporter.incrCounter(NUM_RECORDS, 1); // Every 100 records update application-level status if ((noRecords%100) == 0) { reporter.setStatus(mapTaskId + " processed " + noRecords + " from input-file: " + inputFile); } // Output the result  output.collect(key, val); } }

 public class TokenCounterMapper extends Mapper<Object, Text, Text, IntWritable>{private final static IntWritable one = new IntWritable(1);private Text word = new Text();public void map(Object key, Text value, Context context) throws IOException, InterruptedException {StringTokenizer itr = new StringTokenizer(value.toString()); while (itr.hasMoreTokens()) { word.set(itr.nextToken()); context.write(word, one); } } }

  void map(K1 key, V1 value, OutputCollector<K2, V2> output, Reporter reporter)throws IOException;

     public class MyReducer<K extends WritableComparable, V extends Writable> extends MapReduceBase implements Reducer<K, V, K, V> {static enum MyCounters { NUM_RECORDS }private String reduceTaskId;private int noKeys = 0;public void configure(JobConf job) {reduceTaskId = job.get(JobContext.TASK_ATTEMPT_ID);}public void reduce(K key, Iterator<V> values,OutputCollector<K, V> output, Reporter reporter)throws IOException {// Processint noValues = 0;while (values.hasNext()) {V value = values.next();// Increment the no. of values for this key++noValues;// Process the <key, value> pair (assume this takes a while)// ...// ...// Let the framework know that we are alive, and kicking!if ((noValues%10) == 0) {reporter.progress();}// Process some more// ...// ...// Output the <key, value>
            output.collect(key, value);}// Increment the no. of <key, list of values> pairs processed++noKeys;// Increment countersreporter.incrCounter(NUM_RECORDS, 1);// Every 100 keys update application-level statusif ((noKeys%100) == 0) {reporter.setStatus(reduceTaskId + " processed " + noKeys);}}}

 /*** Submit the job to the cluster and wait for it to finish.* @param verbose print the progress to the user* @return true if the job succeeded* @throws IOException thrown if the communication with the *         <code>JobTracker</code> is lost*/public boolean waitForCompletion(boolean verbose) throws IOException, InterruptedException,ClassNotFoundException {if (state == JobState.DEFINE) {submit();}if (verbose) {monitorAndPrintJob();} else {// get the completion poll interval from the client.int completionPollIntervalMillis = Job.getCompletionPollInterval(cluster.getConf());while (!isComplete()) {try {Thread.sleep(completionPollIntervalMillis);} catch (InterruptedException ie) {}}}return isSuccessful();}

/*** Submit the job to the cluster and return immediately.* @throws IOException*/public void submit() throws IOException, InterruptedException, ClassNotFoundException {ensureState(JobState.DEFINE);setUseNewAPI();connect();final JobSubmitter submitter = getJobSubmitter(cluster.getFileSystem(), cluster.getClient());status = ugi.doAs(new PrivilegedExceptionAction<JobStatus>() {public JobStatus run() throws IOException, InterruptedException, ClassNotFoundException {return submitter.submitJobInternal(Job.this, cluster);}});state = JobState.RUNNING;LOG.info("The url to track the job: " + getTrackingURL());}

  private synchronized void connect()throws IOException, InterruptedException, ClassNotFoundException {if (cluster == null) {cluster = ugi.doAs(new PrivilegedExceptionAction<Cluster>() {public Cluster run()throws IOException, InterruptedException, ClassNotFoundException {return new Cluster(getConfiguration());}});}}

public static enum JobTrackerStatus {INITIALIZING, RUNNING};private ClientProtocolProvider clientProtocolProvider;private ClientProtocol client;private UserGroupInformation ugi;private Configuration conf;private FileSystem fs = null;private Path sysDir = null;private Path stagingAreaDir = null;private Path jobHistoryDir = null;

/*** Internal method for submitting jobs to the system.* * <p>The job submission process involves:* <ol>*   <li>*   Checking the input and output specifications of the job.*   </li>*   <li>*   Computing the {@link InputSplit}s for the job.*   </li>*   <li>*   Setup the requisite accounting information for the *   {@link DistributedCache} of the job, if necessary.*   </li>*   <li>*   Copying the job's jar and configuration to the map-reduce system*   directory on the distributed file-system. *   </li>*   <li>*   Submitting the job to the <code>JobTracker</code> and optionally*   monitoring it's status.*   </li>* </ol></p>* @param job the configuration to submit* @param cluster the handle to the Cluster* @throws ClassNotFoundException* @throws InterruptedException* @throws IOException*/JobStatus submitJobInternal(Job job, Cluster cluster) throws ClassNotFoundException, InterruptedException, IOException {//validate the jobs output specs
    checkSpecs(job);Configuration conf = job.getConfiguration();addMRFrameworkToDistributedCache(conf);Path jobStagingArea = JobSubmissionFiles.getStagingDir(cluster, conf);//configure the command line options correctly on the submitting dfsInetAddress ip = InetAddress.getLocalHost();if (ip != null) {submitHostAddress = ip.getHostAddress();submitHostName = ip.getHostName();conf.set(MRJobConfig.JOB_SUBMITHOST,submitHostName);conf.set(MRJobConfig.JOB_SUBMITHOSTADDR,submitHostAddress);}JobID jobId = submitClient.getNewJobID();job.setJobID(jobId);Path submitJobDir = new Path(jobStagingArea, jobId.toString());JobStatus status = null;try {conf.set(MRJobConfig.USER_NAME,UserGroupInformation.getCurrentUser().getShortUserName());conf.set("hadoop.http.filter.initializers", "org.apache.hadoop.yarn.server.webproxy.amfilter.AmFilterInitializer");conf.set(MRJobConfig.MAPREDUCE_JOB_DIR, submitJobDir.toString());LOG.debug("Configuring job " + jobId + " with " + submitJobDir + " as the submit dir");// get delegation token for the dir
      TokenCache.obtainTokensForNamenodes(job.getCredentials(),new Path[] { submitJobDir }, conf);populateTokenCache(conf, job.getCredentials());// generate a secret to authenticate shuffle transfersif (TokenCache.getShuffleSecretKey(job.getCredentials()) == null) {KeyGenerator keyGen;try {int keyLen = CryptoUtils.isShuffleEncrypted(conf) ? conf.getInt(MRJobConfig.MR_ENCRYPTED_INTERMEDIATE_DATA_KEY_SIZE_BITS, MRJobConfig.DEFAULT_MR_ENCRYPTED_INTERMEDIATE_DATA_KEY_SIZE_BITS): SHUFFLE_KEY_LENGTH;keyGen = KeyGenerator.getInstance(SHUFFLE_KEYGEN_ALGORITHM);keyGen.init(keyLen);} catch (NoSuchAlgorithmException e) {throw new IOException("Error generating shuffle secret key", e);}SecretKey shuffleKey = keyGen.generateKey();TokenCache.setShuffleSecretKey(shuffleKey.getEncoded(),job.getCredentials());}copyAndConfigureFiles(job, submitJobDir);Path submitJobFile = JobSubmissionFiles.getJobConfPath(submitJobDir);// Create the splits for the jobLOG.debug("Creating splits at " + jtFs.makeQualified(submitJobDir));int maps = writeSplits(job, submitJobDir);conf.setInt(MRJobConfig.NUM_MAPS, maps);LOG.info("number of splits:" + maps);// write "queue admins of the queue to which job is being submitted"// to job file.String queue = conf.get(MRJobConfig.QUEUE_NAME,JobConf.DEFAULT_QUEUE_NAME);AccessControlList acl = submitClient.getQueueAdmins(queue);conf.set(toFullPropertyName(queue,QueueACL.ADMINISTER_JOBS.getAclName()), acl.getAclString());// removing jobtoken referrals before copying the jobconf to HDFS// as the tasks don't need this setting, actually they may break// because of it if present as the referral will point to a// different job.
      TokenCache.cleanUpTokenReferral(conf);if (conf.getBoolean(MRJobConfig.JOB_TOKEN_TRACKING_IDS_ENABLED,MRJobConfig.DEFAULT_JOB_TOKEN_TRACKING_IDS_ENABLED)) {// Add HDFS tracking idsArrayList<String> trackingIds = new ArrayList<String>();for (Token<? extends TokenIdentifier> t :job.getCredentials().getAllTokens()) {trackingIds.add(t.decodeIdentifier().getTrackingId());}conf.setStrings(MRJobConfig.JOB_TOKEN_TRACKING_IDS,trackingIds.toArray(new String[trackingIds.size()]));}// Set reservation info if it existsReservationId reservationId = job.getReservationId();if (reservationId != null) {conf.set(MRJobConfig.RESERVATION_ID, reservationId.toString());}// Write job file to submit dir
      writeConf(conf, submitJobFile);//// Now, actually submit the job (using the submit name)//
      printTokens(jobId, job.getCredentials());
      status = submitClient.submitJob(jobId, submitJobDir.toString(), job.getCredentials());if (status != null) {return status;} else {throw new IOException("Could not launch job");}} finally {if (status == null) {LOG.info("Cleaning up the staging area " + submitJobDir);if (jtFs != null && submitJobDir != null)jtFs.delete(submitJobDir, true);}}}

 @Overridepublic JobStatus submitJob(JobID jobId, String jobSubmitDir, Credentials ts)throws IOException, InterruptedException {addHistoryToken(ts);// Construct necessary information to start the MR AMApplicationSubmissionContext appContext =createApplicationSubmissionContext(conf, jobSubmitDir, ts);// Submit to ResourceManagertry {
      ApplicationId applicationId =resMgrDelegate.submitApplication(appContext);ApplicationReport appMaster = resMgrDelegate.getApplicationReport(applicationId);String diagnostics =(appMaster == null ?"application report is null" : appMaster.getDiagnostics());if (appMaster == null|| appMaster.getYarnApplicationState() == YarnApplicationState.FAILED|| appMaster.getYarnApplicationState() == YarnApplicationState.KILLED) {throw new IOException("Failed to run job : " +diagnostics);}return clientCache.getClient(jobId).getJobStatus(jobId);} catch (YarnException e) {throw new IOException(e);}}

@Overridepublic ApplicationIdsubmitApplication(ApplicationSubmissionContext appContext)throws YarnException, IOException {ApplicationId applicationId = appContext.getApplicationId();if (applicationId == null) {throw new ApplicationIdNotProvidedException("ApplicationId is not provided in ApplicationSubmissionContext");}SubmitApplicationRequest request =Records.newRecord(SubmitApplicationRequest.class);request.setApplicationSubmissionContext(appContext);// Automatically add the timeline DT into the CLC// Only when the security and the timeline service are both enabledif (isSecurityEnabled() && timelineServiceEnabled) {addTimelineDelegationToken(appContext.getAMContainerSpec());}//TODO: YARN-1763:Handle RM failovers during the submitApplication call.
    rmClient.submitApplication(request);int pollCount = 0;long startTime = System.currentTimeMillis();EnumSet<YarnApplicationState> waitingStates = EnumSet.of(YarnApplicationState.NEW,YarnApplicationState.NEW_SAVING,YarnApplicationState.SUBMITTED);EnumSet<YarnApplicationState> failToSubmitStates = EnumSet.of(YarnApplicationState.FAILED,YarnApplicationState.KILLED);        while (true) {try {ApplicationReport appReport = getApplicationReport(applicationId);YarnApplicationState state = appReport.getYarnApplicationState();if (!waitingStates.contains(state)) {if(failToSubmitStates.contains(state)) {throw new YarnException("Failed to submit " + applicationId + " to YARN : " + appReport.getDiagnostics());}LOG.info("Submitted application " + applicationId);break;}long elapsedMillis = System.currentTimeMillis() - startTime;if (enforceAsyncAPITimeout() &&elapsedMillis >= asyncApiPollTimeoutMillis) {throw new YarnException("Timed out while waiting for application " +applicationId + " to be submitted successfully");}// Notify the client through the log every 10 poll, in case the client// is blocked here too long.if (++pollCount % 10 == 0) {LOG.info("Application submission is not finished, " +"submitted application " + applicationId +" is still in " + state);}try {Thread.sleep(submitPollIntervalMillis);} catch (InterruptedException ie) {LOG.error("Interrupted while waiting for application "+ applicationId+ " to be successfully submitted.");}} catch (ApplicationNotFoundException ex) {// FailOver or RM restart happens before RMStateStore saves// ApplicationStateLOG.info("Re-submit application " + applicationId + "with the " +"same ApplicationSubmissionContext");rmClient.submitApplication(request);}}return applicationId;}

 @Overridepublic SubmitApplicationResponse submitApplication(SubmitApplicationRequest request) throws YarnException {ApplicationSubmissionContext submissionContext = request.getApplicationSubmissionContext();ApplicationId applicationId = submissionContext.getApplicationId();// ApplicationSubmissionContext needs to be validated for safety - only// those fields that are independent of the RM's configuration will be// checked here, those that are dependent on RM configuration are validated// in RMAppManager.
String user = null;try {// Safetyuser = UserGroupInformation.getCurrentUser().getShortUserName();} catch (IOException ie) {LOG.warn("Unable to get the current user.", ie);RMAuditLogger.logFailure(user, AuditConstants.SUBMIT_APP_REQUEST,ie.getMessage(), "ClientRMService","Exception in submitting application", applicationId);throw RPCUtil.getRemoteException(ie);}// Check whether app has already been put into rmContext,// If it is, simply return the responseif (rmContext.getRMApps().get(applicationId) != null) {LOG.info("This is an earlier submitted application: " + applicationId);return SubmitApplicationResponse.newInstance();}if (submissionContext.getQueue() == null) {submissionContext.setQueue(YarnConfiguration.DEFAULT_QUEUE_NAME);}if (submissionContext.getApplicationName() == null) {submissionContext.setApplicationName(YarnConfiguration.DEFAULT_APPLICATION_NAME);}if (submissionContext.getApplicationType() == null) {submissionContext.setApplicationType(YarnConfiguration.DEFAULT_APPLICATION_TYPE);} else {if (submissionContext.getApplicationType().length() > YarnConfiguration.APPLICATION_TYPE_LENGTH) {submissionContext.setApplicationType(submissionContext.getApplicationType().substring(0,YarnConfiguration.APPLICATION_TYPE_LENGTH));}}try {// call RMAppManager to submit application directly
      rmAppManager.submitApplication(submissionContext,System.currentTimeMillis(), user);LOG.info("Application with id " + applicationId.getId() + " submitted by user " + user);RMAuditLogger.logSuccess(user, AuditConstants.SUBMIT_APP_REQUEST,"ClientRMService", applicationId);} catch (YarnException e) {LOG.info("Exception in submitting application with id " +applicationId.getId(), e);RMAuditLogger.logFailure(user, AuditConstants.SUBMIT_APP_REQUEST,e.getMessage(), "ClientRMService","Exception in submitting application", applicationId);throw e;}SubmitApplicationResponse response = recordFactory.newRecordInstance(SubmitApplicationResponse.class);return response;}

 @SuppressWarnings("unchecked")protected void submitApplication(ApplicationSubmissionContext submissionContext, long submitTime,String user) throws YarnException {ApplicationId applicationId = submissionContext.getApplicationId();RMAppImpl application =createAndPopulateNewRMApp(submissionContext, submitTime, user);ApplicationId appId = submissionContext.getApplicationId();if (UserGroupInformation.isSecurityEnabled()) {try {this.rmContext.getDelegationTokenRenewer().addApplicationAsync(appId,parseCredentials(submissionContext),submissionContext.getCancelTokensWhenComplete(),application.getUser());} catch (Exception e) {LOG.warn("Unable to parse credentials.", e);// Sending APP_REJECTED is fine, since we assume that the// RMApp is in NEW state and thus we haven't yet informed the// scheduler about the existence of the applicationassert application.getState() == RMAppState.NEW;this.rmContext.getDispatcher().getEventHandler().handle(new RMAppRejectedEvent(applicationId, e.getMessage()));throw RPCUtil.getRemoteException(e);}} else {// Dispatcher is not yet started at this time, so these START events// enqueued should be guaranteed to be first processed when dispatcher// gets started.this.rmContext.getDispatcher().getEventHandler().handle(new RMAppEvent(applicationId, RMAppEventType.START));}}

 @SuppressWarnings("unchecked")protected void submitApplication(ApplicationSubmissionContext submissionContext, long submitTime,String user) throws YarnException {ApplicationId applicationId = submissionContext.getApplicationId();RMAppImpl application =createAndPopulateNewRMApp(submissionContext, submitTime, user);ApplicationId appId = submissionContext.getApplicationId();if (UserGroupInformation.isSecurityEnabled()) {try {this.rmContext.getDelegationTokenRenewer().addApplicationAsync(appId,parseCredentials(submissionContext),submissionContext.getCancelTokensWhenComplete(),application.getUser());} catch (Exception e) {LOG.warn("Unable to parse credentials.", e);// Sending APP_REJECTED is fine, since we assume that the// RMApp is in NEW state and thus we haven't yet informed the// scheduler about the existence of the applicationassert application.getState() == RMAppState.NEW;this.rmContext.getDispatcher().getEventHandler().handle(new RMAppRejectedEvent(applicationId, e.getMessage()));throw RPCUtil.getRemoteException(e);}} else {// Dispatcher is not yet started at this time, so these START events// enqueued should be guaranteed to be first processed when dispatcher// gets started.this.rmContext.getDispatcher().getEventHandler().handle(new RMAppEvent(applicationId, RMAppEventType.START));}}

  /*** Asynchronously add application tokens for renewal.* @param applicationId added application* @param ts tokens* @param shouldCancelAtEnd true if tokens should be canceled when the app is* done else false. * @param user user*/public void addApplicationAsync(ApplicationId applicationId, Credentials ts,boolean shouldCancelAtEnd, String user) {
    processDelegationTokenRenewerEvent(new DelegationTokenRenewerAppSubmitEvent(applicationId, ts, shouldCancelAtEnd, user));}

  private void processDelegationTokenRenewerEvent(DelegationTokenRenewerEvent evt) {serviceStateLock.readLock().lock();try {if (isServiceStarted) {renewerService.execute(new DelegationTokenRenewerRunnable(evt));} else {pendingEventQueue.add(evt);}} finally {serviceStateLock.readLock().unlock();}}

 @Overridepublic void run() {if (evt instanceof DelegationTokenRenewerAppSubmitEvent) {DelegationTokenRenewerAppSubmitEvent appSubmitEvt =(DelegationTokenRenewerAppSubmitEvent) evt; handleDTRenewerAppSubmitEvent(appSubmitEvt);} else if (evt.getType().equals(DelegationTokenRenewerEventType.FINISH_APPLICATION)) {DelegationTokenRenewer.this.handleAppFinishEvent(evt);}}

 @SuppressWarnings("unchecked")private void handleDTRenewerAppSubmitEvent(DelegationTokenRenewerAppSubmitEvent event) {/** For applications submitted with delegation tokens we are not submitting* the application to scheduler from RMAppManager. Instead we are doing* it from here. The primary goal is to make token renewal as a part of* application submission asynchronous so that client thread is not* blocked during app submission.*/try {// Setup tokens for renewalDelegationTokenRenewer.this.handleAppSubmitEvent(event);rmContext.getDispatcher().getEventHandler().handle(new RMAppEvent(event.getApplicationId(), RMAppEventType.START));} catch (Throwable t) {LOG.warn("Unable to add the application to the delegation token renewer.",t);// Sending APP_REJECTED is fine, since we assume that the// RMApp is in NEW state and thus we havne't yet informed the// Scheduler about the existence of the application
        rmContext.getDispatcher().getEventHandler().handle(new RMAppRejectedEvent(event.getApplicationId(), t.getMessage()));}}}

private void handleAppSubmitEvent(DelegationTokenRenewerAppSubmitEvent evt)throws IOException, InterruptedException {ApplicationId applicationId = evt.getApplicationId();Credentials ts = evt.getCredentials();boolean shouldCancelAtEnd = evt.shouldCancelAtEnd();if (ts == null) {return; // nothing to add
    }if (LOG.isDebugEnabled()) {LOG.debug("Registering tokens for renewal for:" +" appId = " + applicationId);}Collection<Token<?>> tokens = ts.getAllTokens();long now = System.currentTimeMillis();// find tokens for renewal, but don't add timers until we know// all renewable tokens are valid// At RM restart it is safe to assume that all the previously added tokens// are valid
    appTokens.put(applicationId,Collections.synchronizedSet(new HashSet<DelegationTokenToRenew>()));Set<DelegationTokenToRenew> tokenList = new HashSet<DelegationTokenToRenew>();boolean hasHdfsToken = false;for (Token<?> token : tokens) {if (token.isManaged()) {if (token.getKind().equals(new Text("HDFS_DELEGATION_TOKEN"))) {LOG.info(applicationId + " found existing hdfs token " + token);hasHdfsToken = true;}DelegationTokenToRenew dttr = allTokens.get(token);if (dttr == null) {dttr = new DelegationTokenToRenew(Arrays.asList(applicationId), token,getConfig(), now, shouldCancelAtEnd, evt.getUser());try {renewToken(dttr);} catch (IOException ioe) {throw new IOException("Failed to renew token: " + dttr.token, ioe);}}tokenList.add(dttr);}}if (!tokenList.isEmpty()) {// Renewing token and adding it to timer calls are separated purposefully// If user provides incorrect token then it should not be added for// renewal.for (DelegationTokenToRenew dtr : tokenList) {DelegationTokenToRenew currentDtr =allTokens.putIfAbsent(dtr.token, dtr);if (currentDtr != null) {// another job beat us
          currentDtr.referringAppIds.add(applicationId);appTokens.get(applicationId).add(currentDtr);} else {appTokens.get(applicationId).add(dtr);setTimerForTokenRenewal(dtr);}}}if (!hasHdfsToken) {requestNewHdfsDelegationToken(Arrays.asList(applicationId), evt.getUser(),shouldCancelAtEnd);}}