[jira] [Updated] (SAMZA-142) Kafka changelog starts from latest offset instead of earliest

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

     [ https://issues.apache.org/jira/browse/SAMZA-142?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel ]

Chris Riccomini updated SAMZA-142:
----------------------------------

    Attachment: SAMZA-142.0.patch

Attaching a patch. RB available at:

https://reviews.apache.org/r/18539/

Changes:

1. Get metadata for change logs in container.
2. Use metadata to get oldest offsets for every change log.
3. Remove unused message collector in TaskStorage.
4. Update test stateful task to use the largest offset instead of smallest (to triger this bug). Since we switched to largest, I had to delete the second verification of 1-2-3-2 in the test since it's not re-processed when the job re-starts now (since we are using largest instead of smallest).

> Kafka changelog starts from latest offset instead of earliest
> -------------------------------------------------------------
>
>                 Key: SAMZA-142
>                 URL: https://issues.apache.org/jira/browse/SAMZA-142
>             Project: Samza
>          Issue Type: Bug
>          Components: container, kv
>    Affects Versions: 0.6.0
>            Reporter: Chris Riccomini
>         Attachments: SAMZA-142.0.patch
>
>
> SamzaContainer and TaskStorageManager improperly begin reading changelog streams from the "latest" offset in the default case, when using a KafkaSystemFactory for the changelog stream.
> In the logs you can see:
> {noformat}
> 2014-02-11 02:38:33 SamzaContainer$ [INFO] Got change log system streams: Map(realtime-state-store -> SystemStream [system=kafka, stream=realtime-state-store])
> ...
> 2014-02-11 02:38:36 SamzaContainer [INFO] Starting task instance stores.
> ...
> 2014-02-11 02:38:36 BrokerProxy [INFO] Creating new SimpleConsumer for host localhost:10251 for system kafka
> 2014-02-11 02:38:36 GetOffset [INFO] Checking if auto.offset.reset is defined for topic realtime-state-store
> 2014-02-11 02:38:36 GetOffset [INFO] Got reset of type largest.
> 2014-02-11 02:38:36 GetOffset [INFO] Final offset to be returned for Topic and Partition [realtime-state-store,96] = 3358324
> {noformat}
> This is wrong, as it means we never read any messages from the changelog. In Kafka's case (and in general), we want to start reading from the EARLIEST offset in the log (always zero for Kafka). This is caused by TaskStorageManager's startConsumers method, which calls register with a null offset. With a KafkaSystemConsumer, this means to defer to the auto.offset.reset setting, which defaults to "largest" if it isn't defined.
> Some possible ways that I can think of to fix this:
> 1. Introduce some kind of EARLIEST/LATEST constants for SystemConsumer.register, have TaskStorageManager use it, and KafkaSystemConsumer honor it.
> 2. Default KafkaSystemFactory.getConsumer to force all changelog stream auto-offset-resets to be EARLIEST when not defined.
> 3. Change KafkaSystemFactory auto-offset-reset to EARLIEST.
> 4. Add some kind of alternative "getChangelogConsumer" method to SystemConsumer.
> 5. Change the meaning of SystemConsumer.register(..., null) to always mean "earliest". Right now it means "defer to the consumer to decide what an appropriate offset is, since I don't have one."
> There are many more tweaks we could make as well. In general, I think the question is, do we want to fix this at the Kafka-level, or at the framework-wide level. (2) and (3) are Kafka-specific. (1), (4), and (5) are framework-wide, and would apply to all SystemConsumers.
> I think I'd rather fix this at the framework level. Right now, I think something like (1) seems like the best solution. It does add yet another constraint for SystemConsumer implementations, though, which is annoying.
> What do you think about having a SystemConsumer.START_FROM_EARLIEST_OFFSET = "__samza_EARLIEST" constant, and using that in TaskStorageManager.startConsumers? SystemConsumers that don't honor this constant couldn't be used as changelog streams.
> In the meantime, as a workaround, you can forcibly configure the auto.offset.reset setting to be smallest for all changelog streams.



--
This message was sent by Atlassian JIRA
(v6.1.5#6160)