public static Map<GlobalStreamId, Grouping> acker_inputs(StormTopology topology) {
        Map<GlobalStreamId, Grouping> spout_inputs = new HashMap<>();
        Map<String, SpoutSpec> spout_ids = topology.get_spouts();
        for (Entry<String, SpoutSpec> spout : spout_ids.entrySet()) {
            String id = spout.getKey();

            GlobalStreamId stream = new GlobalStreamId(id, ACKER_INIT_STREAM_ID);

            Grouping group = Thrift.mkFieldsGrouping(JStormUtils.mk_list("id"));

            spout_inputs.put(stream, group);
        }

        Map<String, Bolt> bolt_ids = topology.get_bolts();
        Map<GlobalStreamId, Grouping> bolt_inputs = new HashMap<>();
        for (Entry<String, Bolt> bolt : bolt_ids.entrySet()) {
            String id = bolt.getKey();

            GlobalStreamId streamAck = new GlobalStreamId(id, ACKER_ACK_STREAM_ID);
            Grouping groupAck = Thrift.mkFieldsGrouping(JStormUtils.mk_list("id"));

            GlobalStreamId streamFail = new GlobalStreamId(id, ACKER_FAIL_STREAM_ID);
            Grouping groupFail = Thrift.mkFieldsGrouping(JStormUtils.mk_list("id"));

            bolt_inputs.put(streamAck, groupAck);
            bolt_inputs.put(streamFail, groupFail);
        }

        Map<GlobalStreamId, Grouping> allInputs = new HashMap<>();
        allInputs.putAll(bolt_inputs);
        allInputs.putAll(spout_inputs);
        return allInputs;
    }
public static Map < GlobalStreamId , Grouping > acker_inputs ( StormTopology topology ) { Map < GlobalStreamId , Grouping > spout_inputs = new HashMap < > ( ) ; Map < String , SpoutSpec > spout_ids = topology . get_spouts ( ) ; for ( Entry < String , SpoutSpec > spout : spout_ids . entrySet ( ) ) { String id = spout . getKey ( ) ; GlobalStreamId stream = new GlobalStreamId ( id , ACKER_INIT_STREAM_ID ) ; Grouping group = Thrift . mkFieldsGrouping ( JStormUtils . mk_list ( "id" ) ) ; spout_inputs . put ( stream , group ) ; } Map < String , Bolt > bolt_ids = topology . get_bolts ( ) ; Map < GlobalStreamId , Grouping > bolt_inputs = new HashMap < > ( ) ; for ( Entry < String , Bolt > bolt : bolt_ids . entrySet ( ) ) { String id = bolt . getKey ( ) ; GlobalStreamId streamAck = new GlobalStreamId ( id , ACKER_ACK_STREAM_ID ) ; Grouping groupAck = Thrift . mkFieldsGrouping ( JStormUtils . mk_list ( "id" ) ) ; GlobalStreamId streamFail = new GlobalStreamId ( id , ACKER_FAIL_STREAM_ID ) ; Grouping groupFail = Thrift . mkFieldsGrouping ( JStormUtils . mk_list ( "id" ) ) ; bolt_inputs . put ( streamAck , groupAck ) ; bolt_inputs . put ( streamFail , groupFail ) ; } Map < GlobalStreamId , Grouping > allInputs = new HashMap < > ( ) ; allInputs . putAll ( bolt_inputs ) ; allInputs . putAll ( spout_inputs ) ; return allInputs ; }
Generate acker's input Map<GlobalStreamId, Grouping>

for spout <GlobalStreamId(spoutId, ACKER_INIT_STREAM_ID), ...> for
bolt <GlobalStreamId(boltId, ACKER_ACK_STREAM_ID), ...> <GlobalStreamId(boltId,
ACKER_FAIL_STREAM_ID), ...>