@Test
public void testFileSetDelete() throws Exception {

  List<String> resources = Lists.newArrayList();
  // This version is currently serving.
  resources.add(getResource(FILESET, 100));
  // This version if offlined but the helix resource and HDFS data is still there.
  resources.add(getResource(FILESET, 7400));
  when(mockHelixAdmin.getResourcesInCluster(CLUSTER_NAME)).thenReturn(resources);
  FileSetInfo fsInfo = new FileSetInfo(FILESET,
      TerrapinUtil.helixResourceToHdfsDir(resources.get(0)),
      2,
      (List)Lists.newArrayList(),
      new Options());
  fsInfo.deleted = true;
  when(mockZkManager.getCandidateHdfsDirMap()).thenReturn((Map)ImmutableMap.of(
      FILESET, new ImmutablePair(fsInfo, null)));
  setupBaseDirListing(Lists.newArrayList(FILESET));
  setupListingForFileSet(FILESET, resources, 2);
  String servingResource = resources.get(0);
  String oldResource = resources.get(1);
  testRoutingTableProvider.setOnlinePartitionMap((Map)ImmutableMap.of(
      servingResource, ImmutableMap.of(
          TerrapinUtil.getViewPartitionName(servingResource, 0),
          ImmutableList.of(hostConfig1, hostConfig2)),
          TerrapinUtil.getViewPartitionName(servingResource, 1),
          ImmutableList.of(hostConfig1, hostConfig3)));
  HdfsManager.Rebalancer rebalancer = hdfsManager.createAndGetRebalancer();
  rebalancer.reconcileAndRebalance();
  // In the first rebalance, the serving resource will get offlined while the older
  // resource will be deleted (since its already offlined.
  checkIdealStateModified(servingResource, (Map) Maps.newHashMap(), 2);
  checkIdealStateModified(servingResource, (Map) Maps.newHashMap(), 2);
  checkHdfsBlocksNotRetrieved(servingResource, 2);
  checkHdfsDataNotDeleted(servingResource);
  checkViewInfoNotDeleted(servingResource);
  checkResourceNotDeleted(servingResource);
  checkIdealStateNotModified(oldResource);
  checkHdfsBlocksNotRetrieved(oldResource, 2);
  checkHdfsDataDeleted(oldResource);
  checkViewInfoDeleted(oldResource);
  checkResourceDeleted(oldResource);
  // Make sure that the fileset is not deleted yet.
  verify(mockDfsClient, times(0)).delete(eq(Constants.HDFS_DATA_DIR + "/" + FILESET), eq(true));
  verify(mockZkManager, times(0)).deleteFileSetInfo(eq(FILESET));
  verify(mockZkManager, times(0)).unlockFileSet(eq(FILESET));
  // In the 2nd rebalance, the fileset will be deleted.
  // Change HDFS listing & helix listing since one of the resources was deleted.
  when(mockHelixAdmin.getResourcesInCluster(eq(CLUSTER_NAME))).thenReturn(
      resources.subList(0, 1));
  setupListingForFileSet(FILESET, resources.subList(0, 1), 2);
  // The serving resource has been offlined.
  testRoutingTableProvider.setOnlinePartitionMap((Map)Maps.newHashMap());
  rebalancer.reconcileAndRebalance();
  // Note that this check means that the ideal state was NOT modified since these
  // mock calls from the previous reconcileAndRebalance call.
  checkIdealStateModified(servingResource, (Map) Maps.newHashMap(), 2);
  checkHdfsBlocksNotRetrieved(servingResource, 2);
  checkViewInfoDeleted(servingResource);
  checkResourceDeleted(servingResource);
  checkHdfsDataDeleted(servingResource);
  verify(mockDfsClient, times(1)).delete(eq(Constants.HDFS_DATA_DIR + "/" + FILESET), eq(true));
  verify(mockZkManager, times(1)).deleteFileSetInfo(eq(FILESET));
  verify(mockZkManager, times(1)).unlockFileSet(eq(FILESET));
}