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 ; }
direct send tuple to special task