Add components for error handling and traversing logic for consumer

[areknoster]

Current implementation doesn't handle missing messages very gracefully.
A mechanism to implement traverse stop could also be useful.