Write offsets to datasets only if message is processed

BTW, I think it's also a good idea to emit a user metrics on how many messages have been polled from Kafka.