gets a topology tree & nodes for topology graph utilization

@param stormTopology jstorm sys-topology
@return a TopologyGraph
501817
public static TopologyGraph getTopologyGraph(StormTopology stormTopology, List<MetricInfo> componentMetrics) {
        Map<String, Bolt> bolts = stormTopology.get_bolts();
        Map<String, SpoutSpec> spouts = stormTopology.get_spouts();

        //remove system bolts
        String[] remove_ids = new String[]{"__acker", "__system", "__topology_master"};
        for (String id : remove_ids) {
            bolts.remove(id);
        }

        //<id, node>
        Map<String, TopologyNode> nodes = Maps.newHashMap();
        //<from:to,edge>
        Map<String, TopologyEdge> edges = Maps.newHashMap();


        List<TreeNode> roots = Lists.newArrayList();            //this is used to get tree depth
        Map<String, TreeNode> linkMap = Maps.newHashMap();      //this is used to get tree depth
        // init the nodes
        for (Map.Entry<String, SpoutSpec> entry : spouts.entrySet()) {
            String componentId = entry.getKey();
            nodes.put(componentId, new TopologyNode(componentId, componentId, true));

            TreeNode node = new TreeNode(componentId);
            roots.add(node);
            linkMap.put(componentId, node);
        }
        for (Map.Entry<String, Bolt> entry : bolts.entrySet()) {
            String componentId = entry.getKey();
            nodes.put(componentId, new TopologyNode(componentId, componentId, false));
            linkMap.put(componentId, new TreeNode(componentId));
        }

        //init the edges
        int edgeId = 1;
        for (Map.Entry<String, Bolt> entry : bolts.entrySet()) {
            String componentId = entry.getKey();
            Bolt bolt = entry.getValue();
            TreeNode node = linkMap.get(componentId);
            for (Map.Entry<GlobalStreamId, Grouping> input : bolt.get_common().get_inputs().entrySet()) {
                GlobalStreamId streamId = input.getKey();
                String src = streamId.get_componentId();
                if (nodes.containsKey(src)) {
                    TopologyEdge edge = new TopologyEdge(src, componentId, edgeId++);
                    edges.put(edge.getKey(), edge);
                    // put into linkMap
                    linkMap.get(src).addChild(node);
                    node.addSource(src);
                    node.addParent(linkMap.get(src));
                }
            }
        }

        //calculate whether has circle
        boolean isFixed = false;
        while (!isFixed) {
            isFixed = true;
            for (TreeNode node : linkMap.values()) {
                for (TreeNode parent : node.getParents()) {
                    if (!node.addSources(parent.getSources())) {
                        isFixed = false;
                    }
                }
            }
        }


        // fill value to edges
        fillTPSValue2Edge(componentMetrics, edges);
        fillTLCValue2Edge(componentMetrics, edges);


        // fill value to nodes
        fillValue2Node(componentMetrics, nodes);

        // calculate notes' depth
        int maxDepth = bfsDepth(roots);

        // set nodes level & get max breadth
        Map<Integer, Integer> counter = Maps.newHashMap();
        int maxBreadth = 1;
        for (Map.Entry<String, TreeNode> entry : linkMap.entrySet()) {
            int layer = entry.getValue().getLayer();
            nodes.get(entry.getKey()).setLevel(layer);
            int breadth = 1;
            if (counter.containsKey(layer)) {
                breadth = counter.get(layer) + 1;
            }
            counter.put(layer, breadth);
            if (maxBreadth < breadth) {
                maxBreadth = breadth;
            }
        }

        //adjust graph for components in one line
        String PREFIX = "__adjust_prefix_";
        int count = 1;
        for (TreeNode node : linkMap.values()) {
            int layer = node.getLayer();
            node.setBreadth(counter.get(layer));
        }
        for (TreeNode tree : linkMap.values()) {
            if (isInOneLine(tree.getChildren())) {
                String id = PREFIX + count++;
                TopologyNode node = new TopologyNode(id, id, false);
                node.setLevel(tree.getLayer() + 1);
                node.setIsHidden(true);
                nodes.put(id, node);

                TreeNode furthest = getFurthestNode(tree.getChildren());

                TopologyEdge toEdge = new TopologyEdge(id, furthest.getComponentId(), edgeId++);
                toEdge.setIsHidden(true);
                edges.put(toEdge.getKey(), toEdge);

                TopologyEdge fromEdge = new TopologyEdge(tree.getComponentId(), id, edgeId++);
                fromEdge.setIsHidden(true);
                edges.put(fromEdge.getKey(), fromEdge);
            }
        }

        TopologyGraph graph = new TopologyGraph(Lists.newArrayList(nodes.values()), Lists.newArrayList(edges.values()));

        graph.setDepth(maxDepth);
        graph.setBreadth(maxBreadth);

        //create graph object
        return graph;
    }
