public void call ( ) throws LauncherException , PackingException , SubmitDryRunResponse { SchedulerStateManagerAdaptor statemgr = Runtime . schedulerStateManagerAdaptor ( runtime ) ; TopologyAPI . Topology topology = Runtime . topology ( runtime ) ; String topologyName = Context . topologyName ( config ) ; PackingPlan packedPlan = LauncherUtils . getInstance ( ) . createPackingPlan ( config , runtime ) ; if ( Context . dryRun ( config ) ) { throw new SubmitDryRunResponse ( topology , config , packedPlan ) ; } // initialize the launcher launcher . initialize ( config , runtime ) ; // Set topology def first since we determine whether a topology is running // by checking the existence of topology def // store the trimmed topology definition into the state manager // TODO(rli): log-and-false anti-pattern is too nested on this path. will not refactor Boolean result = statemgr . setTopology ( trimTopology ( topology ) , topologyName ) ; if ( result == null || ! result ) { throw new LauncherException ( String . format ( "Failed to set topology definition for topology '%s'" , topologyName ) ) ; } result = statemgr . setPackingPlan ( createPackingPlan ( packedPlan ) , topologyName ) ; if ( result == null || ! result ) { statemgr . deleteTopology ( topologyName ) ; throw new LauncherException ( String . format ( "Failed to set packing plan for topology '%s'" , topologyName ) ) ; } // store the execution state into the state manager ExecutionEnvironment . ExecutionState executionState = createExecutionState ( ) ; result = statemgr . setExecutionState ( executionState , topologyName ) ; if ( result == null || ! result ) { statemgr . deletePackingPlan ( topologyName ) ; statemgr . deleteTopology ( topologyName ) ; throw new LauncherException ( String . format ( "Failed to set execution state for topology '%s'" , topologyName ) ) ; } // launch the topology, clear the state if it fails if ( ! launcher . launch ( packedPlan ) ) { statemgr . deleteExecutionState ( topologyName ) ; statemgr . deletePackingPlan ( topologyName ) ; statemgr . deleteTopology ( topologyName ) ; throw new LauncherException ( String . format ( "Failed to launch topology '%s'" , topologyName ) ) ; } }
Call launcher to launch topology