Support Fetch v8 protocol (including client-side throttling)

[ankon]

This is a pre-condition for #713 (we need Fetch v11 for KIP-392).

v8 is exactly the same as v7, the difference is purely in how quota throttling is to be understood. See https://cwiki.apache.org/confluence/display/KAFKA/KIP-219+-+Improve+quota+communication

This provides the last puzzle piece to resolve #161.

[Nevon]

If we start supporting v8, then we should also start honoring the throttle time by not sending any more fetch requests during that time, if I'm reading the KIP right. To make things more complicated, we should only introduce this backoff if the throttle time in the response came from a Fetch response v8, since for v7 the broker will have already throttled the client by delaying the response.

[ankon]

Fair enough. My ultimate goal is to get to KIP-392 as quickly as possible, so I would be looking into a minimum implementation for client-side throttling here. :)

I see that KIP-219 changes a lot of requests/responses so that each of these can indicate client-side throttling, but from what I understand this still means we can only support it for some requests. IOW: It looks like supporting it for Fetch v8 is fine, as long as there is a reasonable path forward to add similar behavior to other APIs.

What the java client does looks quite reasonable:

1. Identify per response whether this is a "client-side throttling response", and then if so provide a default "throttling time" value (apache/kafka@1facab3#diff-bcbf176f64322c655e3016ca72cc3f58R165-R177)
2. When processing any response check for whether that is applicable, and if there is needed throttling mark the connection state for that broker accordingly
3. Delay sending stuff on that connection

The Java client does use a version callback, looking at JavaScript and the KafkaJS approach to requests it seems easier to have a flag for "should client-side throttle" and a value for the "how long aspect". One could even put these into the same value (clientSideThrottleTimeMs), and define that a value > 0 changes the connection state.

For the delaying then the Connection could be having such a "throtte until" flag, and if that is set it would basically not send out messages until that time is hit. I'm still reading through these areas.

WDYT, would such an approach be acceptable for merging?

[Nevon]

I think that sounds very reasonable. And as you say, we can opt into this API-per-API as we upgrade versions.

[ankon]

For the test failures: These now come from the various places that check responses against a pre-defined "expected" response. Depending on the Kafka version though we will now get a different response.

I'm going to fix these in the next commit:

1. Always set throttleTime to 0 for Fetch V8 responses: This is the server-side throttling applied to the response, and with Fetch V8 the broker won't apply any. Older code can check for that though, and because we never implemented tests for throttling the value will be "just fine".
2. Modify the tests to accept the clientSideThrottleTime as an optional field: When running with older versions of Kafka this won't be there, and with never versions it will be -- but then set to the 0 value as there aren't any quotas defined.

[ankon]

#786 fixes two issues with tests that lead to errors now. I rather not mix up the commits though, so for this branch it's somewhat expected that the tests mentioned in that PR fail here.

[ankon]

I rebased and shuffled the changes a bit around:

1. First some cosmetics
2. Then implement infrastructure for client-side throttling
3. Then use that infrastructure in the Fetch v8 request/response definitions