Major Description
Introduce a systems.<system name>.streams.<stream name>.samza.offset.default configuration, which specifies what to do when no checkpoint exists for an input topic. This is a similar to setting to Kafka's auto.offset.reset setting. Developers will be able to specify "oldest", "latest", or "fail".
we should also add the ability to override offsets for specific stream partitions. Something like:
systems.<system name>.streams.<stream name>.samza.force.offsets=0:123,1:123,2:123
The format I'm proposing is:
<partition string>:<force offset>,...
This is obviously dependent on offsets not having ':' or ',' in them, which I think is a safe assumption.
This setting would force the system consumer to be registered with the specified offset for the given SSP (ignoring both the checkpoint, if it exists, and the samza.reset.offset setting).