showuon commented on code in PR #15474: URL: https://github.com/apache/kafka/pull/15474#discussion_r1520857036
########## clients/src/main/java/org/apache/kafka/common/record/MemoryRecordsBuilder.java: ########## @@ -263,13 +262,8 @@ public RecordsInfo info() { } else if (maxTimestamp == RecordBatch.NO_TIMESTAMP) { return new RecordsInfo(RecordBatch.NO_TIMESTAMP, lastOffset); } else { - long shallowOffsetOfMaxTimestamp; - // Use the last offset when dealing with record batches - if (compressionType != CompressionType.NONE || magic >= RecordBatch.MAGIC_VALUE_V2) - shallowOffsetOfMaxTimestamp = lastOffset; - else - shallowOffsetOfMaxTimestamp = offsetOfMaxTimestamp; - return new RecordsInfo(maxTimestamp, shallowOffsetOfMaxTimestamp); + // For create time, we always use offsetOfMaxTimestamp for the correct time -> offset mapping + return new RecordsInfo(maxTimestamp, offsetOfMaxTimestamp); Review Comment: > This has the same issue. The semantic for MAX_TIMESTAMP is the first offset with the max timestamp. So, if timestamp is TimestampType.LOG_APPEND_TIME, we need to use the baseOffset, instead of lastOffset. Yes, updated. > Also, could we remove the shallow part in RecordsInfo.shallowOffsetOfMaxTimestamp? I've added a comment in another PR: https://github.com/apache/kafka/pull/15476#issuecomment-1990459827 . We can do all the renaming there. ########## clients/src/main/java/org/apache/kafka/common/record/MemoryRecordsBuilder.java: ########## @@ -263,13 +262,8 @@ public RecordsInfo info() { } else if (maxTimestamp == RecordBatch.NO_TIMESTAMP) { return new RecordsInfo(RecordBatch.NO_TIMESTAMP, lastOffset); } else { - long shallowOffsetOfMaxTimestamp; - // Use the last offset when dealing with record batches - if (compressionType != CompressionType.NONE || magic >= RecordBatch.MAGIC_VALUE_V2) - shallowOffsetOfMaxTimestamp = lastOffset; - else - shallowOffsetOfMaxTimestamp = offsetOfMaxTimestamp; - return new RecordsInfo(maxTimestamp, shallowOffsetOfMaxTimestamp); + // For create time, we always use offsetOfMaxTimestamp for the correct time -> offset mapping + return new RecordsInfo(maxTimestamp, offsetOfMaxTimestamp); Review Comment: > This has the same issue. The semantic for MAX_TIMESTAMP is the first offset with the max timestamp. So, if timestamp is TimestampType.LOG_APPEND_TIME, we need to use the baseOffset, instead of lastOffset. Yes, updated. > Also, could we remove the shallow part in RecordsInfo.shallowOffsetOfMaxTimestamp? I've added a comment in another PR: https://github.com/apache/kafka/pull/15476#issuecomment-1990459827 . We can do all the renaming there. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: jira-unsubscr...@kafka.apache.org For queries about this service, please contact Infrastructure at: us...@infra.apache.org