eachBatch should respect partition restriction despite concurrency

[milindl]

Fixes #222

Thanks @omer-riv for filing this issue and providing a reproducing case!

Details of the issue:
In the cache, next() "gives up" a partition when it tries to fetch something, and returns empty handed. The message returned to the user in this case is also null.

In the cache, nextN "gives up" partition assignment not only when it returns empty handed, but when we consume to the end of the partition (either leftover size = N or with -1 to consume everything). This is problematic because, we're returning a not-null set of messages to the user but not retaining partition ownership for that user.

Solution: Make nextN like next.

Check the cache test for more details.

[confluent-cla-assistant]

🎉 All Contributor License Agreements have been signed. Ready to merge.
_{Please push an empty commit if you would like to re-run the checks to verify CLA status for all contributors.}

[emasab]

LGTM, thanks Milind and Omer!