Write offsets to datasets only if message is processed

We talked about the user metrics for Kafka. The useful metrics will be on a topic level so it is better to do it downstream. Not going to address that.