code: public List<Integer> get(String stream, List<Object> tuple, Collection<Tuple> anchors, Object rootId) {
          List<Integer> outTasks = new ArrayList<>();
  
          // get grouper, then get which task should tuple be sent to.
          Map<String, MkGrouper> componentCrouping = streamComponentGrouper.get(stream);
          if (componentCrouping == null) {
              // if the target component's parallelism is 0, don't need send to
              // them
              LOG.debug("Failed to get Grouper of " + stream + " when " + debugIdStr);
              return outTasks;
          }
  
          for (Entry<String, MkGrouper> ee : componentCrouping.entrySet()) {
              String targetComponent = ee.getKey();
              MkGrouper g = ee.getValue();
  
              if (GrouperType.direct.equals(g.gettype())) {
                  throw new IllegalArgumentException("Cannot do regular emit to direct stream");
              }
  
              outTasks.addAll(g.grouper(tuple));
          }
  
          if (isDebug(anchors, rootId)) {
              LOG.info(debugIdStr + stream + " to " + outTasks + ":" + tuple.toString());
          }
          int num_out_tasks = outTasks.size();
          taskStats.send_tuple(stream, num_out_tasks);
  
          return outTasks;
      }
docs: send tuple according to grouping