DelegationTokenRenewer extends AbstractService { protected ConcurrentMap<Token<?>, DelegationTokenToRenew> getAllTokens() { return allTokens; } DelegationTokenRenewer(); @VisibleForTesting Set<Token<?>> getDelegationTokens(); void addApplicationAsync(ApplicationId applicationId, Credentials ts,
      boolean shouldCancelAtEnd, String user, Configuration tokenConf); void addApplicationAsyncDuringRecovery(ApplicationId applicationId,
      Credentials ts, boolean shouldCancelAtEnd, String user,
      Configuration tokenConf); void addApplicationSync(ApplicationId applicationId, Credentials ts,
      boolean shouldCancelAtEnd, String user); void applicationFinished(ApplicationId applicationId); void updateKeepAliveApplications(List<ApplicationId> appIds); void setRMContext(RMContext rmContext); @VisibleForTesting void setDelegationTokenRenewerPoolTracker(boolean flag); @VisibleForTesting
  static final Text HDFS_DELEGATION_KIND; static final String SCHEME; static final String RM_SYSTEM_CREDENTIALS_VALID_TIME_REMAINING; static final long DEFAULT_RM_SYSTEM_CREDENTIALS_VALID_TIME_REMAINING; }
@Test (timeout = 30000) public void testAppSubmissionWithPreviousToken() throws Exception{ MockRM rm = new TestSecurityMockRM(conf, null); rm.start(); final MockNM nm1 = new MockNM("127.0.0.1:1234", 15120, rm.getResourceTrackerService()); nm1.registerNode(); Text userText1 = new Text("user"); DelegationTokenIdentifier dtId1 = new DelegationTokenIdentifier(userText1, new Text("renewer1"), userText1); final Token<DelegationTokenIdentifier> token1 = new Token<DelegationTokenIdentifier>(dtId1.getBytes(), "password1".getBytes(), dtId1.getKind(), new Text("service1")); Credentials credentials = new Credentials(); credentials.addToken(userText1, token1); Resource resource = Records.newRecord(Resource.class); resource.setMemorySize(200); MockRMAppSubmissionData data1 = MockRMAppSubmissionData.Builder .createWithResource(resource, rm) .withAppName("name") .withUser("user") .withMaxAppAttempts(2) .withCredentials(credentials) .withCancelTokensWhenComplete(false) .build(); RMApp app1 = MockRMAppSubmitter.submit(rm, data1); MockAM am1 = MockRM.launchAndRegisterAM(app1, rm, nm1); rm.waitForState(app1.getApplicationId(), RMAppState.RUNNING); DelegationTokenRenewer renewer = rm.getRMContext().getDelegationTokenRenewer(); DelegationTokenToRenew dttr = renewer.getAllTokens().get(token1); Assert.assertNotNull(dttr); MockRMAppSubmissionData data = MockRMAppSubmissionData.Builder .createWithResource(resource, rm) .withAppName("name") .withUser("user") .withMaxAppAttempts(2) .withCredentials(credentials) .withCancelTokensWhenComplete(true) .build(); RMApp app2 = MockRMAppSubmitter.submit(rm, data); MockAM am2 = MockRM.launchAndRegisterAM(app2, rm, nm1); rm.waitForState(app2.getApplicationId(), RMAppState.RUNNING); finishAMAndWaitForComplete(app2, rm, nm1, am2, dttr); Assert.assertTrue(rm.getRMContext().getDelegationTokenRenewer() .getAllTokens().containsKey(token1)); finishAMAndWaitForComplete(app1, rm, nm1, am1, dttr); Assert.assertFalse(Renewer.cancelled); }
@Test public void testRenewTokenUsingTokenConfProvidedByApp() throws Exception{ conf.set(CommonConfigurationKeysPublic.HADOOP_SECURITY_AUTHENTICATION, "kerberos"); UserGroupInformation.setConfiguration(conf); MockRM rm = new TestSecurityMockRM(conf, null); rm.start(); final MockNM nm1 = new MockNM("127.0.0.1:1234", 15120, rm.getResourceTrackerService()); nm1.registerNode(); Text userText1 = new Text("user1"); DelegationTokenIdentifier dtId1 = new DelegationTokenIdentifier(userText1, new Text("renewer1"), userText1); final Token<DelegationTokenIdentifier> token1 = new Token<DelegationTokenIdentifier>(dtId1.getBytes(), "password1".getBytes(), dtId1.getKind(), new Text("service1")); Credentials credentials = new Credentials(); credentials.addToken(userText1, token1); Configuration appConf = new Configuration(false); appConf.set("dfs.nameservices", "mycluster1,mycluster2"); appConf.set("dfs.namenode.rpc-address.mycluster2.nn1", "123.0.0.1"); appConf.set("dfs.namenode.rpc-address.mycluster2.nn2", "123.0.0.2"); appConf.set("dfs.ha.namenodes.mycluster2", "nn1,nn2"); appConf.set("dfs.client.failover.proxy.provider.mycluster2", "provider"); DataOutputBuffer dob = new DataOutputBuffer(); appConf.write(dob); ByteBuffer tokenConf = ByteBuffer.wrap(dob.getData(), 0, dob.getLength()); final int confSize = appConf.size(); RMApp app = submitApp(rm, credentials, tokenConf); GenericTestUtils.waitFor(new Supplier<Boolean>() { public Boolean get() { DelegationTokenToRenew toRenew = rm.getRMContext().getDelegationTokenRenewer().getAllTokens() .get(token1); return toRenew != null && toRenew.conf != null && toRenew.conf.size() == confSize && toRenew.conf .get("dfs.namenode.rpc-address.mycluster2.nn1").equals("123.0.0.1"); } }, 200, 10000); }