docs: public boolean runScheduler() {
      IScheduler scheduler = null;
  
      String statemgrClass = Context.stateManagerClass(config);
      IStateManager statemgr;
  
      try {
        // create an instance of state manager
        statemgr = ReflectionUtils.newInstance(statemgrClass);
      } catch (IllegalAccessException | InstantiationException | ClassNotFoundException e) {
        LOG.log(Level.SEVERE, "Failed to instantiate instances using config: " + config, e);
        return false;
      }
  
      SchedulerServer server = null;
      boolean isSuccessful = false;
  
      // Put it in a try block so that we can always clean resources
      try {
        // initialize the state manager
        statemgr.initialize(config);
  
        // TODO(mfu): timeout should read from config
        SchedulerStateManagerAdaptor adaptor = new SchedulerStateManagerAdaptor(statemgr, 5000);
  
        // get a packed plan and schedule it
        PackingPlans.PackingPlan serializedPackingPlan = adaptor.getPackingPlan(topology.getName());
        if (serializedPackingPlan == null) {
          LOG.log(Level.SEVERE, "Failed to fetch PackingPlan for topology:{0} from the state manager",
              topology.getName());
          return false;
        }
        LOG.log(Level.INFO, "Packing plan fetched from state: {0}", serializedPackingPlan);
        PackingPlan packedPlan = new PackingPlanProtoDeserializer().fromProto(serializedPackingPlan);
  
        // build the runtime config
        LauncherUtils launcherUtils = LauncherUtils.getInstance();
        Config runtime = Config.newBuilder()
            .putAll(launcherUtils.createPrimaryRuntime(topology))
            .putAll(launcherUtils.createAdaptorRuntime(adaptor))
            .put(Key.SCHEDULER_SHUTDOWN, getShutdown())
            .put(Key.SCHEDULER_PROPERTIES, properties)
            .build();
  
        Config ytruntime = launcherUtils.createConfigWithPackingDetails(runtime, packedPlan);
  
        // invoke scheduler
        scheduler = launcherUtils.getSchedulerInstance(config, ytruntime);
        if (scheduler == null) {
          return false;
        }
  
        isSuccessful = scheduler.onSchedule(packedPlan);
        if (!isSuccessful) {
          LOG.severe("Failed to schedule topology");
          return false;
        }
  
        // Failures in server initialization throw exceptions
        // get the scheduler server endpoint for receiving requests
        server = getServer(ytruntime, scheduler, schedulerServerPort);
        // start the server to manage runtime requests
        server.start();
  
        // write the scheduler location to state manager
        // Make sure it happens after IScheduler.onScheduler
        isSuccessful = SchedulerUtils.setSchedulerLocation(
            runtime, String.format("%s:%d", server.getHost(), server.getPort()), scheduler);
  
        if (isSuccessful) {
          // wait until kill request or some interrupt occurs if the scheduler starts successfully
          LOG.info("Waiting for termination... ");
          Runtime.schedulerShutdown(ytruntime).await();
        }
      } catch (IOException e) {
        LOG.log(Level.SEVERE, "Failed to start server", e);
        return false;
      } finally {
        // Clean the resources
        if (server != null) {
          server.stop();
        }
  
        // 4. Close the resources
        SysUtils.closeIgnoringExceptions(scheduler);
        SysUtils.closeIgnoringExceptions(statemgr);
      }
  
      return isSuccessful;
    }