docs: public void submitTopology() throws TopologySubmissionException {
      // build primary runtime config first
      Config primaryRuntime = Config.newBuilder()
            .putAll(LauncherUtils.getInstance().createPrimaryRuntime(topology)).build();
      // call launcher directly here if in dry-run mode
      if (Context.dryRun(config)) {
        callLauncherRunner(primaryRuntime);
        return;
      }
      // 1. Do prepare work
      // create an instance of state manager
      String statemgrClass = Context.stateManagerClass(config);
      IStateManager statemgr;
  
      // Create an instance of the launcher class
      String launcherClass = Context.launcherClass(config);
      ILauncher launcher;
  
      // create an instance of the uploader class
      String uploaderClass = Context.uploaderClass(config);
      IUploader uploader;
  
      // create an instance of state manager
      try {
        statemgr = ReflectionUtils.newInstance(statemgrClass);
      } catch (IllegalAccessException | InstantiationException | ClassNotFoundException e) {
        throw new TopologySubmissionException(
            String.format("Failed to instantiate state manager class '%s'", statemgrClass), e);
      }
  
      // create an instance of launcher
      try {
        launcher = ReflectionUtils.newInstance(launcherClass);
      } catch (IllegalAccessException | InstantiationException | ClassNotFoundException e) {
        throw new LauncherException(
            String.format("Failed to instantiate launcher class '%s'", launcherClass), e);
      }
  
      // create an instance of uploader
      try {
        uploader = ReflectionUtils.newInstance(uploaderClass);
      } catch (IllegalAccessException | InstantiationException | ClassNotFoundException e) {
        throw new UploaderException(
            String.format("Failed to instantiate uploader class '%s'", uploaderClass), e);
      }
  
      // Put it in a try block so that we can always clean resources
      try {
        // initialize the state manager
        statemgr.initialize(config);
  
        // initialize the uploader
        uploader.initialize(config);
  
        // TODO(mfu): timeout should read from config
        SchedulerStateManagerAdaptor adaptor = new SchedulerStateManagerAdaptor(statemgr, 5000);
  
        // Check if topology is already running
        validateSubmit(adaptor, topology.getName());
  
        LOG.log(Level.FINE, "Topology {0} to be submitted", topology.getName());
  
        Config runtimeWithoutPackageURI = Config.newBuilder()
            .putAll(primaryRuntime)
            .putAll(LauncherUtils.getInstance().createAdaptorRuntime(adaptor))
            .put(Key.LAUNCHER_CLASS_INSTANCE, launcher)
            .build();
  
        PackingPlan packingPlan = LauncherUtils.getInstance()
            .createPackingPlan(config, runtimeWithoutPackageURI);
  
        // The packing plan might call for a number of containers different than the config
        // settings. If that's the case we need to modify the configs to match.
        runtimeWithoutPackageURI =
            updateNumContainersIfNeeded(runtimeWithoutPackageURI, topology, packingPlan);
  
        // If the packing plan is valid we will upload necessary packages
        URI packageURI = uploadPackage(uploader);
  
        // Update the runtime config with the packageURI
        Config runtimeAll = Config.newBuilder()
            .putAll(runtimeWithoutPackageURI)
            .put(Key.TOPOLOGY_PACKAGE_URI, packageURI)
            .build();
  
        callLauncherRunner(runtimeAll);
  
      } catch (LauncherException | PackingException e) {
        // we undo uploading of topology package only if launcher fails to
        // launch topology, which will throw LauncherException or PackingException
        uploader.undo();
        throw e;
      } finally {
        SysUtils.closeIgnoringExceptions(uploader);
        SysUtils.closeIgnoringExceptions(launcher);
        SysUtils.closeIgnoringExceptions(statemgr);
      }
    }