Add a check to make debugging easier when the client has been closed previously (e.g., an misplaced defer() closed it).
Address Evan's comments on the previous version.  Add checks for Closed() to a few other locations including consumer.go.
Code review fix
Merge pull request #117 from Shopify/rsrsps

Rebased version of #54 (closed checks)
Make mockbroker's ID an int32 to avoid a boatload of casts
Merge pull request #118 from Shopify/mockbroker-tweak

Make mockbroker's ID an int32 to avoid a boatload of casts
Validate the offset received from kafka when consuming
Only update the stored offset on successful events
Only disconnect the broker on error conditions

Otherwise we pull it out from underneath other producers/consumers who might be
talking to it.
Merge pull request #119 from Shopify/validate-offset

Validate the offset received from kafka when consuming
Merge pull request #120 from Shopify/leave-broker-connected

Only disconnect the broker on error conditions
Make sure we always call errorCb in flushRequest

Also return an error when we fail to re-enqueue, and make sure to close the
channel in the errorCb used for synchronous sends.

In combination, fixes #65
Return better error when server returns no offsets

Return OffsetOutOfRange instead of IncompleteResponse. Apparently when the
offset is out of range for an offset request, it doesn't return an error, just a
blank array, so translate that appropriately.

Fixes #123
Merge pull request #125 from Shopify/getoffset-better-error

Return better error when server returns no offsets
Merge pull request #124 from Shopify/fix-producer-deadlocks

Make sure we always call errorCb in flushRequest
Actually wrap the relevant underlying error in some cases

Might help debugging #71
Merge pull request #126 from Shopify/better-producer-errors

Actually wrap the relevant underlying error in some cases
Add two new channels to the Producer for handling acknowledgments from
Kafka and to retrieve messages that failed for possible replay.

ReturnFailedMessages in the config will trigger any messages that were not
successfully sent / acked to be sent on the failed channel as []byte.
The application using this package is responsible for reading from the
failed channel, retrieved via the Failed() function on the Producer   If
ReturnFailedMessages is true, then the application must consume the
Failed() channel or it will eventually block.

ReturnAckCounts in the config will trigger ack counts per partition, per
topic to be sent on the acked channel after each flush.
The application using this package is responsible for reading from the
acked channel, retrieved via the Acked() function on the Producer   If
ReturnFailedMessages is true, then the application must consume the
Acked() channel or it will eventually block.

This commit also includes a change for the error sent in the
DroppedMessagesError struct to the errorCb to be the block error instead
of the error returned from Produce.
Changes based on PR review.  Still need to decide on best way to return potential key with []byte message of failures.  Possibly a new struct.
Change failed channel to type *ConsumerEvent so all relevent information including Key, Value, Partition, Topic and Err can be consumed by the async caller.