Introduce max_concurrent_consumed_messages: stop reading when threshold is reached on slow consumer callbacks

Introduce max_concurrent_consumed_messages: stop reading when threshold is reached on slow consumer callbacks #343