Major Description
It is wrong for KafkaStorage to look for the offset with largest timestamp smaller than the "start timestamp" on replay. Instead, it should look for the smallest timestamp that is equal to or larger than the "start timestamp" since "start timestamp" means messages before that time have all been processed.