### Question:
SparkCredentialsUpdater extends AbstractIdleService implements Runnable { @Override public void run() { long nextUpdateTime = updateIntervalMs; try { if (generation == 0) { generation = findLatestGeneration(); } generation++; Location credentialsFile = credentialsDir.append(fileNamePrefix + SPARK_YARN_CREDS_COUNTER_DELIM + generation); Location tempFile = credentialsDir.append(credentialsFile.getName() + SPARK_YARN_CREDS_TEMP_EXTENSION); Credentials credentials = credentialsSupplier.get(); try (DataOutputStream os = new DataOutputStream(tempFile.getOutputStream("600"))) { credentials.writeTokenStorageToStream(os); } if (!credentialsFile.equals(tempFile.renameTo(credentialsFile))) { throw new IOException("Failed to rename from " + tempFile + " to " + credentialsFile); } LOG.debug("Credentials written to {}", credentialsFile); nextUpdateTime = getNextUpdateDelay(credentials); LOG.debug("Next credentials refresh at {}ms later", nextUpdateTime); scheduler.schedule(this, nextUpdateTime, TimeUnit.MILLISECONDS); cleanup(); } catch (Exception e) { long retryDelay = Math.min(60000, nextUpdateTime); LOG.warn("Exception raised when saving credentials. Retry in {}ms", retryDelay, e); scheduler.schedule(this, retryDelay, TimeUnit.MILLISECONDS); } } SparkCredentialsUpdater(Supplier<Credentials> credentialsSupplier,
                                 Location credentialsDir, String fileNamePrefix, long updateIntervalMs,
                                 long cleanupExpireMs, int minFilesToKeep); @Override void run();  }### Answer:
@Test public void testUpdater() throws Exception { Location credentialsDir = Locations.toLocation(TEMPORARY_FOLDER.newFolder()); SparkCredentialsUpdater updater = new SparkCredentialsUpdater(createCredentialsSupplier(), credentialsDir, "credentials", TimeUnit.DAYS.toMillis(1), TimeUnit.DAYS.toMillis(1), Integer.MAX_VALUE) { @Override long getNextUpdateDelay(Credentials credentials) throws IOException { return TimeUnit.DAYS.toMillis(1); } }; Assert.assertTrue(credentialsDir.list().isEmpty()); UserGroupInformation.getCurrentUser().addToken(new Token<>(Bytes.toBytes("id"), Bytes.toBytes("pass"), new Text("kind"), new Text("service"))); updater.startAndWait(); try { List<Location> expectedFiles = new ArrayList<>(); expectedFiles.add(credentialsDir.append("credentials-1")); for (int i = 1; i <= 10; i++) { Assert.assertEquals(expectedFiles, listAndSort(credentialsDir)); Credentials newCredentials = new Credentials(); try (DataInputStream is = new DataInputStream(expectedFiles.get(expectedFiles.size() - 1).getInputStream())) { newCredentials.readTokenStorageStream(is); } Credentials userCredentials = UserGroupInformation.getCurrentUser().getCredentials(); for (Token<? extends TokenIdentifier> token : userCredentials.getAllTokens()) { Assert.assertEquals(token, newCredentials.getToken(token.getService())); } UserGroupInformation.getCurrentUser().addToken(new Token<>(Bytes.toBytes("id" + i), Bytes.toBytes("pass" + i), new Text("kind" + i), new Text("service" + i))); updater.run(); expectedFiles.add(credentialsDir.append("credentials-" + (i + 1))); } } finally { updater.stopAndWait(); } }