Unnecessary wakeup and notifications of DCP stream

The implementation in the producer tries to "unpause" the DCP stream every time the core calls the step() method as part of fetching the "next" item. This happend unconditionally, even if the stream was paused due to the buffer ack window being full (and would require a message from the consumer to free up space). By unconditionally unpause every time we loose the information on how long we continuously waited for the client to free up some space (and we can't take any "actions" like disconnect the client due to the window being full for 15 days). 

The stream do keep some metrics around (the number of times it was paused due to the various reasons and the aggregated duration it spent in the paused state for the reason), but that information is only reported when the stream gets disconnected (so for a stream which never disconnect we won't see them). We do report the number of times the stream was paused / unpaused to prometheus, but those counters do not sched any light on why the stream was paused.

Example: There is an idle bucket and a new DCP consumer connects and we see no progress after a short while. Is that because the client window is full; is it fully caught up; is there a bug due to a missing notification so it is blocked on an empty "ready queue"

In addition to the unconditional wakeup there is an unnecessary call to schedule a notification as part of processing the buffer ack message. The received message from the consumer and the step method will always be called from the same thread context and the core first execute incomming messages (the buffer ack) and when its done processing input requests it'll try making progess on the DCP stream by calling step()