831431
@SuppressWarnings("unchecked")
    public static Map component_conf(TopologyContext topology_context, String component_id) {
        Map<Object, Object> componentConf = new HashMap<>();
        String jconf = topology_context.getComponentCommon(component_id).get_json_conf();
        if (jconf != null) {
            componentConf = (Map<Object, Object>) JStormUtils.from_json(jconf);
        }

        return componentConf;
    }
80379
public TopologyMaster.MetricsCacheLocation getMetricsCacheLocation(String topologyName) {
    return awaitResult(delegate.getMetricsCacheLocation(null, topologyName));
  }
1079563
private static TopologyAPI.TopologyState getRuntimeTopologyState(
      String topologyName,
      SchedulerStateManagerAdaptor statemgr) throws TMasterException {
    PhysicalPlans.PhysicalPlan plan = statemgr.getPhysicalPlan(topologyName);

    if (plan == null) {
      throw new TMasterException(String.format(
        "Failed to get physical plan for topology '%s'", topologyName));
    }

    return plan.getTopology().getState();
  }
579155
public static void addTopologyMaster(Map stormConf, StormTopology ret) {
        // generate outputs
        HashMap<String, StreamInfo> outputs = new HashMap<>();

        List<String> list = JStormUtils.mk_list(TopologyMaster.FILED_CTRL_EVENT);
        outputs.put(TOPOLOGY_MASTER_CONTROL_STREAM_ID, Thrift.outputFields(list));
        list = JStormUtils.mk_list(TopologyMaster.FIELD_METRIC_WORKER, TopologyMaster.FIELD_METRIC_METRICS);
        outputs.put(TOPOLOGY_MASTER_METRICS_STREAM_ID, Thrift.outputFields(list));
        // TM --> heartbeat
        list = JStormUtils.mk_list(TopologyMaster.FILED_HEARBEAT_EVENT);
        outputs.put(TOPOLOGY_MASTER_HB_STREAM_ID, Thrift.outputFields(list));
        // TM --> register metrics
        list = JStormUtils.mk_list(TopologyMaster.FIELD_REGISTER_METRICS);
        outputs.put(TOPOLOGY_MASTER_REGISTER_METRICS_STREAM_ID, Thrift.outputFields(list));
        // TM --> register metrics resp
        list = JStormUtils.mk_list(TopologyMaster.FIELD_REGISTER_METRICS_RESP);
        outputs.put(TOPOLOGY_MASTER_REGISTER_METRICS_RESP_STREAM_ID, Thrift.outputFields(list));
        list = JStormUtils.mk_list(TopologyMaster.FILED_UDF_STREAM_EVENT);
        outputs.put(TopologyMaster.USER_DEFINED_STREAM, Thrift.outputFields(list));
        

        IBolt topologyMaster = new TopologyMaster();

        // generate inputs
        Map<GlobalStreamId, Grouping> inputs = topoMasterInputs(ret);

        // generate topology master which will be stored in topology
        Bolt topologyMasterBolt = Thrift.mkBolt(inputs, topologyMaster, outputs, 1);

        // add output stream to spout/bolt
        for (Entry<String, Bolt> e : ret.get_bolts().entrySet()) {
            Bolt bolt = e.getValue();
            ComponentCommon common = bolt.get_common();
            List<String> fields = JStormUtils.mk_list(TopologyMaster.FIELD_METRIC_WORKER, TopologyMaster.FIELD_METRIC_METRICS);
            common.put_to_streams(TOPOLOGY_MASTER_METRICS_STREAM_ID, Thrift.directOutputFields(fields));
            fields = JStormUtils.mk_list(TopologyMaster.FILED_HEARBEAT_EVENT);
            common.put_to_streams(TOPOLOGY_MASTER_HB_STREAM_ID, Thrift.directOutputFields(fields));
            fields = JStormUtils.mk_list(TopologyMaster.FILED_CTRL_EVENT);
            common.put_to_streams(TOPOLOGY_MASTER_CONTROL_STREAM_ID, Thrift.directOutputFields(fields));
            fields = JStormUtils.mk_list(TopologyMaster.FIELD_REGISTER_METRICS);
            common.put_to_streams(TOPOLOGY_MASTER_REGISTER_METRICS_STREAM_ID, Thrift.directOutputFields(fields));
            fields = JStormUtils.mk_list(TopologyMaster.FIELD_REGISTER_METRICS_RESP);
            common.put_to_streams(TOPOLOGY_MASTER_REGISTER_METRICS_RESP_STREAM_ID, Thrift.directOutputFields(fields));
            fields = JStormUtils.mk_list(TopologyMaster.FILED_UDF_STREAM_EVENT);
            common.put_to_streams(TopologyMaster.USER_DEFINED_STREAM, Thrift.directOutputFields(fields));

            GlobalStreamId stream = new GlobalStreamId(TOPOLOGY_MASTER_COMPONENT_ID, TOPOLOGY_MASTER_CONTROL_STREAM_ID);
            common.put_to_inputs(stream, Thrift.mkDirectGrouping());
            bolt.set_common(common);
        }

        for (Entry<String, SpoutSpec> kv : ret.get_spouts().entrySet()) {
            SpoutSpec spout = kv.getValue();
            ComponentCommon common = spout.get_common();
            List<String> fields = JStormUtils.mk_list(TopologyMaster.FIELD_METRIC_WORKER, TopologyMaster.FIELD_METRIC_METRICS);
            common.put_to_streams(TOPOLOGY_MASTER_METRICS_STREAM_ID, Thrift.directOutputFields(fields));
            fields = JStormUtils.mk_list(TopologyMaster.FILED_HEARBEAT_EVENT);
            common.put_to_streams(TOPOLOGY_MASTER_HB_STREAM_ID, Thrift.directOutputFields(fields));
            fields = JStormUtils.mk_list(TopologyMaster.FILED_CTRL_EVENT);
            common.put_to_streams(TOPOLOGY_MASTER_CONTROL_STREAM_ID, Thrift.directOutputFields(fields));
            fields = JStormUtils.mk_list(TopologyMaster.FIELD_REGISTER_METRICS);
            common.put_to_streams(TOPOLOGY_MASTER_REGISTER_METRICS_STREAM_ID, Thrift.directOutputFields(fields));
            fields = JStormUtils.mk_list(TopologyMaster.FIELD_REGISTER_METRICS_RESP);
            common.put_to_streams(TOPOLOGY_MASTER_REGISTER_METRICS_RESP_STREAM_ID, Thrift.directOutputFields(fields));
            fields = JStormUtils.mk_list(TopologyMaster.FILED_UDF_STREAM_EVENT);
            common.put_to_streams(TopologyMaster.USER_DEFINED_STREAM, Thrift.directOutputFields(fields));

            GlobalStreamId stream = new GlobalStreamId(TOPOLOGY_MASTER_COMPONENT_ID, TOPOLOGY_MASTER_CONTROL_STREAM_ID);
            common.put_to_inputs(stream, Thrift.mkDirectGrouping());
            spout.set_common(common);
        }

        ret.put_to_bolts(TOPOLOGY_MASTER_COMPONENT_ID, topologyMasterBolt);
    }
