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 ) ; } }
Submit and run topology in simulator