@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);
    }
  }
}

@ 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 ) ; } } }
emits a tuple received from the pulsar consumer unless there are any failed messages
