Start time functionality

Created by: matthew-d-jones

Start time functionality is broken by an immediate partition rebalance after subscribe is called, which resets the offsets.