bug: handle and log throttled FetchResponses

[dnwe]

Kafka 2.2 returns a throttled FetchResponse with an empty array of topics.
Previously Sarama treated this as an ErrIncompleteResponse and would
abandon and re-create the subscription instead of retrying the
FetchRequest.

[mimaison]

LGTM

[mimaison]

@bai Since 2.0, Kafka can return an empty FetchResponse in case Quotas are violated (KIP-219).

Without this fix, when this happens, we get an error, for example:

kafka: error while consuming hub-crusher-6partition/0: kafka: response did not contain all the expected topic/partition blocks

[bai]

Many thanks!

[gaffneyc]

Should this in some way slow down FetchResponse requests? When we have a process that is behind in messages we've started to see a ton of FetchResponse entries in the log and eventually processing for a single consumer comes to a complete halt.

[dnwe]

@gaffneyc yes a well-behaved client should backoff its request rate when throttled, but recent broker versions do also enforce this at the backend by sleeping on the socket

[gaffneyc]

Okay, to make sure I understand you're saying the server's socket is handling the throttling of requests so nothing needs to be done on the client side? Since we're seeing throttle values of 15s is there any configuration we should do in our client side to deal with that?

What we're seeing is a consumer coming to a complete deadlock and only a consumer rebalance will fix it. I'm guessing it would be better to open a new ticket.