protected boolean fillBuffer ( ) { if ( ! _inProgress . isEmpty ( ) || ! _queue . isEmpty ( ) ) { throw new IllegalStateException ( "cannot fill buffer when buffer or pending messages are non-empty" ) ; } if ( _iterator == null ) { // create a stream of messages from _consumer using the streams as defined on construction final Map < String , List < KafkaStream < byte [ ] , byte [ ] > > > streams = _consumer . createMessageStreams ( Collections . singletonMap ( _topic , 1 ) ) ; _iterator = streams . get ( _topic ) . get ( 0 ) . iterator ( ) ; } // We'll iterate the stream in a try-clause; kafka stream will poll its client channel for the next message, // throwing a ConsumerTimeoutException when the configured timeout is exceeded. try { int size = 0 ; while ( size < _bufSize && _iterator . hasNext ( ) ) { final MessageAndMetadata < byte [ ] , byte [ ] > message = _iterator . next ( ) ; final KafkaMessageId id = new KafkaMessageId ( message . partition ( ) , message . offset ( ) ) ; _inProgress . put ( id , message . message ( ) ) ; size ++ ; } } catch ( final InvalidMessageException e ) { LOG . warn ( e . getMessage ( ) , e ) ; } catch ( final ConsumerTimeoutException e ) { // ignore, storm will call nextTuple again at some point in the near future // timeout does *not* mean that no messages were read (state is checked below) } if ( _inProgress . size ( ) > 0 ) { // set _queue to all currently pending kafka message ids _queue . addAll ( _inProgress . keySet ( ) ) ; LOG . debug ( "buffer now has {} messages to be emitted" , _queue . size ( ) ) ; // message(s) appended to buffer return true ; } else { // no messages appended to buffer return false ; } }
Refills the buffer with messages from the configured kafka topic if available .