Avoid StackOverflowException when Kafka.poll() 'timeout' param is 0

[mariobriggs]

implement fetchBatch() as a loop when kafka.poll is not large enough interval for data to reach kafkaClient from kafkaServer

[markgrover]

Thanks for this @mariobriggs
Should we take out the recursive call below then?

I think the whole if clause can go then?

      if (!iter.hasNext) {
        if ( requestOffset < part.untilOffset ) {
          return getNext()
        }
        assert(requestOffset == part.untilOffset, errRanOutBeforeEnd(part))
        finished = true
        null.asInstanceOf[R]
      } else {

Because iter.hasNext will always be true when fetchBatch returns since it won't return until it has something non-empty to return.

BTW, we should definitely test to make sure this works when the topic is empty and doesn't stall. I will take care of that. Let me know what you think of the above.

[mariobriggs]

I think the removal of recursive call is right.

make sure this works when the topic is empty
<<
the '&& requestOffset < part.untilOffset' catches that i think

[markgrover]

Thanks!

[markgrover]

I am merging this, thanks again for this @mariobriggs
I can take care of the follow-up but do let me know if you disagree.