Wait for pending messages when pausing consumers

When we pause a consumer, we may have messages that we're already processing. We need some way to wait for these to clear.