Easier way to stop processors on failure

[faolivera]

It would be useful to be able to stop consuming events from the stream if some persistent problem is detected while processing events (database connection problem for instance) and be able to start consuming events again after certain condition is given.

To do that it is needed to stop a processor when its inner consumer detects an error.
The issue is there is no a direct way to map the failing consumer to the processor that created it, so when an error happens inside an observer there is no an easy way to figure out which processor it is needed to be stopped and recreated

[dehora]

This is probably going to need some design, possibly a carrier object that tracks processors and consumers.

In the meantime, there's been some changes to stop behaviour in 0.9.0 along with the option to throw a NonRetryableNakadiException from the observer to kill the consumer pipeline.

[olayinkasf]

Would it also be reasonable to have a shutdown on the client to cleanly shutdown all processors. IMO at the moment it is still unclear how to approach a proper shutdown.