samza-dev mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From Shanthoosh Venkataraman <santhoshvenkat1...@gmail.com>
Subject Re: Review Request 52476: Do not load task store which are older than delete tombstones.
Date Sat, 22 Oct 2016 22:06:20 GMT


> On Oct. 21, 2016, 12:45 a.m., Prateek Maheshwari wrote:
> > samza-core/src/main/java/org/apache/samza/storage/StorageRecovery.java, line 254
> > <https://reviews.apache.org/r/52476/diff/4/?file=1541854#file1541854line254>
> >
> >     Does jobConfig.getChangeLog...() (implicit conversion) not work?

No, the implicit conversion doesn't work here. The convenience method is part of StorageConfig
class.


> On Oct. 21, 2016, 12:45 a.m., Prateek Maheshwari wrote:
> > samza-core/src/main/scala/org/apache/samza/storage/TaskStorageManager.scala, line
130
> > <https://reviews.apache.org/r/52476/diff/4/?file=1541857#file1541857line130>
> >
> >     First %s should be store name. Add another %s at the end for loggedStoreDir.

This log message belongs to the task store, which would in itself contain the store name.
Adding store name here is unnecessary.


> On Oct. 21, 2016, 12:45 a.m., Prateek Maheshwari wrote:
> > samza-kafka/src/main/scala/org/apache/samza/config/KafkaConfig.scala, line 144
> > <https://reviews.apache.org/r/52476/diff/4/?file=1541859#file1541859line144>
> >
> >     implicit conversion should probably work.

No, implicit conversion doesn't work here, that is the reason for creating the object explicitly.


> On Oct. 21, 2016, 12:45 a.m., Prateek Maheshwari wrote:
> > samza-core/src/main/scala/org/apache/samza/storage/TaskStorageManager.scala, line
186
> > <https://reviews.apache.org/r/52476/diff/4/?file=1541857#file1541857line186>
> >
> >     s/partition/logged storage partition to be consistent with next message.

Done.


> On Oct. 21, 2016, 12:45 a.m., Prateek Maheshwari wrote:
> > samza-core/src/main/scala/org/apache/samza/storage/TaskStorageManager.scala, line
133
> > <https://reviews.apache.org/r/52476/diff/4/?file=1541857#file1541857line133>
> >
> >     Log both last modified time and delete retention ms values too.

Done.


> On Oct. 21, 2016, 12:45 a.m., Prateek Maheshwari wrote:
> > samza-core/src/main/scala/org/apache/samza/config/StorageConfig.scala, line 57
> > <https://reviews.apache.org/r/52476/diff/4/?file=1541855#file1541855line57>
> >
> >     getChangeLogDeleteRetentionsInMs

Done.


- Shanthoosh


-----------------------------------------------------------
This is an automatically generated e-mail. To reply, visit:
https://reviews.apache.org/r/52476/#review153478
-----------------------------------------------------------


On Oct. 22, 2016, 10:06 p.m., Shanthoosh Venkataraman wrote:
> 
> -----------------------------------------------------------
> This is an automatically generated e-mail. To reply, visit:
> https://reviews.apache.org/r/52476/
> -----------------------------------------------------------
> 
> (Updated Oct. 22, 2016, 10:06 p.m.)
> 
> 
> Review request for samza.
> 
> 
> Repository: samza
> 
> 
> Description
> -------
> 
> Every local task store is backed up by a kafka changelog topic. Due to log compaction,
delete tombstones of the changelog topic have a ttl of delete.retention.ms. Replaying the
events from the changelog that has missing delete tombstones, would result in creation of
an inconsistent local store(due to the missing of some delete events). This patch deletes
the local stores in which difference between current time and last modified time of the offset
file is greater than delete.retention.ms during the container startup.
> 
> 
> Diffs
> -----
> 
>   samza-core/src/main/java/org/apache/samza/storage/StorageRecovery.java 9329edf7d724f3a0d9235354bb77936f713e3b5f

>   samza-core/src/main/scala/org/apache/samza/config/StorageConfig.scala be3f1068f7921c9c8f8697577efea6580672813e

>   samza-core/src/main/scala/org/apache/samza/container/SamzaContainer.scala 05a996c98075ea8ed3767af666b9beeb1933f2a6

>   samza-core/src/main/scala/org/apache/samza/storage/TaskStorageManager.scala 0b7bcdda1639eea8239a69c31bdf42558e9077d2

>   samza-core/src/test/scala/org/apache/samza/storage/TestTaskStorageManager.scala 4d40f520e54beb643acd8410c772b75e2f6a9162

>   samza-kafka/src/main/scala/org/apache/samza/config/KafkaConfig.scala 973ab8cfb3d248bec7efe5e338f5e667f097556d

> 
> Diff: https://reviews.apache.org/r/52476/diff/
> 
> 
> Testing
> -------
> 
> Unit testing and manual testing has been done to verify the functionality.
> 
> 
> Thanks,
> 
> Shanthoosh Venkataraman
> 
>


Mime
  • Unnamed multipart/alternative (inline, None, 0 bytes)
View raw message