### Question:
TaskService { public void completeJob(ClusterJob job, Cluster cluster) throws IOException, IllegalAccessException { job.setJobStatus(ClusterJob.Status.COMPLETE); clusterStore.writeClusterJob(job); LOG.debug("Job {} is complete", job.getJobId()); if (job.getClusterAction() == ClusterAction.CLUSTER_DELETE) { cluster.setStatus(Cluster.Status.TERMINATED); } else { cluster.setStatus(Cluster.Status.ACTIVE); } clusterStore.writeCluster(cluster); serverStats.getSuccessfulClusterStats().incrementStat(job.getClusterAction()); if (job.getClusterAction() == ClusterAction.CLUSTER_DELETE) { wipeSensitiveFields(cluster); } callbackQueues.add(cluster.getAccount().getTenantId(), new Element(gson.toJson(new CallbackData(CallbackData.Type.SUCCESS, cluster, job)))); } @Inject private  TaskService(ClusterStoreService clusterStoreService,
                      CredentialStore credentialStore,
                      ServerStats serverStats,
                      QueueService queueService,
                      IdService idService,
                      Gson gson); List<ClusterTask> getRetryTask(ClusterTask task); void failJobAndSetClusterStatus(ClusterJob job, Cluster cluster, Cluster.Status status, String message); void failJobAndSetClusterStatus(ClusterJob job, Cluster cluster); void failJobAndTerminateCluster(ClusterJob job, Cluster cluster, String message); void failJob(ClusterJob job); void startJob(ClusterJob job, Cluster cluster); void completeJob(ClusterJob job, Cluster cluster); void startTask(ClusterTask clusterTask); void dropTask(ClusterTask clusterTask); void failTask(ClusterTask clusterTask, int status); void completeTask(ClusterTask clusterTask, int status);  }### Answer:
@Test public void testOnlyDeleteFinishWipesCredentials() throws Exception { Account account = new Account("user", "tenant"); String clusterId = "123"; Cluster cluster = Cluster.builder() .setName("cluster1") .setID(clusterId) .setProvider(Entities.ProviderExample.RACKSPACE) .setClusterTemplate(Entities.ClusterTemplateExample.HDFS) .setAccount(account) .setStatus(Cluster.Status.ACTIVE) .build(); Map<String, Object> sensitiveFields = Maps.newHashMap(); sensitiveFields.put("key", "keycontents"); credentialStore.set(account.getTenantId(), clusterId, sensitiveFields); long jobNum = 1; for (ClusterAction action : ClusterAction.values()) { if (action == ClusterAction.CLUSTER_DELETE) { continue; } JobId jobId = new JobId(clusterId, jobNum); ClusterJob job = new ClusterJob(jobId, action); cluster.setLatestJobId(jobId.getId()); taskService.completeJob(job, cluster); jobNum++; Assert.assertEquals(sensitiveFields, credentialStore.get(account.getTenantId(), clusterId)); } JobId jobId = new JobId(clusterId, jobNum); ClusterJob job = new ClusterJob(jobId, ClusterAction.CLUSTER_DELETE); cluster.setLatestJobId(jobId.getId()); taskService.completeJob(job, cluster); Assert.assertTrue(credentialStore.get(account.getTenantId(), clusterId).isEmpty()); }