code: @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())));
    }
docs: Return the next record when available. Will never time out since this is a streaming source.