@Test
@PrepareForTest({ControllerUtil.class})

public void testLoadFileSet() throws Exception {
  PowerMockito.mockStatic(ControllerUtil.class);
  TerrapinLoadRequest request =
      new TerrapinLoadRequest("fileset", "/terrapin/data/fileset/12345564534", 200);
  Options requestOptions = new Options();
  requestOptions.setNumVersionsToKeep(2);
  request.setOptions(requestOptions);
  FileSetInfo fileSetInfo = new FileSetInfo(
      request.getFileSet(), "/terrapin/data/fileset/12345563434",
      request.getExpectedNumPartitions(),
      Lists.newArrayList(mock(FileSetInfo.ServingInfo.class)), new Options()
  );
  String resourceName = TerrapinUtil.hdfsDirToHelixResource(request.getHdfsDirectory());
  CustomModeISBuilder idealStateBuilder = new CustomModeISBuilder(resourceName);
  idealStateBuilder.assignInstanceAndState(resourceName + "$0", "host0", "ONLINE");
  idealStateBuilder.assignInstanceAndState(resourceName + "$1", "host1", "ONLINE");
  idealStateBuilder.setStateModel("OnlineOffline");
  idealStateBuilder.setNumReplica(2);
  idealStateBuilder.setNumPartitions(request.getExpectedNumPartitions());
  IdealState is = idealStateBuilder.build();
  is.setBucketSize(2);
  is.setRebalanceMode(IdealState.RebalanceMode.CUSTOMIZED);
  ViewInfo viewInfo1 = mock(ViewInfo.class);
  ViewInfo viewInfo2 = mock(ViewInfo.class);
  when(configuration.getInt(eq(Constants.NUM_SERVING_REPLICAS), eq(3))).thenReturn(3);
  when(zkManager.getFileSetInfo(eq(request.getFileSet()))).thenReturn(fileSetInfo);
  when(zkManager.getViewInfo(eq(resourceName))).thenReturn(viewInfo1).thenReturn(viewInfo2);
  doNothing().when(zkManager).setFileSetInfo(eq(request.getFileSet()), any(FileSetInfo.class));
  when(viewInfo1.getNumOnlinePartitions()).thenReturn(request.getExpectedNumPartitions() / 2);
  when(viewInfo2.getNumOnlinePartitions()).thenReturn(request.getExpectedNumPartitions());
  when(helixAdmin.getResourcesInCluster(CLUSTER)).thenReturn(new ArrayList<String>());
  when(ControllerUtil.buildIdealStateForHdfsDir(any(DFSClient.class),
      anyString(), anyString(), any(PartitionerType.class), anyInt(), anyBoolean())).
          thenReturn(is);
  doNothing().when(helixAdmin).addResource(eq(CLUSTER), eq(resourceName), eq(is));
  doNothing().when(helixAdmin).addResource(eq(CLUSTER), eq(resourceName),
      eq(is.getNumPartitions()), eq("OnlineOffline"), eq("CUSTOMIZED"), eq(is.getBucketSize()));
  doNothing().when(helixAdmin).setResourceIdealState(eq(CLUSTER), eq(resourceName), eq(is));
  serviceImpl.loadFileSet(request).apply();
  ArgumentCaptor<FileSetInfo> fileSetInfoCaptor = ArgumentCaptor.forClass(FileSetInfo.class);
  verify(zkManager).setFileSetInfo(eq(request.getFileSet()), fileSetInfoCaptor.capture());
  FileSetInfo capturedInfo = fileSetInfoCaptor.getValue();
  assertEquals(request.getFileSet(), capturedInfo.fileSetName);
  assertEquals(request.getExpectedNumPartitions(), capturedInfo.servingInfo.numPartitions);
  assertEquals(Lists.newArrayList(fileSetInfo.servingInfo),
      (ArrayList)capturedInfo.oldServingInfoList);
}