org.apache.hadoop.fs.FilterFileSystem#org.apache.hadoop.yarn.api.ApplicationClientProtocol源码实例Demo

@SuppressWarnings("unchecked")
@Override
public long renew(Token<?> token, Configuration conf) throws IOException,
    InterruptedException {
  final ApplicationClientProtocol rmClient = getRmClient(token, conf);
  if (rmClient != null) {
    try {
      RenewDelegationTokenRequest request =
          Records.newRecord(RenewDelegationTokenRequest.class);
      request.setDelegationToken(convertToProtoToken(token));
      return rmClient.renewDelegationToken(request).getNextExpirationTime();
    } catch (YarnException e) {
      throw new IOException(e);
    } finally {
      RPC.stopProxy(rmClient);
    }
  } else {
    return localSecretManager.renewToken(
        (Token<RMDelegationTokenIdentifier>)token, getRenewer(token));
  }
}
 

@SuppressWarnings("unchecked")
@Override
public void cancel(Token<?> token, Configuration conf) throws IOException,
    InterruptedException {
  final ApplicationClientProtocol rmClient = getRmClient(token, conf);
  if (rmClient != null) {
    try {
      CancelDelegationTokenRequest request =
          Records.newRecord(CancelDelegationTokenRequest.class);
      request.setDelegationToken(convertToProtoToken(token));
      rmClient.cancelDelegationToken(request);
    } catch (YarnException e) {
      throw new IOException(e);
    } finally {
      RPC.stopProxy(rmClient);
    }
  } else {
    localSecretManager.cancelToken(
        (Token<RMDelegationTokenIdentifier>)token, getRenewer(token));
  }
}
 

private static ApplicationClientProtocol getRmClient(Token<?> token,
    Configuration conf) throws IOException {
  String[] services = token.getService().toString().split(",");
  for (String service : services) {
    InetSocketAddress addr = NetUtils.createSocketAddr(service);
    if (localSecretManager != null) {
      // return null if it's our token
      if (localServiceAddress.getAddress().isAnyLocalAddress()) {
        if (NetUtils.isLocalAddress(addr.getAddress()) &&
            addr.getPort() == localServiceAddress.getPort()) {
          return null;
        }
      } else if (addr.equals(localServiceAddress)) {
        return null;
      }
    }
  }
  return ClientRMProxy.createRMProxy(conf, ApplicationClientProtocol.class);
}
 

@Private
@Override
protected InetSocketAddress getRMAddress(YarnConfiguration conf,
    Class<?> protocol) throws IOException {
  if (protocol == ApplicationClientProtocol.class) {
    return conf.getSocketAddr(YarnConfiguration.RM_ADDRESS,
        YarnConfiguration.DEFAULT_RM_ADDRESS,
        YarnConfiguration.DEFAULT_RM_PORT);
  } else if (protocol == ResourceManagerAdministrationProtocol.class) {
    return conf.getSocketAddr(
        YarnConfiguration.RM_ADMIN_ADDRESS,
        YarnConfiguration.DEFAULT_RM_ADMIN_ADDRESS,
        YarnConfiguration.DEFAULT_RM_ADMIN_PORT);
  } else if (protocol == ApplicationMasterProtocol.class) {
    setAMRMTokenService(conf);
    return conf.getSocketAddr(YarnConfiguration.RM_SCHEDULER_ADDRESS,
        YarnConfiguration.DEFAULT_RM_SCHEDULER_ADDRESS,
        YarnConfiguration.DEFAULT_RM_SCHEDULER_PORT);
  } else {
    String message = "Unsupported protocol found when creating the proxy " +
        "connection to ResourceManager: " +
        ((protocol != null) ? protocol.getClass().getName() : "null");
    LOG.error(message);
    throw new IllegalStateException(message);
  }
}
 

private void verifyGetClientAMToken(String submitter, String queueAdmin,
    String queueName, boolean setupACLs) throws Exception {
  ApplicationId applicationId =
      submitAppAndGetAppId(submitter, queueName, setupACLs);
  final GetApplicationReportRequest appReportRequest =
      GetApplicationReportRequest.newInstance(applicationId);

  ApplicationClientProtocol submitterClient = getRMClientForUser(submitter);
  ApplicationClientProtocol adMinUserClient = getRMClientForUser(queueAdmin);

  GetApplicationReportResponse submitterGetReport =
      submitterClient.getApplicationReport(appReportRequest);
  GetApplicationReportResponse adMinUserGetReport =
      adMinUserClient.getApplicationReport(appReportRequest);

  Assert.assertEquals(submitterGetReport.getApplicationReport()
    .getClientToAMToken(), adMinUserGetReport.getApplicationReport()
    .getClientToAMToken());
}
 

