SimpleConsumer may allow consume() after stop()

[yungchin]

This issue was raised as part of #215 - if there's still a message on some queue, consume() will return that instead of raising a ConsumerStoppedException. That's an issue because there won't be an offset auto-committer at that point.