Fix message retries
Refactor leaderDispatcher

We actually don't have to keep two buffers: as soon as we see a retry and flush
our chaser we can start sending out subsequent retries right away. This should
improve recovery time in cases where the broker goes away.
Remove one sync point from the fast path

In the fast path, the brokerDispatcher was acting as an unnecessary sync point
(bottleneck) between disparate partitions; there's no reason for the
leaderDispatcher not to send each message directly to the messageAggregator. The
only time a sync is actually necessary is when the leader changes due to a
broker going offline or similar.

Instead of a singleton goroutine whose input/output channels were acting as
implicit synchronization, use a map with a mutex that we only lock when we
really actually have to. This has other implicit benefits:
- reference-counting for proper shutdown also lets us clean up unused broker
  goroutines automatically (partitionDispatcher and leaderDispatcher goroutines
  still hang around until Close is called)
- one fewer goroutine, making up for the addition of the retryHandler
wrap all goroutines in withRecover