docs: public void submitTopology(String name, Config heronConfig, HeronTopology heronTopology) {
      TopologyAPI.Topology topologyToRun =
          heronTopology.
              setConfig(heronConfig).
              setName(name).
              setState(TopologyAPI.TopologyState.RUNNING).
              getTopology();
  
      if (!TopologyUtils.verifyTopology(topologyToRun)) {
        throw new RuntimeException("Topology object is Malformed");
      }
  
      // TODO (nlu): add simulator support stateful processing
      if (isTopologyStateful(heronConfig)) {
        throw new RuntimeException("Stateful topology is not supported");
      }
  
      TopologyManager topologyManager = new TopologyManager(topologyToRun);
  
      LOG.info("Physical Plan: \n" + topologyManager.getPhysicalPlan());
  
      // Create the stream executor
      streamExecutor = new StreamExecutor(topologyManager);
  
      // Create the metrics executor
      metricsExecutor = new MetricsExecutor(systemConfig);
  
      // Create instance Executor
      for (PhysicalPlans.Instance instance : topologyManager.getPhysicalPlan().getInstancesList()) {
        InstanceExecutor instanceExecutor = new InstanceExecutor(
            topologyManager.getPhysicalPlan(),
            instance.getInstanceId()
        );
  
        streamExecutor.addInstanceExecutor(instanceExecutor);
        metricsExecutor.addInstanceExecutor(instanceExecutor);
        instanceExecutors.add(instanceExecutor);
      }
  
      // Start - run executors
      // Add exception handler for any uncaught exception here.
      Thread.setDefaultUncaughtExceptionHandler(new DefaultExceptionHandler());
  
      threadsPool.execute(metricsExecutor);
      threadsPool.execute(streamExecutor);
      for (InstanceExecutor instanceExecutor : instanceExecutors) {
        threadsPool.execute(instanceExecutor);
      }
    }