public void start ( ) throws Exception { // 1. Do prepare work // create an instance of state manager String statemgrClass = Context . stateManagerClass ( config ) ; LOG . info ( "Context.stateManagerClass " + statemgrClass ) ; IStateManager statemgr ; try { statemgr = ReflectionUtils . newInstance ( statemgrClass ) ; } catch ( IllegalAccessException | InstantiationException | ClassNotFoundException e ) { throw new Exception ( 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 ) ; Boolean b = statemgr . setMetricsCacheLocation ( metricsCacheLocation , topologyName ) . get ( 5000 , TimeUnit . MILLISECONDS ) ; if ( b != null && b ) { LOG . info ( "metricsCacheLocation " + metricsCacheLocation . toString ( ) ) ; LOG . info ( "topologyName " + topologyName . toString ( ) ) ; LOG . info ( "Starting Metrics Cache HTTP Server" ) ; metricsCacheManagerHttpServer . start ( ) ; // 2. The MetricsCacheServer would run in the main thread // We do it in the final step since it would await the main thread LOG . info ( "Starting Metrics Cache Server" ) ; metricsCacheManagerServer . start ( ) ; metricsCacheManagerServerLoop . loop ( ) ; } else { throw new RuntimeException ( "Failed to set metricscahe location." ) ; } } finally { // 3. Do post work basing on the result // Currently nothing to do here // 4. Close the resources SysUtils . closeIgnoringExceptions ( statemgr ) ; } }
start statemgr_client metricscache_server http_server