> On Feb. 4, 2015, 2:15 a.m., Joel Koshy wrote: > > core/src/main/scala/kafka/api/OffsetCommitRequest.scala, line 48 > > <https://reviews.apache.org/r/27391/diff/11/?file=832423#file832423line48> > > > > I our convention is to include the if in the previous line.
I checked the code base and it seems we do not have a consensus here.. and personally I would prefer this as it actually make the logic clearer. > On Feb. 4, 2015, 2:15 a.m., Joel Koshy wrote: > > core/src/main/scala/kafka/common/OffsetMetadataAndError.scala, line 36 > > <https://reviews.apache.org/r/27391/diff/11/?file=832425#file832425line36> > > > > (This is also a public API change - although you did add an Object > > wrapper further down that comes close to the original API.) I think the wrapper MessageAndMetadata preserves the existing public API right? > On Feb. 4, 2015, 2:15 a.m., Joel Koshy wrote: > > core/src/main/scala/kafka/server/OffsetManager.scala, line 65 > > <https://reviews.apache.org/r/27391/diff/11/?file=832430#file832430line65> > > > > Should we call this maxOffsetRetentionMs instead? Not exactly, as it is just the default offset retention, not the upper limit: users can specify a value larger than this default and it will still be accepted. > On Feb. 4, 2015, 2:15 a.m., Joel Koshy wrote: > > core/src/main/scala/kafka/server/KafkaApis.scala, line 163 > > <https://reviews.apache.org/r/27391/diff/11/?file=832427#file832427line163> > > > > Shouldn't the commit timestamp _always_ be set to the current time? > > > > What I was thinking is this: > > If v0: > > - An explicit timestamp is provided only to override the v0 default > > retention which is add the server-side retention to the current timestamp. > > The (true) commit timestamp - i.e., receive time is useful for debugging > > purposes. So if an explicit timestamp is provided in v0 then use that to > > compute the absolute expire timestamp which will be the given commit > > timestamp; so you would store (commitTimestamp = now; expireTimestamp = > > given commitTimeStamp); if v0 and commit timestamp is default, then you > > would store (commitTimestamp = now, expireTimestamp = now + offsetRetention) > > - if v1: (commitTimestamp = now, expireTimestamp = now + > > offsetRetention) > > > > This way, you should have correct expiration behavior for v0, v1 and v2 > > and at the same time have the true commit timestamp - i.e., the receive > > time at the broker which is useful for debugging. (also see comment in > > OffsetManager) In v0/v1, the commit timestamp can be specified as a future timestamp so the expiration timestamp = commit timestamp + retention (in v0/v1 it is always the default value). This behavior should not be respected, i.e. offsets already stored in v0 and v1 format should be expired correctly using 0.8.2 code. Details can be found in Jun's comments and my replies. > On Feb. 4, 2015, 2:15 a.m., Joel Koshy wrote: > > core/src/main/scala/kafka/server/OffsetManager.scala, line 557 > > <https://reviews.apache.org/r/27391/diff/11/?file=832430#file832430line557> > > > > follow-up from above comment... > > and here you would set: > > commitTimestamp = timestamp > > expireTimestamp = timestamp > > > > So do you think this would work overall? > > > > I could be wrong - this patch has proven to be much trickier than we > > originally thought. See the comments above. - Guozhang ----------------------------------------------------------- This is an automatically generated e-mail. To reply, visit: https://reviews.apache.org/r/27391/#review70836 ----------------------------------------------------------- On Jan. 24, 2015, 12:06 a.m., Guozhang Wang wrote: > > ----------------------------------------------------------- > This is an automatically generated e-mail. To reply, visit: > https://reviews.apache.org/r/27391/ > ----------------------------------------------------------- > > (Updated Jan. 24, 2015, 12:06 a.m.) > > > Review request for kafka. > > > Bugs: KAFKA-1634 > https://issues.apache.org/jira/browse/KAFKA-1634 > > > Repository: kafka > > > Description > ------- > > Incorporated Jun's comments > > > Diffs > ----- > > clients/src/main/java/org/apache/kafka/common/protocol/Protocol.java > 7517b879866fc5dad5f8d8ad30636da8bbe7784a > clients/src/main/java/org/apache/kafka/common/protocol/types/Struct.java > 121e880a941fcd3e6392859edba11a94236494cc > > clients/src/main/java/org/apache/kafka/common/requests/OffsetCommitRequest.java > 3ee5cbad55ce836fd04bb954dcf6ef2f9bc3288f > > clients/src/test/java/org/apache/kafka/common/requests/RequestResponseTest.java > df37fc6d8f0db0b8192a948426af603be3444da4 > core/src/main/scala/kafka/api/OffsetCommitRequest.scala > 050615c72efe7dbaa4634f53943bd73273d20ffb > core/src/main/scala/kafka/api/OffsetFetchRequest.scala > c7604b9cdeb8f46507f04ed7d35fcb3d5f150713 > core/src/main/scala/kafka/common/OffsetMetadataAndError.scala > 4cabffeacea09a49913505db19a96a55d58c0909 > core/src/main/scala/kafka/consumer/ZookeeperConsumerConnector.scala > 5487259751ebe19f137948249aa1fd2637d2deb4 > core/src/main/scala/kafka/server/KafkaApis.scala > ec8d9f7ba44741db40875458bd524c4062ad6a26 > core/src/main/scala/kafka/server/KafkaConfig.scala > 6d74983472249eac808d361344c58cc2858ec971 > core/src/main/scala/kafka/server/KafkaServer.scala > 89200da30a04943f0b9befe84ab17e62b747c8c4 > core/src/main/scala/kafka/server/OffsetManager.scala > 0bdd42fea931cddd072c0fff765b10526db6840a > core/src/main/scala/kafka/server/ReplicaManager.scala > e58fbb922e93b0c31dff04f187fcadb4ece986d7 > core/src/test/scala/unit/kafka/api/RequestResponseSerializationTest.scala > cd16ced5465d098be7a60498326b2a98c248f343 > core/src/test/scala/unit/kafka/server/OffsetCommitTest.scala > 5b93239cdc26b5be7696f4e7863adb9fbe5f0ed5 > core/src/test/scala/unit/kafka/server/ServerShutdownTest.scala > ba1e48e4300c9fb32e36e7266cb05294f2a481e5 > > Diff: https://reviews.apache.org/r/27391/diff/ > > > Testing > ------- > > > Thanks, > > Guozhang Wang > >
