class MessageIterator implements Iterator<ConsumerRecord<byte[], byte[]>>, AutoCloseable {

  @Override
  public boolean hasNext() {
    if (recordIter != null && recordIter.hasNext()) {
      return true;
    }

    long nextPosition = kafkaConsumer.position(topicPartition);
    if (nextPosition >= endOffset) {
      return false;
    }

    ConsumerRecords<byte[], byte[]> consumerRecords;
    Stopwatch stopwatch = logger.isDebugEnabled() ? Stopwatch.createStarted() : null;
    try {
      consumerRecords = kafkaConsumer.poll(kafkaPollTimeOut);
    } catch (KafkaException ke) {
      throw UserException.dataReadError(ke).message(ke.getMessage()).build(logger);
    } finally {
      if (stopwatch != null) {
        stopwatch.stop();
      }
    }

    if (consumerRecords.isEmpty()) {
      throw UserException.dataReadError()
        .message("Failed to fetch messages within %s milliseconds. " +
          "Consider increasing the value of the property: %s",
          kafkaPollTimeOut, ExecConstants.KAFKA_POLL_TIMEOUT)
        .build(logger);
    }

    if (stopwatch != null) {
      long lastFetchTime = stopwatch.elapsed(TimeUnit.MILLISECONDS);
      logger.debug("Time taken to fetch : {} milliseconds", lastFetchTime);
      totalFetchTime += lastFetchTime;
      logger.debug("Total number of messages fetched : {}", consumerRecords.count());
    }

    recordIter = consumerRecords.iterator();
    return recordIter.hasNext();
  }

  public  MessageIterator(final KafkaConsumer<byte[], byte[]> kafkaConsumer, final KafkaPartitionScanSpec subScanSpec,
      final long kafkaPollTimeOut);

}