Fix to support batched/compressed messages

These messages span offsets, so "next" may sometimes be zero.
KafkaContext changed to pass the actual message offset.