private void verifyKillAppFailure(String submitter, String killer,
    String queueName, boolean setupACLs) throws Exception {

  ApplicationId applicationId =
      submitAppAndGetAppId(submitter, queueName, setupACLs);

  final KillApplicationRequest finishAppRequest =
      KillApplicationRequest.newInstance(applicationId);

  ApplicationClientProtocol killerClient = getRMClientForUser(killer);

  // Kill app as the killer
  try {
    killerClient.forceKillApplication(finishAppRequest);
    Assert.fail("App killing by the enemy should fail!!");
  } catch (YarnException e) {
    LOG.info("Got exception while killing app as the enemy", e);
    Assert.assertTrue(e.getMessage().contains(
      "User " + killer + " cannot perform operation MODIFY_APP on "
          + applicationId));
  }

  getRMClientForUser(submitter).forceKillApplication(finishAppRequest);
}
 

private long renewDelegationToken(final UserGroupInformation loggedInUser,
    final ApplicationClientProtocol clientRMService,
    final org.apache.hadoop.yarn.api.records.Token dToken)
    throws IOException, InterruptedException {
  long nextExpTime = loggedInUser.doAs(new PrivilegedExceptionAction<Long>() {
    @Override
    public Long run() throws YarnException, IOException {
      RenewDelegationTokenRequest request = Records
          .newRecord(RenewDelegationTokenRequest.class);
      request.setDelegationToken(dToken);
      return clientRMService.renewDelegationToken(request)
          .getNextExpirationTime();
    }
  });
  return nextExpTime;
}
 

private ApplicationClientProtocol getClientRMProtocolWithDT(
    org.apache.hadoop.yarn.api.records.Token token,
    final InetSocketAddress rmAddress, String user, final Configuration conf) {
  // Maybe consider converting to Hadoop token, serialize de-serialize etc
  // before trying to renew the token.

  UserGroupInformation ugi = UserGroupInformation
      .createRemoteUser(user);
  ugi.addToken(ConverterUtils.convertFromYarn(token, rmAddress));

  final YarnRPC rpc = YarnRPC.create(conf);
  ApplicationClientProtocol clientRMWithDT = ugi
      .doAs(new PrivilegedAction<ApplicationClientProtocol>() {
        @Override
        public ApplicationClientProtocol run() {
          return (ApplicationClientProtocol) rpc.getProxy(ApplicationClientProtocol.class,
              rmAddress, conf);
        }
      });
  return clientRMWithDT;
}
 

@Override
protected void serviceStart() throws Exception {
  try {
    rmClient = ClientRMProxy.createRMProxy(getConfig(),
        ApplicationClientProtocol.class);
    if (historyServiceEnabled) {
      historyClient.start();
    }
    if (timelineServiceEnabled) {
      timelineClient.start();
    }
  } catch (IOException e) {
    throw new YarnRuntimeException(e);
  }
  super.serviceStart();
}
 

@SuppressWarnings("unchecked")
@Override
public long renew(Token<?> token, Configuration conf) throws IOException,
    InterruptedException {
  final ApplicationClientProtocol rmClient = getRmClient(token, conf);
  if (rmClient != null) {
    try {
      RenewDelegationTokenRequest request =
          Records.newRecord(RenewDelegationTokenRequest.class);
      request.setDelegationToken(convertToProtoToken(token));
      return rmClient.renewDelegationToken(request).getNextExpirationTime();
    } catch (YarnException e) {
      throw new IOException(e);
    } finally {
      RPC.stopProxy(rmClient);
    }
  } else {
    return localSecretManager.renewToken(
        (Token<RMDelegationTokenIdentifier>)token, getRenewer(token));
  }
}
 

