nileshkumar3 commented on code in PR #23780:
URL: https://github.com/apache/kafka/pull/23780#discussion_r4437124989
##########
storage/src/main/java/org/apache/kafka/storage/internals/log/LogSegment.java:
##########
@@ -750,12 +750,121 @@ public long getFirstBatchTimestamp() {
* @return the timestamp and offset of the first message that meets the
requirements. Empty will be returned if there is no such message.
*/
public Optional<FileRecords.TimestampAndOffset> findOffsetByTimestamp(long
timestampMs, long startingOffset, int maxRecordBodySize) throws IOException {
+ return findOffsetByTimestamp(timestampMs, startingOffset,
Long.MAX_VALUE, maxRecordBodySize);
+ }
+
+ /**
+ * Search for the first message whose timestamp is at least {@code
timestampMs} and whose offset is at least
+ * {@code startingOffset} and strictly less than {@code
exclusiveEndOffset}.
+ * {@code exclusiveEndOffset} is a batch boundary, such as the high
watermark or last stable offset.
+ * Pass {@link Long#MAX_VALUE} to search the whole segment.
+ */
+ public Optional<FileRecords.TimestampAndOffset> findOffsetByTimestamp(long
timestampMs, long startingOffset,
+ long
exclusiveEndOffset, int maxRecordBodySize) throws IOException {
+ if (exclusiveEndOffset <= baseOffset)
+ return Optional.empty();
+
// Get the index entry with a timestamp less than or equal to the
target timestamp
TimestampOffset timestampOffset = timeIndex().lookup(timestampMs);
int position = offsetIndex().lookup(Math.max(timestampOffset.offset(),
startingOffset)).position();
+ int endPosition = endPosition(exclusiveEndOffset);
+ if (position >= endPosition)
+ return Optional.empty();
+
+ return Optional.ofNullable(log.searchForTimestamp(timestampMs,
position, endPosition, startingOffset, maxRecordBodySize));
+ }
+
+ /**
+ * Largest timestamp in this segment, and the offset of the record that
holds it.
+ * Empty when the segment has no batches or the batch does not expose a
timestamp offset.
+ */
+ public Optional<FileRecords.TimestampAndOffset>
fetchMaxTimestampAndOffset(int maxRecordBodySize) throws IOException {
Review Comment:
Done. It's now a private helper.
--
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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
