[jira] [Commented] (SAMZA-461) Race when initializing offsets at job startup leads to skipped messages

["Chris Riccomini (JIRA)" <ji...@apache.org>]

    [ https://issues.apache.org/jira/browse/SAMZA-461?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=14202200#comment-14202200 ] 

Chris Riccomini commented on SAMZA-461:
---------------------------------------

I think the most intuitive thing is to have the KafkaSystemConsumer pay attention to the samza.offset.default setting if auto.offset.reset isn't set. If samza.offset.default is set to "oldest", then auto.offset.reset should be set to oldest as well, unless it's already been set manually in the configs. That way, if the oldest offset is out of range, auto.offset.reset will kick in, and get the currently oldest offset.

Another way we could handle this would be to properly model it in Samza, but that's a bit tricky. We don't have the concept of "out of range" right now. I could see us getting into a race condition where we continue getting out of range exceptions, refreshing the metadata, and getting another out of range exception.

> Race when initializing offsets at job startup leads to skipped messages
> -----------------------------------------------------------------------
>
>                 Key: SAMZA-461
>                 URL: https://issues.apache.org/jira/browse/SAMZA-461
>             Project: Samza
>          Issue Type: Bug
>            Reporter: Ben Kirwin
>
> If the default offset is set to oldest, a Samza job should start from the very beginning of the stream:
> {code}
> systems.kafka.samza.offset.default=oldest
> {code}
> However, if the very first messages are added to the stream while the job is booting up, it's possible for those messages to be skipped entirely.
> When there are no messages in a stream, Samza reads the 'oldest' offset as null. This null value is added to the map of starting offsets in the offset manager. When the Kafka broker proxy gets the null offset, it complains:
> {code}
> It appears that we received an invalid or empty offset [...] Attempting to use Kafka's auto.offset.reset setting. This can result in data loss if processing continues.
> {code}
> If auto.offset.reset is not manually configured, this defaults to starting with the latest value. If messages have appeared in the stream in the meantime, the job will start *after* those messages, and data is indeed lost.
> It seems like setting oldestOffset to equal upcomingOffset would solve the issue. (It's also semantically reasonable -- the upcoming offset is indeed the oldest offset that will ever be read.)



--
This message was sent by Atlassian JIRA
(v6.3.4#6332)