Major Description
When SystemProducer got a MessageSizeTooLargeException from Kafka broker, the container keeps trying and the job stucks at send call. The behavior in 0.9.0 will be different (i.e. the container will shutdown on exception). The SystemConsumer deserialization exception exhibit a similar behavior.
It would be nice to add a ListenableTask interface w/ an onException() handler that allows the user to catch these exceptions and allow application logic on the exceptions.