periodically checkpoint messages and timestamp to Kafka