Change aggregate to time out per consumer batch.
Merge branch '0.4.x'