code: public List<Integer> get(Integer out_task_id, String stream, List<Object> tuple,
                               Collection<Tuple> anchors, Object root_id) {
  
          // in order to improve acker's performance, skip checking
          // String target_component =
          // topologyContext.getComponentId(out_task_id);
          // Map<String, MkGrouper> component_prouping = streamComponentGrouper
          // .get(stream);
          // MkGrouper grouping = component_prouping.get(target_component);
          // if (grouping != null &&
          // !GrouperType.direct.equals(grouping.gettype())) {
          // throw new IllegalArgumentException(
          // "Cannot emitDirect to a task expecting a regular grouping");
          // }
  
          if (isDebug(anchors, root_id)) {
              LOG.info(debugIdStr + stream + " to " + out_task_id + ":" + tuple);
          }
  
          taskStats.send_tuple(stream, 1);
  
          List<Integer> out_tasks = new ArrayList<>();
          out_tasks.add(out_task_id);
          return out_tasks;
      }
docs: direct send tuple to special task