Handling of backend errors #69
Description
Sink and source implementations could do a better job of handling errors served by the backend that it could remedy.
As an example, if a MetadataOutOfDate error is served by a Kafka backend, the implementation could refresh the metadata rather than serve the error directly back to the client.