726093
public static Collection<IAutoCredentials> GetAutoCredentials(Map storm_conf) {
        try {
            Set<IAutoCredentials> autos = new HashSet<IAutoCredentials>();
            Collection<String> clazzes = (Collection<String>) storm_conf.get(Config.TOPOLOGY_AUTO_CREDENTIALS);
            if (clazzes != null) {
                for (String clazz : clazzes) {
                    IAutoCredentials a = (IAutoCredentials) Class.forName(clazz).newInstance();
                    a.prepare(storm_conf);
                    autos.add(a);
                }
            }
            LOG.info("Got AutoCreds " + autos);
            return autos;
        } catch (Exception e) {
            throw new RuntimeException(e);
        }
    }
816936
@SuppressWarnings("unchecked")
    public static void validate_ids(StormTopology topology, String topologyId) throws InvalidTopologyException {
        String topologyName = topologyIdToName(topologyId);
        if (!charValidate(topologyName)) {
            throw new InvalidTopologyException(topologyName + " is not a valid topology name. " + nameErrorInfo);
        }

        List<String> list = new ArrayList<>();
        for (StormTopology._Fields field : Thrift.STORM_TOPOLOGY_FIELDS) {
            Object value = topology.getFieldValue(field);
            if (value != null) {
                Map<String, Object> obj_map = (Map<String, Object>) value;
                Set<String> commids = obj_map.keySet();

                for (String id : commids) {
                    if (system_id(id) || !charComponentValidate(id)) {
                        throw new InvalidTopologyException(id + " is not a valid component id. " + compErrorInfo);
                    }
                }

                for (Object obj : obj_map.values()) {
                    validate_component(obj);
                }

                list.addAll(commids);
            }
        }

        List<String> offending = JStormUtils.getRepeat(list);
        if (!offending.isEmpty()) {
            throw new InvalidTopologyException("Duplicate component ids: " + offending);
        }
    }
