public static StormTopology normalizeTopology ( Map stormConf , StormTopology topology , boolean fromConf ) { StormTopology ret = topology . deepCopy ( ) ; Map < String , Object > rawComponents = ThriftTopologyUtils . getComponents ( topology ) ; Map < String , Object > components = ThriftTopologyUtils . getComponents ( ret ) ; if ( ! rawComponents . keySet ( ) . equals ( components . keySet ( ) ) ) { String errMsg = "Failed to normalize topology binary!" ; LOG . info ( errMsg + " raw components:" + rawComponents . keySet ( ) + ", normalized " + components . keySet ( ) ) ; throw new InvalidParameterException ( errMsg ) ; } for ( Entry < String , Object > entry : components . entrySet ( ) ) { Object component = entry . getValue ( ) ; String componentName = entry . getKey ( ) ; ComponentCommon common = null ; if ( component instanceof Bolt ) { common = ( ( Bolt ) component ) . get_common ( ) ; if ( fromConf ) { Integer paraNum = ConfigExtension . getBoltParallelism ( stormConf , componentName ) ; if ( paraNum != null ) { LOG . info ( "Set " + componentName + " as " + paraNum ) ; common . set_parallelism_hint ( paraNum ) ; } } } if ( component instanceof SpoutSpec ) { common = ( ( SpoutSpec ) component ) . get_common ( ) ; if ( fromConf ) { Integer paraNum = ConfigExtension . getSpoutParallelism ( stormConf , componentName ) ; if ( paraNum != null ) { LOG . info ( "Set " + componentName + " as " + paraNum ) ; common . set_parallelism_hint ( paraNum ) ; } } } if ( component instanceof StateSpoutSpec ) { common = ( ( StateSpoutSpec ) component ) . get_common ( ) ; if ( fromConf ) { Integer paraNum = ConfigExtension . getSpoutParallelism ( stormConf , componentName ) ; if ( paraNum != null ) { LOG . info ( "Set " + componentName + " as " + paraNum ) ; common . set_parallelism_hint ( paraNum ) ; } } } Map componentMap = new HashMap ( ) ; String jsonConfString = common . get_json_conf ( ) ; if ( jsonConfString != null ) { componentMap . putAll ( ( Map ) JStormUtils . from_json ( jsonConfString ) ) ; } Integer taskNum = componentParalism ( stormConf , common ) ; componentMap . put ( Config . TOPOLOGY_TASKS , taskNum ) ; // change the executor's task number common . set_parallelism_hint ( taskNum ) ; LOG . info ( "Set " + componentName + " parallelism " + taskNum ) ; common . set_json_conf ( JStormUtils . to_json ( componentMap ) ) ; } return ret ; }
finalize component s task parallelism