public void manageTopology ( ) throws TopologyRuntimeManagementException , TMasterException , PackingException { String topologyName = Context . topologyName ( config ) ; // 1. Do prepare work // create an instance of state manager String statemgrClass = Context . stateManagerClass ( config ) ; IStateManager statemgr ; try { statemgr = ReflectionUtils . newInstance ( statemgrClass ) ; } catch ( IllegalAccessException | InstantiationException | ClassNotFoundException e ) { throw new TopologyRuntimeManagementException ( String . format ( "Failed to instantiate state manager class '%s'" , statemgrClass ) , e ) ; } // Put it in a try block so that we can always clean resources try { // initialize the statemgr statemgr . initialize ( config ) ; // TODO(mfu): timeout should read from config SchedulerStateManagerAdaptor adaptor = new SchedulerStateManagerAdaptor ( statemgr , 5000 ) ; boolean hasExecutionData = validateRuntimeManage ( adaptor , topologyName ) ; // 2. Try to manage topology if valid // invoke the appropriate command to manage the topology LOG . log ( Level . FINE , "Topology: {0} to be {1}ed" , new Object [ ] { topologyName , command } ) ; // build the runtime config Config runtime = Config . newBuilder ( ) . put ( Key . TOPOLOGY_NAME , Context . topologyName ( config ) ) . put ( Key . SCHEDULER_STATE_MANAGER_ADAPTOR , adaptor ) . build ( ) ; // Create a ISchedulerClient basing on the config ISchedulerClient schedulerClient = getSchedulerClient ( runtime ) ; callRuntimeManagerRunner ( runtime , schedulerClient , ! hasExecutionData ) ; } finally { // 3. Do post work basing on the result // Currently nothing to do here // 4. Close the resources SysUtils . closeIgnoringExceptions ( statemgr ) ; } }
Manager a topology 1 . Instantiate necessary resources 2 . Valid whether the runtime management is legal 3 . Complete the runtime management for a specific command