703703
public CacheTopology computePreferredTopology(Map<Address, CacheStatusResponse> statusResponseMap) {
      List<Partition> partitions = computePartitions(statusResponseMap, "");

      return partitions.size() != 0 ? selectPreferredPartition(partitions).topology : null;
   }
134837
private String getTopologyFromRandomNodeWithRetry(Set<Host> activeHosts) {
		int count = NUM_RETRIES_PER_NODE;
		String nodeResponse;
		Exception lastEx;
		final Host randomHost = getRandomHost(activeHosts);
		do {
			try {
				lastEx = null;
				nodeResponse = getResponseViaHttp(randomHost.getHostName());
				if (nodeResponse != null) {
					Logger.info("Received topology from " + randomHost);
					return nodeResponse;
				}
			} catch (Exception e) {
				Logger.info("cannot get topology from : " + randomHost);
				lastEx = e;
			} finally {
				count--;
			}

		} while ((count > 0));

		if (lastEx != null) {
			if (lastEx instanceof ConnectTimeoutException) {
				throw new TimeoutException("Unable to obtain topology", lastEx).setHost(randomHost);
			}
			throw new DynoException(String.format("Unable to obtain topology from %s", randomHost), lastEx);
		} else {
			throw new DynoException(String.format("Could not contact dynomite manager for token map on %s", randomHost));
		}

    }
282618
public JsonNode getApplicationsTree() throws IOException {
        Invocation invocation = getJerseyClient().target(getEndpoint() + "/v1/applications/fetch").request().buildGet();
        return invocation.invoke().readEntity(JsonNode.class);
    }