@SuppressWarnings("unchecked")
@Override
public void cancel(Token<?> token, Configuration conf) throws IOException,
    InterruptedException {
  final ApplicationClientProtocol rmClient = getRmClient(token, conf);
  if (rmClient != null) {
    try {
      CancelDelegationTokenRequest request =
          Records.newRecord(CancelDelegationTokenRequest.class);
      request.setDelegationToken(convertToProtoToken(token));
      rmClient.cancelDelegationToken(request);
    } catch (YarnException e) {
      throw new IOException(e);
    } finally {
      RPC.stopProxy(rmClient);
    }
  } else {
    localSecretManager.cancelToken(
        (Token<RMDelegationTokenIdentifier>)token, getRenewer(token));
  }
}
 

private static ApplicationClientProtocol getRmClient(Token<?> token,
    Configuration conf) throws IOException {
  String[] services = token.getService().toString().split(",");
  for (String service : services) {
    InetSocketAddress addr = NetUtils.createSocketAddr(service);
    if (localSecretManager != null) {
      // return null if it's our token
      if (localServiceAddress.getAddress().isAnyLocalAddress()) {
        if (NetUtils.isLocalAddress(addr.getAddress()) &&
            addr.getPort() == localServiceAddress.getPort()) {
          return null;
        }
      } else if (addr.equals(localServiceAddress)) {
        return null;
      }
    }
  }
  return ClientRMProxy.createRMProxy(conf, ApplicationClientProtocol.class);
}
 

@Private
@Override
protected InetSocketAddress getRMAddress(YarnConfiguration conf,
    Class<?> protocol) throws IOException {
  if (protocol == ApplicationClientProtocol.class) {
    return conf.getSocketAddr(YarnConfiguration.RM_ADDRESS,
        YarnConfiguration.DEFAULT_RM_ADDRESS,
        YarnConfiguration.DEFAULT_RM_PORT);
  } else if (protocol == ResourceManagerAdministrationProtocol.class) {
    return conf.getSocketAddr(
        YarnConfiguration.RM_ADMIN_ADDRESS,
        YarnConfiguration.DEFAULT_RM_ADMIN_ADDRESS,
        YarnConfiguration.DEFAULT_RM_ADMIN_PORT);
  } else if (protocol == ApplicationMasterProtocol.class) {
    setAMRMTokenService(conf);
    return conf.getSocketAddr(YarnConfiguration.RM_SCHEDULER_ADDRESS,
        YarnConfiguration.DEFAULT_RM_SCHEDULER_ADDRESS,
        YarnConfiguration.DEFAULT_RM_SCHEDULER_PORT);
  } else {
    String message = "Unsupported protocol found when creating the proxy " +
        "connection to ResourceManager: " +
        ((protocol != null) ? protocol.getClass().getName() : "null");
    LOG.error(message);
    throw new IllegalStateException(message);
  }
}
 

private void verifyGetClientAMToken(String submitter, String queueAdmin,
    String queueName, boolean setupACLs) throws Exception {
  ApplicationId applicationId =
      submitAppAndGetAppId(submitter, queueName, setupACLs);
  final GetApplicationReportRequest appReportRequest =
      GetApplicationReportRequest.newInstance(applicationId);

  ApplicationClientProtocol submitterClient = getRMClientForUser(submitter);
  ApplicationClientProtocol adMinUserClient = getRMClientForUser(queueAdmin);

  GetApplicationReportResponse submitterGetReport =
      submitterClient.getApplicationReport(appReportRequest);
  GetApplicationReportResponse adMinUserGetReport =
      adMinUserClient.getApplicationReport(appReportRequest);

  Assert.assertEquals(submitterGetReport.getApplicationReport()
    .getClientToAMToken(), adMinUserGetReport.getApplicationReport()
    .getClientToAMToken());
}
 

private void verifyKillAppFailure(String submitter, String killer,
    String queueName, boolean setupACLs) throws Exception {

  ApplicationId applicationId =
      submitAppAndGetAppId(submitter, queueName, setupACLs);

  final KillApplicationRequest finishAppRequest =
      KillApplicationRequest.newInstance(applicationId);

  ApplicationClientProtocol killerClient = getRMClientForUser(killer);

  // Kill app as the killer
  try {
    killerClient.forceKillApplication(finishAppRequest);
    Assert.fail("App killing by the enemy should fail!!");
  } catch (YarnException e) {
    LOG.info("Got exception while killing app as the enemy", e);
    Assert.assertTrue(e.getMessage().contains(
      "User " + killer + " cannot perform operation MODIFY_APP on "
          + applicationId));
  }

  getRMClientForUser(submitter).forceKillApplication(finishAppRequest);
}
 

