code: 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;
      }
docs: 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), ...>