YarnClientImpl extends YarnClient { @Override public ApplicationId submitApplication(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); if (isSecurityEnabled() && timelineV1ServiceEnabled && getConfig().get(YarnConfiguration.TIMELINE_HTTP_AUTH_TYPE) .equals(KerberosAuthenticationHandler.TYPE)) { addTimelineDelegationToken(appContext.getAMContainerSpec()); } try { if (isSecurityEnabled()) { addLogAggregationDelegationToken(appContext.getAMContainerSpec()); } } catch (Exception e) { LOG.warn("Failed to obtain delegation token for Log Aggregation Path", e); } 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"); } 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) { String msg = "Interrupted while waiting for application " + applicationId + " to be successfully submitted."; LOG.error(msg); throw new YarnException(msg, ie); } } catch (ApplicationNotFoundException ex) { LOG.info("Re-submit application " + applicationId + "with the " + "same ApplicationSubmissionContext"); rmClient.submitApplication(request); } } return applicationId; } YarnClientImpl(); @Override YarnClientApplication createApplication(); @Override ApplicationId submitApplication(ApplicationSubmissionContext appContext); @Override void failApplicationAttempt(ApplicationAttemptId attemptId); @Override void killApplication(ApplicationId applicationId); @Override void killApplication(ApplicationId applicationId, String diagnostics); @Override ApplicationReport getApplicationReport(ApplicationId appId); org.apache.hadoop.security.token.Token<AMRMTokenIdentifier> getAMRMToken(ApplicationId appId); @Override List<ApplicationReport> getApplications(); @Override List<ApplicationReport> getApplications(Set<String> applicationTypes); @Override List<ApplicationReport> getApplications(
      EnumSet<YarnApplicationState> applicationStates); @Override List<ApplicationReport> getApplications(Set<String> applicationTypes,
      EnumSet<YarnApplicationState> applicationStates); @Override List<ApplicationReport> getApplications(Set<String> applicationTypes,
      EnumSet<YarnApplicationState> applicationStates,
      Set<String> applicationTags); @Override List<ApplicationReport> getApplications(Set<String> queues,
      Set<String> users, Set<String> applicationTypes,
      EnumSet<YarnApplicationState> applicationStates); @Override List<ApplicationReport> getApplications(
      GetApplicationsRequest request); @Override YarnClusterMetrics getYarnClusterMetrics(); @Override List<NodeReport> getNodeReports(NodeState... states); @Override Token getRMDelegationToken(Text renewer); @Override QueueInfo getQueueInfo(String queueName); @Override List<QueueUserACLInfo> getQueueAclsInfo(); @Override List<QueueInfo> getAllQueues(); @Override List<QueueInfo> getRootQueueInfos(); @Override List<QueueInfo> getChildQueueInfos(String parent); @Private @VisibleForTesting void setRMClient(ApplicationClientProtocol rmClient); @Override ApplicationAttemptReport getApplicationAttemptReport(
      ApplicationAttemptId appAttemptId); @Override List<ApplicationAttemptReport> getApplicationAttempts(
      ApplicationId appId); @Override ContainerReport getContainerReport(ContainerId containerId); @Override List<ContainerReport> getContainers(
      ApplicationAttemptId applicationAttemptId); @Override void moveApplicationAcrossQueues(ApplicationId appId,
      String queue); @Override GetNewReservationResponse createReservation(); @Override ReservationSubmissionResponse submitReservation(
      ReservationSubmissionRequest request); @Override ReservationUpdateResponse updateReservation(
      ReservationUpdateRequest request); @Override ReservationDeleteResponse deleteReservation(
      ReservationDeleteRequest request); @Override ReservationListResponse listReservations(
          ReservationListRequest request); @Override Map<NodeId, Set<String>> getNodeToLabels(); @Override Map<String, Set<NodeId>> getLabelsToNodes(); @Override Map<String, Set<NodeId>> getLabelsToNodes(Set<String> labels); @Override List<NodeLabel> getClusterNodeLabels(); @Override Priority updateApplicationPriority(ApplicationId applicationId,
      Priority priority); @Override void signalToContainer(ContainerId containerId,
      SignalContainerCommand command); @Override UpdateApplicationTimeoutsResponse updateApplicationTimeouts(
      UpdateApplicationTimeoutsRequest request); @Override Map<String, Resource> getResourceProfiles(); @Override Resource getResourceProfile(String profile); @Override List<ResourceTypeInfo> getResourceTypeInfo(); @Override Set<NodeAttributeInfo> getClusterAttributes(); @Override Map<NodeAttributeKey, List<NodeToAttributeValue>> getAttributesToNodes(
      Set<NodeAttributeKey> attributes); @Override Map<String, Set<NodeAttribute>> getNodeToAttributes(
      Set<String> hostNames); @Override void shellToContainer(ContainerId containerId,
      ShellContainerCommand command);  }