807404
@SuppressWarnings("unchecked")
    private void doRefreshTopologies() {
        for (String topology : JStormMetrics.SYS_TOPOLOGIES) {
            if (!context.getTopologyMetricContexts().containsKey(topology)) {
                LOG.info("adding {} to metric context.", topology);
                Map conf = new HashMap();
                if (topology.equals(JStormMetrics.CLUSTER_METRIC_KEY)) {
                    //there's no need to consider sample rate when cluster metrics merge
                    conf.put(ConfigExtension.TOPOLOGY_METRIC_SAMPLE_RATE, 1.0);
                }
                Set<ResourceWorkerSlot> workerSlot = Sets.newHashSet(new ResourceWorkerSlot());
                TopologyMetricContext metricContext = new TopologyMetricContext(topology, workerSlot, conf);
                context.getTopologyMetricContexts().putIfAbsent(topology, metricContext);
                syncMetaFromCache(topology, context.getTopologyMetricContexts().get(topology));
            }

            // in the first 10 min, sync every 1 min
            if (context.getNimbusData().uptime() < INIT_SYNC_REMOTE_META_TIME_SEC) {
                syncMetaFromRemote(topology, context.getTopologyMetricContexts().get(topology),
                        Lists.newArrayList(MetaType.TOPOLOGY, MetaType.WORKER, MetaType.NIMBUS));
            } else { // after that, sync every 5 min
                checkTimeAndSyncMetaFromRemote(topology, context.getTopologyMetricContexts().get(topology),
                        Lists.newArrayList(MetaType.values()));

            }
        }

        Map<String, Assignment> assignMap;
        try {
            assignMap = Cluster.get_all_assignment(context.getStormClusterState(), null);
            for (Entry<String, Assignment> entry : assignMap.entrySet()) {
                String topology = entry.getKey();
                Assignment assignment = entry.getValue();

                TopologyMetricContext metricContext = context.getTopologyMetricContexts().get(topology);
                if (metricContext == null) {
                    metricContext = new TopologyMetricContext(assignment.getWorkers());
                    metricContext.setTaskNum(NimbusUtils.getTopologyTaskNum(assignment));
                    syncMetaFromCache(topology, metricContext);

                    LOG.info("adding {} to metric context.", topology);
                    context.getTopologyMetricContexts().put(topology, metricContext);
                } else {
                    boolean modify = false;
                    if (metricContext.getTaskNum() != NimbusUtils.getTopologyTaskNum(assignment)) {
                        modify = true;
                        metricContext.setTaskNum(NimbusUtils.getTopologyTaskNum(assignment));
                    }

                    if (!assignment.getWorkers().equals(metricContext.getWorkerSet())) {
                        modify = true;
                        metricContext.setWorkerSet(assignment.getWorkers());
                    }

                    // for normal topologies, only sync topology/component level metrics
                    checkTimeAndSyncMetaFromRemote(topology, metricContext,
                            Lists.newArrayList(MetaType.TOPOLOGY, MetaType.COMPONENT));

                    // we may need to sync meta when task num/workers change
                    metricContext.setSyncMeta(!modify);
                }
            }
        } catch (Exception e1) {
            LOG.warn("Failed to get assignments");
            return;
        }

        List<String> removing = new ArrayList<>();
        for (String topology : context.getTopologyMetricContexts().keySet()) {
            if (!JStormMetrics.SYS_TOPOLOGY_SET.contains(topology) && !assignMap.containsKey(topology)) {
                removing.add(topology);
            }
        }

        for (String topology : removing) {
            LOG.info("removing topology:{}", topology);
            RemoveTopologyEvent.pushEvent(topology);
            syncRemoteTimes.remove(topology);
        }
    }
737737
public static clusternodegroup[] get(nitro_service service) throws Exception{
		clusternodegroup obj = new clusternodegroup();
		clusternodegroup[] response = (clusternodegroup[])obj.get_resources(service);
		return response;
	}
