docs: public void start() throws Exception {
      // 1. Do prepare work
      // create an instance of state manager
      String statemgrClass = Context.stateManagerClass(config);
      LOG.info("Context.stateManagerClass " + statemgrClass);
      IStateManager statemgr;
      try {
        statemgr = ReflectionUtils.newInstance(statemgrClass);
      } catch (IllegalAccessException | InstantiationException | ClassNotFoundException e) {
        throw new Exception(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);
  
        Boolean b = statemgr.setMetricsCacheLocation(metricsCacheLocation, topologyName)
            .get(5000, TimeUnit.MILLISECONDS);
        if (b != null && b) {
          LOG.info("metricsCacheLocation " + metricsCacheLocation.toString());
          LOG.info("topologyName " + topologyName.toString());
  
          LOG.info("Starting Metrics Cache HTTP Server");
          metricsCacheManagerHttpServer.start();
  
          // 2. The MetricsCacheServer would run in the main thread
          // We do it in the final step since it would await the main thread
          LOG.info("Starting Metrics Cache Server");
          metricsCacheManagerServer.start();
          metricsCacheManagerServerLoop.loop();
        } else {
          throw new RuntimeException("Failed to set metricscahe location.");
        }
      } finally {
        // 3. Do post work basing on the result
        // Currently nothing to do here
  
        // 4. Close the resources
        SysUtils.closeIgnoringExceptions(statemgr);
      }
  
  
    }