@SuppressWarnings("unchecked")
@Test
public void testTokenExpiry() throws Exception {
MyContainerManager containerManager = new MyContainerManager();
final MockRMWithAMS rm =
new MockRMWithAMS(conf, containerManager);
rm.start();
final Configuration conf = rm.getConfig();
final YarnRPC rpc = YarnRPC.create(conf);
ApplicationMasterProtocol rmClient = null;
try {
MockNM nm1 = rm.registerNode("localhost:1234", 5120);
RMApp app = rm.submitApp(1024);
nm1.nodeHeartbeat(true);
int waitCount = 0;
while (containerManager.containerTokens == null && waitCount++ < 20) {
LOG.info("Waiting for AM Launch to happen..");
Thread.sleep(1000);
}
Assert.assertNotNull(containerManager.containerTokens);
RMAppAttempt attempt = app.getCurrentAppAttempt();
ApplicationAttemptId applicationAttemptId = attempt.getAppAttemptId();
// Create a client to the RM.
UserGroupInformation currentUser =
UserGroupInformation
.createRemoteUser(applicationAttemptId.toString());
Credentials credentials = containerManager.getContainerCredentials();
final InetSocketAddress rmBindAddress =
rm.getApplicationMasterService().getBindAddress();
Token<? extends TokenIdentifier> amRMToken =
MockRMWithAMS.setupAndReturnAMRMToken(rmBindAddress,
credentials.getAllTokens());
currentUser.addToken(amRMToken);
rmClient = createRMClient(rm, conf, rpc, currentUser);
RegisterApplicationMasterRequest request =
Records.newRecord(RegisterApplicationMasterRequest.class);
rmClient.registerApplicationMaster(request);
FinishApplicationMasterRequest finishAMRequest =
Records.newRecord(FinishApplicationMasterRequest.class);
finishAMRequest
.setFinalApplicationStatus(FinalApplicationStatus.SUCCEEDED);
finishAMRequest.setDiagnostics("diagnostics");
finishAMRequest.setTrackingUrl("url");
rmClient.finishApplicationMaster(finishAMRequest);
// Send RMAppAttemptEventType.CONTAINER_FINISHED to transit RMAppAttempt
// from Finishing state to Finished State. Both AMRMToken and
// ClientToAMToken will be removed.
ContainerStatus containerStatus =
BuilderUtils.newContainerStatus(attempt.getMasterContainer().getId(),
ContainerState.COMPLETE,
"AM Container Finished", 0);
rm.getRMContext()
.getDispatcher()
.getEventHandler()
.handle(
new RMAppAttemptContainerFinishedEvent(applicationAttemptId,
containerStatus));
// Make sure the RMAppAttempt is at Finished State.
// Both AMRMToken and ClientToAMToken have been removed.
int count = 0;
while (attempt.getState() != RMAppAttemptState.FINISHED
&& count < maxWaitAttempts) {
Thread.sleep(100);
count++;
}
Assert.assertTrue(attempt.getState() == RMAppAttemptState.FINISHED);
// Now simulate trying to allocate. RPC call itself should throw auth
// exception.
rpc.stopProxy(rmClient, conf); // To avoid using cached client
rmClient = createRMClient(rm, conf, rpc, currentUser);
AllocateRequest allocateRequest =
Records.newRecord(AllocateRequest.class);
try {
rmClient.allocate(allocateRequest);
Assert.fail("You got to be kidding me! "
+ "Using App tokens after app-finish should fail!");
} catch (Throwable t) {
LOG.info("Exception found is ", t);
// The exception will still have the earlier appAttemptId as it picks it
// up from the token.
Assert.assertTrue(t.getCause().getMessage().contains(
"Password not found for ApplicationAttempt " +
applicationAttemptId.toString()));
}
} finally {
rm.stop();
if (rmClient != null) {
rpc.stopProxy(rmClient, conf); // To avoid using cached client
}
}
}