docs: public void manageTopology()
        throws TopologyRuntimeManagementException, TMasterException, PackingException {
      String topologyName = Context.topologyName(config);
      // 1. Do prepare work
      // create an instance of state manager
      String statemgrClass = Context.stateManagerClass(config);
      IStateManager statemgr;
      try {
        statemgr = ReflectionUtils.newInstance(statemgrClass);
      } catch (IllegalAccessException | InstantiationException | ClassNotFoundException e) {
        throw new TopologyRuntimeManagementException(String.format(
            "Failed to instantiate state manager class '%s'",
            statemgrClass), e);
      }
  
      // Put it in a try block so that we can always clean resources
      try {
        // initialize the statemgr
        statemgr.initialize(config);
  
        // TODO(mfu): timeout should read from config
        SchedulerStateManagerAdaptor adaptor = new SchedulerStateManagerAdaptor(statemgr, 5000);
  
        boolean hasExecutionData = validateRuntimeManage(adaptor, topologyName);
  
        // 2. Try to manage topology if valid
        // invoke the appropriate command to manage the topology
        LOG.log(Level.FINE, "Topology: {0} to be {1}ed", new Object[]{topologyName, command});
  
        // build the runtime config
        Config runtime = Config.newBuilder()
            .put(Key.TOPOLOGY_NAME, Context.topologyName(config))
            .put(Key.SCHEDULER_STATE_MANAGER_ADAPTOR, adaptor)
            .build();
  
        // Create a ISchedulerClient basing on the config
        ISchedulerClient schedulerClient = getSchedulerClient(runtime);
  
        callRuntimeManagerRunner(runtime, schedulerClient, !hasExecutionData);
      } finally {
        // 3. Do post work basing on the result
        // Currently nothing to do here
  
        // 4. Close the resources
        SysUtils.closeIgnoringExceptions(statemgr);
      }
    }