fix(consumer): follow preferred broker

Historically (before protocol version 11) if we attempted to consume
from a follower, we would get a NotLeaderForPartition response and move
our consumer to the new leader. However, since v11 the Kafka broker
treats us just like any other follower and permits us to consume from
any replica and it is up to us to monitor metadata to determine when the
leadership has changed.

Modifying the handleResponse func to check the topic partition
leadership against the current broker (in the absence of a
preferredReadReplica) and trigger a re-create of the consumer for that
partition

Contributes-to: #1927

consumer.go

@@ -839,10 +839,12 @@ func (bc *brokerConsumer) handleResponses() {
 		child.responseResult = nil
 
 		if result == nil {
-			if child.preferredReadReplica >= 0 && bc.broker.ID() != child.preferredReadReplica {
-				// not an error but needs redispatching to consume from prefered replica
-				child.trigger <- none{}
-				delete(bc.subscriptions, child)
+			if preferredBroker, err := child.preferredBroker(); err == nil {
+				if bc.broker.ID() != preferredBroker.ID() {
+					// not an error but needs redispatching to consume from prefered replica
+					child.trigger <- none{}
+					delete(bc.subscriptions, child)
+				}
 			}
 			continue
 		}