KAFKA-2632: move fetchable check ahead in handleFetchResponse

[guozhangwang]

No description provided.

[ijuma]

I take it that it's not easy to write a test that shows the issue?

[guozhangwang]

It is a bit timing-dependent that you need to have an in-flight fetch request going on while calling pause(); this cannot be easily reproduced since consumer.poll() may trigger pollOnce() multiple times.

[hachikuji]

I think we still need this check. There may be a rebalance between the time a fetch response is handled and the time it is returned to the user.

[guozhangwang]

Thanks @hachikuji , comments addressed.

[hachikuji]

Isn't it a little dangerous to assume that the fetch position in the returned record is still relevant? Suppose we have the following sequence:

1. User calls poll() and we initiate a fetch.
2. User calls commitSync() (or some other blocking operation) and the fetch returns.
3. User calls seek() with a new offset.
4. User calls poll() and now we've written over their seeked offset.

Maybe instead we should reset the fetched position to the consumed position?

[guozhangwang]

@hachikuji I thought about this, but consumed() may return null for the first time so we cannot rely on that value either. Maybe we should just combine these two? I.e.:

If consumed() != null, set to consumed;
otherwise this MUST be the first fetch ever returned before any seek() was called, hence resetting to first offset should be safe.

[hachikuji]

LGTM.

[gwenshap]

LGTM