Commit/rollback consumer offsets via context manager

We've found it useful to control the consumer's commit operations based on the progress of the consuming process (e.g. without auto commit) and wrote a context manager to handle commits and rollbacks. We're using this logic both to defer processing of a topic until messages have been queued for a certain amount of time and to retry processing of the last failed message (and all subsequent messages in the same topic partition) until some external error has been resolved.

Hopefully this is useful to others.