@Test public void testAutomaitcLogAggregationDelegationToken() throws Exception { Configuration conf = getConf(); SecurityUtil.setAuthenticationMethod( UserGroupInformation.AuthenticationMethod.KERBEROS, conf); conf.set(YarnConfiguration.RM_PRINCIPAL, YARN_RM); String remoteRootLogPath = "/tmp/app-logs"; MiniDFSCluster hdfsCluster = null; try { HdfsConfiguration hdfsConfig = new HdfsConfiguration(); hdfsCluster = new MiniDFSCluster.Builder(hdfsConfig) .numDataNodes(1).build(); Path remoteRootLogDir = new Path(remoteRootLogPath); FileSystem fs = hdfsCluster.getFileSystem(); fs.mkdirs(remoteRootLogDir); conf.set(YarnConfiguration.NM_REMOTE_APP_LOG_DIR, fs.getFileStatus(remoteRootLogDir).getPath().toString()); DelegationTokenIdentifier hdfsDT = new DelegationTokenIdentifier(new Text( "test"), new Text(YARN_RM), null); final Token<DelegationTokenIdentifier> dToken = new Token<>(hdfsDT.getBytes(), new byte[0], hdfsDT.getKind(), new Text()); FileSystem mockFs = mock(FileSystem.class); doAnswer(new Answer<Token<?>[]>() { @Override public Token<?>[] answer(InvocationOnMock invocation) { Object[] args = invocation.getArguments(); ((Credentials) args[1]).addToken(hdfsDT.getKind(), dToken); return new Token[]{dToken}; } }).when(mockFs).addDelegationTokens(any(), any()); FileSystemTestHelper.addFileSystemForTesting(fs.getUri(), hdfsConfig, mockFs); YarnClientImpl client = spy(new YarnClientImpl() { @Override protected void serviceStart() { rmClient = mock(ApplicationClientProtocol.class); } @Override protected void serviceStop() { } @Override public ApplicationReport getApplicationReport(ApplicationId appId) { ApplicationReport report = mock(ApplicationReport.class); when(report.getYarnApplicationState()) .thenReturn(YarnApplicationState.RUNNING); return report; } @Override public boolean isSecurityEnabled() { return true; } }); client.init(conf); client.start(); ApplicationSubmissionContext context = mock(ApplicationSubmissionContext.class); ApplicationId applicationId = ApplicationId.newInstance(0, 1); when(context.getApplicationId()).thenReturn(applicationId); DataOutputBuffer dob = new DataOutputBuffer(); Credentials credentials = new Credentials(); credentials.writeTokenStorageToStream(dob); ByteBuffer tokens = ByteBuffer.wrap(dob.getData(), 0, dob.getLength()); ContainerLaunchContext clc = ContainerLaunchContext.newInstance( null, null, null, null, tokens, null); when(context.getAMContainerSpec()).thenReturn(clc); client.submitApplication(context); credentials = new Credentials(); DataInputByteBuffer dibb = new DataInputByteBuffer(); tokens = clc.getTokens(); if (tokens != null) { dibb.reset(tokens); credentials.readTokenStorageStream(dibb); tokens.rewind(); } Collection<Token<? extends TokenIdentifier>> dTokens = credentials.getAllTokens(); Assert.assertEquals("Failed to place token for Log Aggregation Path", 1, dTokens.size()); Assert.assertEquals("Wrong Token for Log Aggregation", hdfsDT.getKind(), dTokens.iterator().next().getKind()); } finally { if (hdfsCluster != null) { hdfsCluster.shutdown(); } } }