private org.apache.hadoop.yarn.api.records.Token getDelegationToken(
    final UserGroupInformation loggedInUser,
    final ApplicationClientProtocol clientRMService, final String renewerString)
    throws IOException, InterruptedException {
  org.apache.hadoop.yarn.api.records.Token token = loggedInUser
      .doAs(new PrivilegedExceptionAction<org.apache.hadoop.yarn.api.records.Token>() {
        @Override
          public org.apache.hadoop.yarn.api.records.Token run()
              throws YarnException, IOException {
          GetDelegationTokenRequest request = Records
              .newRecord(GetDelegationTokenRequest.class);
          request.setRenewer(renewerString);
          return clientRMService.getDelegationToken(request)
              .getRMDelegationToken();
        }
      });
  return token;
}
 

private long renewDelegationToken(final UserGroupInformation loggedInUser,
    final ApplicationClientProtocol clientRMService,
    final org.apache.hadoop.yarn.api.records.Token dToken)
    throws IOException, InterruptedException {
  long nextExpTime = loggedInUser.doAs(new PrivilegedExceptionAction<Long>() {
    @Override
    public Long run() throws YarnException, IOException {
      RenewDelegationTokenRequest request = Records
          .newRecord(RenewDelegationTokenRequest.class);
      request.setDelegationToken(dToken);
      return clientRMService.renewDelegationToken(request)
          .getNextExpirationTime();
    }
  });
  return nextExpTime;
}
 

private ApplicationClientProtocol getClientRMProtocolWithDT(
    org.apache.hadoop.yarn.api.records.Token token,
    final InetSocketAddress rmAddress, String user, final Configuration conf) {
  // Maybe consider converting to Hadoop token, serialize de-serialize etc
  // before trying to renew the token.

  UserGroupInformation ugi = UserGroupInformation
      .createRemoteUser(user);
  ugi.addToken(ConverterUtils.convertFromYarn(token, rmAddress));

  final YarnRPC rpc = YarnRPC.create(conf);
  ApplicationClientProtocol clientRMWithDT = ugi
      .doAs(new PrivilegedAction<ApplicationClientProtocol>() {
        @Override
        public ApplicationClientProtocol run() {
          return (ApplicationClientProtocol) rpc.getProxy(ApplicationClientProtocol.class,
              rmAddress, conf);
        }
      });
  return clientRMWithDT;
}
 

@Override
protected void serviceStart() throws Exception {
  try {
    rmClient = ClientRMProxy.createRMProxy(getConfig(),
        ApplicationClientProtocol.class);
    if (historyServiceEnabled) {
      historyClient.start();
    }
    if (timelineServiceEnabled) {
      timelineClient.start();
    }
  } catch (IOException e) {
    throw new YarnRuntimeException(e);
  }
  super.serviceStart();
}
 

/**
 * Connect to the Resource Manager/Applications Manager<p>
 *
 * @return Handle to communicate with the ASM
 * @throws IOException
 */
public ApplicationClientProtocol connectToASM() throws IOException
{
  YarnConfiguration yarnConf = new YarnConfiguration(conf);
  InetSocketAddress rmAddress = yarnConf.getSocketAddr(
      YarnConfiguration.RM_ADDRESS,
      YarnConfiguration.DEFAULT_RM_ADDRESS,
      YarnConfiguration.DEFAULT_RM_PORT);
  LOG.debug("Connecting to ResourceManager at " + rmAddress);
  return ((ApplicationClientProtocol)rpc.getProxy(ApplicationClientProtocol.class, rmAddress, conf));
}
 

public LogDeletionTask(Configuration conf, long retentionSecs, ApplicationClientProtocol rmClient) {
  this.conf = conf;
  this.retentionMillis = retentionSecs * 1000;
  this.suffix = LogAggregationUtils.getRemoteNodeLogDirSuffix(conf);
  this.remoteRootLogDir =
    new Path(conf.get(YarnConfiguration.NM_REMOTE_APP_LOG_DIR,
        YarnConfiguration.DEFAULT_NM_REMOTE_APP_LOG_DIR));
  this.rmClient = rmClient;
}
 

