If kafka is not available to not retry forever but notify it i someway

[davideicardi]

This allow for example to stop the app from outside. We can call the new onError callback.