@ Override public RecordEnvelope < D > readRecordEnvelopeImpl ( ) throws DataRecordException , IOException { if ( ! _isStarted . get ( ) ) { throw new IOException ( "Streaming extractor has not been started." ) ; } while ( ( _records == null ) || ( ! _records . hasNext ( ) ) ) { synchronized ( _consumer ) { if ( _close . get ( ) ) { throw new ClosedChannelException ( ) ; } _records = _consumer . poll ( this . fetchTimeOut ) . iterator ( ) ; } } ConsumerRecord < S , D > record = _records . next ( ) ; _rowCount . getAndIncrement ( ) ; return new RecordEnvelope < D > ( record . value ( ) , new KafkaWatermark ( _partition , new LongWatermark ( record . offset ( ) ) ) ) ; }
Return the next record when available . Will never time out since this is a streaming source .