i run into similar problem, i think `planInputPartitions(start, end)` is supposed to be idempotent, it can be called multiple times. then how should i do things like receiving messages from sqs upon triggered? the latestOffset() is the place, we need to introduce a cache to hold the result.