Added ability to ack back to underlying broker when a set of messages has been received.