/**
 * Create a new Connection to the RM to fetch Application reports.
 * @param conf the conf to use to know where the RM is.
 */
public AppReportFetcher(Configuration conf) {
  this.conf = conf;
  try {
    applicationsManager = ClientRMProxy.createRMProxy(conf,
        ApplicationClientProtocol.class);
  } catch (IOException e) {
    throw new YarnRuntimeException(e);
  }
}
 

@Override
protected void serviceStart() throws Exception {
  Configuration conf = getConfig();
  YarnRPC rpc = YarnRPC.create(conf);
  this.server =   
    rpc.getServer(ApplicationClientProtocol.class, this,
          clientBindAddress,
          conf, this.rmDTSecretManager,
          conf.getInt(YarnConfiguration.RM_CLIENT_THREAD_COUNT, 
              YarnConfiguration.DEFAULT_RM_CLIENT_THREAD_COUNT));
  
  // Enable service authorization?
  if (conf.getBoolean(
      CommonConfigurationKeysPublic.HADOOP_SECURITY_AUTHORIZATION, 
      false)) {
    InputStream inputStream =
        this.rmContext.getConfigurationProvider()
            .getConfigurationInputStream(conf,
                YarnConfiguration.HADOOP_POLICY_CONFIGURATION_FILE);
    if (inputStream != null) {
      conf.addResource(inputStream);
    }
    refreshServiceAcls(conf, RMPolicyProvider.getInstance());
  }
  
  this.server.start();
  clientBindAddress = conf.updateConnectAddr(YarnConfiguration.RM_BIND_HOST,
                                             YarnConfiguration.RM_ADDRESS,
                                             YarnConfiguration.DEFAULT_RM_ADDRESS,
                                             server.getListenerAddress());
  super.serviceStart();
}
 

private void verifyKillAppSuccess(String submitter, String killer,
    String queueName, boolean setupACLs) throws Exception {
  ApplicationId applicationId =
      submitAppAndGetAppId(submitter, queueName, setupACLs);

  final KillApplicationRequest finishAppRequest =
      KillApplicationRequest.newInstance(applicationId);

  ApplicationClientProtocol ownerClient = getRMClientForUser(killer);

  // Kill app as killer
  ownerClient.forceKillApplication(finishAppRequest);
  resourceManager.waitForState(applicationId, RMAppState.KILLED);
}
 

private ApplicationId submitAppAndGetAppId(String submitter,
    String queueName, boolean setupACLs) throws Exception {

  GetNewApplicationRequest newAppRequest =
      GetNewApplicationRequest.newInstance();

  ApplicationClientProtocol submitterClient = getRMClientForUser(submitter);
  ApplicationId applicationId =
      submitterClient.getNewApplication(newAppRequest).getApplicationId();

  Resource resource = BuilderUtils.newResource(1024, 1);
  Map<ApplicationAccessType, String> acls = createACLs(submitter, setupACLs);
  ContainerLaunchContext amContainerSpec =
      ContainerLaunchContext.newInstance(null, null, null, null, null, acls);

  ApplicationSubmissionContext appSubmissionContext =
      ApplicationSubmissionContext.newInstance(applicationId,
        "applicationName", queueName, null, amContainerSpec, false, true, 1,
        resource, "applicationType");
  appSubmissionContext.setApplicationId(applicationId);
  appSubmissionContext.setQueue(queueName);

  SubmitApplicationRequest submitRequest =
      SubmitApplicationRequest.newInstance(appSubmissionContext);
  submitterClient.submitApplication(submitRequest);
  resourceManager.waitForState(applicationId, RMAppState.ACCEPTED);
  return applicationId;
}
 

private ApplicationClientProtocol getRMClientForUser(String user)
    throws IOException, InterruptedException {
  UserGroupInformation userUGI = UserGroupInformation.createRemoteUser(user);
  ApplicationClientProtocol userClient =
      userUGI
        .doAs(new PrivilegedExceptionAction<ApplicationClientProtocol>() {
          @Override
          public ApplicationClientProtocol run() throws Exception {
            return (ApplicationClientProtocol) rpc.getProxy(
              ApplicationClientProtocol.class, rmAddress, conf);
          }
        });
  return userClient;
}
 

