question: @ Override public void nextTuple ( ) { Message msg ; msg = failedMessages . peek ( ) ; if ( msg != null ) { MessageRetries messageRetries = pendingMessageRetries . get ( msg . getMessageId ( ) ) ; if ( Backoff . shouldBackoff ( messageRetries . getTimeStamp ( ) , TimeUnit . NANOSECONDS , messageRetries . getNumRetries ( ) ) ) { Utils . sleep ( 100 ) ; } else { LOG . info ( "[{}] Retrying failed message {}" , spoutId , msg . getMessageId ( ) ) ; failedMessages . remove ( ) ; mapToValueAndEmit ( msg ) ; } return ; } if ( consumer != null ) { if ( LOG . isDebugEnabled ( ) ) { LOG . debug ( "[{}] Receiving the next message from pulsar consumer to emit to the collector" , spoutId ) ; } try { msg = consumer . receive ( 1 , TimeUnit . SECONDS ) ; if ( msg != null ) { ++ messagesReceived ; messageSizeReceived += msg . getData ( ) . length ; } mapToValueAndEmit ( msg ) ; } catch ( PulsarClientException e ) { LOG . error ( "[{}] Error receiving message from pulsar consumer" , spoutId , e ) ; } } }
target:  Emits a tuple received from the Pulsar consumer unless there are any failed messages