public static void add_acker ( Map stormConf , StormTopology ret ) { String key = Config . TOPOLOGY_ACKER_EXECUTORS ; Integer ackerNum = JStormUtils . parseInt ( stormConf . get ( key ) , 0 ) ; // generate outputs HashMap < String , StreamInfo > outputs = new HashMap <> ( ) ; ArrayList < String > fields = new ArrayList <> ( ) ; fields . add ( "id" ) ; outputs . put ( ACKER_ACK_STREAM_ID , Thrift . directOutputFields ( fields ) ) ; outputs . put ( ACKER_FAIL_STREAM_ID , Thrift . directOutputFields ( fields ) ) ; IBolt ackerbolt = new Acker ( ) ; // generate inputs Map < GlobalStreamId , Grouping > inputs = acker_inputs ( ret ) ; // generate acker which will be stored in topology Bolt acker_bolt = Thrift . mkBolt ( inputs , ackerbolt , outputs , ackerNum ) ; // add every bolt two output stream // ACKER_ACK_STREAM_ID/ACKER_FAIL_STREAM_ID for ( Entry < String , Bolt > e : ret . get_bolts ( ) . entrySet ( ) ) { Bolt bolt = e . getValue ( ) ; ComponentCommon common = bolt . get_common ( ) ; List < String > ackList = JStormUtils . mk_list ( "id" , "ack-val" ) ; common . put_to_streams ( ACKER_ACK_STREAM_ID , Thrift . outputFields ( ackList ) ) ; List < String > failList = JStormUtils . mk_list ( "id" ) ; common . put_to_streams ( ACKER_FAIL_STREAM_ID , Thrift . outputFields ( failList ) ) ; bolt . set_common ( common ) ; } // add every spout output stream ACKER_INIT_STREAM_ID // add every spout two intput source // ((ACKER_COMPONENT_ID, ACKER_ACK_STREAM_ID), directGrouping) // ((ACKER_COMPONENT_ID, ACKER_FAIL_STREAM_ID), directGrouping) for ( Entry < String , SpoutSpec > kv : ret . get_spouts ( ) . entrySet ( ) ) { SpoutSpec bolt = kv . getValue ( ) ; ComponentCommon common = bolt . get_common ( ) ; List < String > initList = JStormUtils . mk_list ( "id" , "init-val" , "spout-task" ) ; common . put_to_streams ( ACKER_INIT_STREAM_ID , Thrift . outputFields ( initList ) ) ; GlobalStreamId ack_ack = new GlobalStreamId ( ACKER_COMPONENT_ID , ACKER_ACK_STREAM_ID ) ; common . put_to_inputs ( ack_ack , Thrift . mkDirectGrouping ( ) ) ; GlobalStreamId ack_fail = new GlobalStreamId ( ACKER_COMPONENT_ID , ACKER_FAIL_STREAM_ID ) ; common . put_to_inputs ( ack_fail , Thrift . mkDirectGrouping ( ) ) ; } ret . put_to_bolts ( ACKER_COMPONENT_ID , acker_bolt ) ; }
Add acker bolt to topology