@Test
public void tesAllJobs() throws Exception {
  final ApplicationClientProtocol applicationsManager = Mockito.mock(ApplicationClientProtocol.class);
  GetApplicationsResponse allApplicationsResponse = Records
      .newRecord(GetApplicationsResponse.class);
  List<ApplicationReport> applications = new ArrayList<ApplicationReport>();
  applications.add(getApplicationReport(YarnApplicationState.FINISHED,
      FinalApplicationStatus.FAILED));
  applications.add(getApplicationReport(YarnApplicationState.FINISHED,
      FinalApplicationStatus.SUCCEEDED));
  applications.add(getApplicationReport(YarnApplicationState.FINISHED,
      FinalApplicationStatus.KILLED));
  applications.add(getApplicationReport(YarnApplicationState.FAILED,
      FinalApplicationStatus.FAILED));
  allApplicationsResponse.setApplicationList(applications);
  Mockito.when(
      applicationsManager.getApplications(Mockito
          .any(GetApplicationsRequest.class))).thenReturn(
      allApplicationsResponse);
  ResourceMgrDelegate resourceMgrDelegate = new ResourceMgrDelegate(
    new YarnConfiguration()) {
    @Override
    protected void serviceStart() throws Exception {
      Assert.assertTrue(this.client instanceof YarnClientImpl);
      ((YarnClientImpl) this.client).setRMClient(applicationsManager);
    }
  };
  JobStatus[] allJobs = resourceMgrDelegate.getAllJobs();

  Assert.assertEquals(State.FAILED, allJobs[0].getState());
  Assert.assertEquals(State.SUCCEEDED, allJobs[1].getState());
  Assert.assertEquals(State.KILLED, allJobs[2].getState());
  Assert.assertEquals(State.FAILED, allJobs[3].getState());
}
 

private void cancelDelegationToken(final UserGroupInformation loggedInUser,
    final ApplicationClientProtocol clientRMService,
    final org.apache.hadoop.yarn.api.records.Token dToken)
    throws IOException, InterruptedException {
  loggedInUser.doAs(new PrivilegedExceptionAction<Void>() {
    @Override
    public Void run() throws YarnException, IOException {
      CancelDelegationTokenRequest request = Records
          .newRecord(CancelDelegationTokenRequest.class);
      request.setDelegationToken(dToken);
      clientRMService.cancelDelegationToken(request);
      return null;
    }
  });
}
 

private ApplicationClientProtocol getRMClientForUser(String user)
    throws IOException, InterruptedException {
  UserGroupInformation userUGI = UserGroupInformation
      .createRemoteUser(user);
  ApplicationClientProtocol userClient = userUGI
      .doAs(new PrivilegedExceptionAction<ApplicationClientProtocol>() {
        @Override
        public ApplicationClientProtocol run() throws Exception {
          return (ApplicationClientProtocol) rpc.getProxy(ApplicationClientProtocol.class,
              rmAddress, conf);
        }
      });
  return userClient;
}
 

private void verifySuperUserAccess() throws Exception {

    AccessControlList viewACL = new AccessControlList("");
    viewACL.addGroup(FRIENDLY_GROUP);
    AccessControlList modifyACL = new AccessControlList("");
    modifyACL.addUser(FRIEND);
    ApplicationId applicationId = submitAppAndGetAppId(viewACL, modifyACL);

    final GetApplicationReportRequest appReportRequest = recordFactory
        .newRecordInstance(GetApplicationReportRequest.class);
    appReportRequest.setApplicationId(applicationId);
    final KillApplicationRequest finishAppRequest = recordFactory
        .newRecordInstance(KillApplicationRequest.class);
    finishAppRequest.setApplicationId(applicationId);

    ApplicationClientProtocol superUserClient = getRMClientForUser(SUPER_USER);

    // View as the superUser
    superUserClient.getApplicationReport(appReportRequest);

    // List apps as superUser
    Assert.assertEquals("App view by super-user should list the apps!!", 2,
        superUserClient.getApplications(
            recordFactory.newRecordInstance(GetApplicationsRequest.class))
            .getApplicationList().size());

    // Kill app as the superUser
    superUserClient.forceKillApplication(finishAppRequest);
    resourceManager.waitForState(applicationId, RMAppState.KILLED);
  }