Improve KafkaReadStream control

[vietj]

Currently the KafkaReadStream delivers item by batch of 10 (this should be configurable btw). When doing seek operations, the stream needs to be paused otherwise the handler will likely receive unwanted records that have been fetch and are delivered by the batch. This seems due to the fact that the current iterator is accessed from the worker thread only, unlike the paused flag.

One way to achieve it might be to pause() the consumer in seek operations. Alternatively the current iterator could be modified outside of the worker thread and checked when delivering items. Setting it to null would mean that a seek operation was required.

[vietj]

@ppatierno what's your opinion on this ?

[anhldbk]

@vietj Is pause a good way to handle slow consumers? Under a certain case, invoking pause will trigger operations on revoking/assigning paritions, right?