DelegationTokenRenewer extends AbstractService { @VisibleForTesting protected Token<?>[] obtainSystemTokensForUser(String user, final Credentials credentials) throws IOException, InterruptedException { UserGroupInformation proxyUser = UserGroupInformation.createProxyUser(user, UserGroupInformation.getLoginUser()); Token<?>[] newTokens = proxyUser.doAs(new PrivilegedExceptionAction<Token<?>[]>() { @Override public Token<?>[] run() throws Exception { FileSystem fs = FileSystem.get(getConfig()); try { return fs.addDelegationTokens( UserGroupInformation.getLoginUser().getUserName(), credentials); } finally { fs.close(); } } }); return newTokens; } DelegationTokenRenewer(); @VisibleForTesting Set<Token<?>> getDelegationTokens(); void addApplicationAsync(ApplicationId applicationId, Credentials ts,
      boolean shouldCancelAtEnd, String user); void addApplicationAsyncDuringRecovery(ApplicationId applicationId,
      Credentials ts, boolean shouldCancelAtEnd, String user); void addApplicationSync(ApplicationId applicationId, Credentials ts,
      boolean shouldCancelAtEnd, String user); void applicationFinished(ApplicationId applicationId); void updateKeepAliveApplications(List<ApplicationId> appIds); void setRMContext(RMContext rmContext); static final String SCHEME; static final String RM_SYSTEM_CREDENTIALS_VALID_TIME_REMAINING; static final long DEFAULT_RM_SYSTEM_CREDENTIALS_VALID_TIME_REMAINING; }
@Test public void testFSLeakInObtainSystemTokensForUser() throws Exception{ Credentials credentials = new Credentials(); String user = "test"; int oldCounter = MyFS.getInstanceCounter(); delegationTokenRenewer.obtainSystemTokensForUser(user, credentials); delegationTokenRenewer.obtainSystemTokensForUser(user, credentials); delegationTokenRenewer.obtainSystemTokensForUser(user, credentials); Assert.assertEquals(oldCounter, MyFS.getInstanceCounter()); }