Store fetched offsets separately.

Fetch requests can be repeated if we get a ConsumerFetchSizeTooSmall
or if _fetch() is called multiple times for some reason. We don't want
to re-fetch messages that are already in our queue, so store the offsets
of the last enqueued messages from each partition.
Only use timeout if it's not None
Make get_messages() update and commit offsets just before returning