private List<Integer> boltEmit(String out_stream_id,            Collection<Tuple> anchors, List<Object> values, Integer out_task_id) {        timer.start();        List<Integer> out_tasks = null;        try {            if (out_task_id != null) {                out_tasks = sendTargets.get(out_task_id, out_stream_id, values);            } else {                out_tasks = sendTargets.get(out_stream_id, values);            }            for (Integer t : out_tasks) {                Map<Long, Long> anchors_to_ids = new HashMap<Long, Long>();                if (anchors != null) {                    for (Tuple a : anchors) {                        //Long edge_id = MessageId.generateId();                        Long edge_id = MessageId.generateId(random);                        long now = System.currentTimeMillis();                        if (now - lastRotate > rotateTime) {                            pending_acks.rotate();                            lastRotate = now;                        }                        put_xor(pending_acks, a, edge_id);                        for (Long root_id : a.getMessageId().getAnchorsToIds()                                .keySet()) {                            put_xor(anchors_to_ids, root_id, edge_id);                        }                    }                }            }            return out_tasks;        } catch (Exception e) {            LOG.error("bolt emit", e);        }finally {            timer.stop();        }        return new ArrayList<Integer>();    }