public void handleInstanceExecutor ( ) { for ( InstanceExecutor executor : taskIdToInstanceExecutor . values ( ) ) { boolean isLocalSpout = spoutSets . contains ( executor . getComponentName ( ) ) ; int taskId = executor . getTaskId ( ) ; int items = executor . getStreamOutQueue ( ) . size ( ) ; for ( int i = 0 ; i < items ; i ++ ) { Message msg = executor . getStreamOutQueue ( ) . poll ( ) ; if ( msg instanceof HeronTuples . HeronTupleSet ) { HeronTuples . HeronTupleSet tupleSet = ( HeronTuples . HeronTupleSet ) msg ; if ( tupleSet . hasData ( ) ) { HeronTuples . HeronDataTupleSet d = tupleSet . getData ( ) ; TopologyAPI . StreamId streamId = d . getStream ( ) ; for ( HeronTuples . HeronDataTuple tuple : d . getTuplesList ( ) ) { List < Integer > outTasks = this . topologyManager . getListToSend ( streamId , tuple ) ; outTasks . addAll ( tuple . getDestTaskIdsList ( ) ) ; if ( outTasks . isEmpty ( ) ) { LOG . severe ( "Nobody to send the tuple to" ) ; } copyDataOutBound ( taskId , isLocalSpout , streamId , tuple , outTasks ) ; } } if ( tupleSet . hasControl ( ) ) { HeronTuples . HeronControlTupleSet c = tupleSet . getControl ( ) ; for ( HeronTuples . AckTuple ack : c . getAcksList ( ) ) { copyControlOutBound ( tupleSet . getSrcTaskId ( ) , ack , true ) ; } for ( HeronTuples . AckTuple fail : c . getFailsList ( ) ) { copyControlOutBound ( tupleSet . getSrcTaskId ( ) , fail , false ) ; } } } } } }
Handle the execution of the instance