Experimental support for consumer offset restore and commit
Don't keep bp.hasMessages permanently signalled
Don't call bp.flush(p) twice in a row