1006060
public static Map<GlobalStreamId, Grouping> topoMasterInputs(StormTopology topology) {
        GlobalStreamId stream;
        Grouping group;

        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();

            stream = new GlobalStreamId(id, TOPOLOGY_MASTER_HB_STREAM_ID);
            group = Thrift.mkAllGrouping();
            spout_inputs.put(stream, group);

            stream = new GlobalStreamId(id, TOPOLOGY_MASTER_METRICS_STREAM_ID);
            group = Thrift.mkAllGrouping();
            spout_inputs.put(stream, group);

            stream = new GlobalStreamId(id, TOPOLOGY_MASTER_CONTROL_STREAM_ID);
            group = Thrift.mkAllGrouping();
            spout_inputs.put(stream, group);

            stream = new GlobalStreamId(id, TOPOLOGY_MASTER_REGISTER_METRICS_STREAM_ID);
            group = Thrift.mkAllGrouping();
            spout_inputs.put(stream, group);

            stream = new GlobalStreamId(id, TOPOLOGY_MASTER_REGISTER_METRICS_RESP_STREAM_ID);
            group = Thrift.mkAllGrouping();
            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();
            stream = new GlobalStreamId(id, TOPOLOGY_MASTER_HB_STREAM_ID);
            group = Thrift.mkAllGrouping();
            bolt_inputs.put(stream, group);

            stream = new GlobalStreamId(id, TOPOLOGY_MASTER_METRICS_STREAM_ID);
            group = Thrift.mkAllGrouping();
            bolt_inputs.put(stream, group);

            stream = new GlobalStreamId(id, TOPOLOGY_MASTER_CONTROL_STREAM_ID);
            group = Thrift.mkAllGrouping();
            bolt_inputs.put(stream, group);

            stream = new GlobalStreamId(id, TOPOLOGY_MASTER_REGISTER_METRICS_STREAM_ID);
            group = Thrift.mkAllGrouping();
            bolt_inputs.put(stream, group);

            stream = new GlobalStreamId(id, TOPOLOGY_MASTER_REGISTER_METRICS_RESP_STREAM_ID);
            group = Thrift.mkAllGrouping();
            bolt_inputs.put(stream, group);
        }

        Map<GlobalStreamId, Grouping> himself_inputs = new HashMap<>();
        stream = new GlobalStreamId(TOPOLOGY_MASTER_COMPONENT_ID, TOPOLOGY_MASTER_HB_STREAM_ID);
        group = Thrift.mkAllGrouping();
        himself_inputs.put(stream, group);

        stream = new GlobalStreamId(TOPOLOGY_MASTER_COMPONENT_ID, TOPOLOGY_MASTER_METRICS_STREAM_ID);
        group = Thrift.mkAllGrouping();
        himself_inputs.put(stream, group);

        stream = new GlobalStreamId(TOPOLOGY_MASTER_COMPONENT_ID, TOPOLOGY_MASTER_REGISTER_METRICS_STREAM_ID);
        group = Thrift.mkAllGrouping();
        himself_inputs.put(stream, group);

        stream = new GlobalStreamId(TOPOLOGY_MASTER_COMPONENT_ID, TOPOLOGY_MASTER_REGISTER_METRICS_RESP_STREAM_ID);
        group = Thrift.mkAllGrouping();
        himself_inputs.put(stream, group);

        Map<GlobalStreamId, Grouping> allInputs = new HashMap<>();
        allInputs.putAll(bolt_inputs);
        allInputs.putAll(spout_inputs);
        allInputs.putAll(himself_inputs);
        return allInputs;
    }
406799
def _get_topology(self):
        
        self._topology.open()
        with self.__lock:
            self._kill_cursors_executor.open()
        return self._topology
1234803
def get_topology(self):
        
        topology = {'name': self._name,
                    'resources_type': 'local',
                    'topology': {},
                    'type': 'topology',
                    'version': '1.0'}

        if self._links:
            topology['topology']['links'] = self._links
        if self._nodes:
            topology['topology']['nodes'] = self._nodes
        if self._servers:
            topology['topology']['servers'] = self._servers
        if self._notes:
            topology['topology']['notes'] = self._notes
        if self._shapes['ellipse']:
            topology['topology']['ellipses'] = self._shapes['ellipse']
        if self._shapes['rectangle']:
            topology['topology']['rectangles'] = \
                self._shapes['rectangle']
        if self._images:
            topology['topology']['images'] = self._images

        return topology
789854
public TopologyAPI.Topology trimTopology(TopologyAPI.Topology topology) {

    // create a copy of the topology physical plan
    TopologyAPI.Topology.Builder builder = TopologyAPI.Topology.newBuilder().mergeFrom(topology);

    // clear the state of user spout java objects - which can be potentially huge
    for (TopologyAPI.Spout.Builder spout : builder.getSpoutsBuilderList()) {
      spout.getCompBuilder().clearSerializedObject();
    }

    // clear the state of user spout java objects - which can be potentially huge
    for (TopologyAPI.Bolt.Builder bolt : builder.getBoltsBuilderList()) {
      bolt.getCompBuilder().clearSerializedObject();
    }

    return builder.build();
  }
66258
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);
    }