Remove external calls to disconnectBroker

[eapache]

It is now only called from one place in the client. This looks simple but is
actually super-subtle, and depends on lazy broker connections (PR #309).

disconnectBroker does a whole bunch of different things:

• calls Close on the broker connection
• adds the address to the internal deadBrokerAddrs map even if the broker is
  not a seed, which I think is wrong, since resurrectDeadBrokers will use it to
  repopulate the seedBrokerAddrs list
• rotates seedBrokers (if the broker was a seed broker)
• removes it from the brokers map (otherwise)

In the producer and consumer where we used to call disconnectBroker:

• We now call Close directly on the broker.
• The broker we are dealing with is not a seed broker, so the seedBrokers do not
  need rotating, and I don't think it's a problem that it no longer gets added
  to deadBrokerAddrs.
• The reason we removed it from the broker map was so that the next request for
  that broker would trigger a metadata request and reopen the connection. The
  producer and consumer both manually trigger metadata requests when necessary,
  and the fact that we now have lazy connection opening means simply closing it
  (which we do, see first bullet) is enough to cause the connection to reopen
  the next time it is requested, even if no metadata refresh is requested.

In a subsequent PR, the last remaining call will be inlined and refactored. This will effectively fix the last non-trivial issue spawned from #15.

@Shopify/kafka

[eapache]

The lazy connections have been merged so this is ready for review. The code is trivial (not much to review there), I'm more interested in somebody verifying that my logic is correct around deadBrokerAddrs, connection bouncing, etc.

As part of #309 this has already undergone some decent practical testing using the vagrant cluster and willem's stressproducer.

[wvanbergen]

Did we run any consumer tests by any chance?

[eapache]

Just ran some of those using your new console-consumer, but I'm not too concerned. The changes are symmetric so if the producer works fine then the consumer should (and does) also.

[wvanbergen]

I think the logic here is sound.

I don't think it's necessary to Close() the broker connection for every error we see in the producer or consumer, but this is one of these optimizations that are easy to fuck up.

What is the number of (producer) retries we potentially waste under relatively normal operations?

1. Produce requests fails (e.g. not leader for partition), so it asks for new metadata.
2. Metadata update request fails. This is already unlikely, because it happens only if it completely drains the Metadata retries as well, right?

[eapache]

I don't think it's necessary to Close() the broker connection for every error we see in the producer or consumer

You don't? If a broker gives us a network error or an unexpected protocol error then I don't see what choice we have besides disconnecting from that one and asking metdata for another one?

What is the number of (producer) retries we potentially waste under relatively normal operations?

No more than we ever have?

[wvanbergen]

I meant, not for every error. E.g. if we get a NotLeaderForPartition, we don;t technically have to disconnect. Or does that error never end up in abort?

Anyway, that's an optimization we don't really need.

[wvanbergen]

[eapache]

I meant, not for every error. E.g. if we get a NotLeaderForPartition, we don;t technically have to disconnect. Or does that error never end up in abort?

abort cleans up the entire worker - the only errors that end up there are ones returned from broker.Fetch(), nothing at the protocol layer.