shantanukawale-stripe commented on code in PR #24028:
URL: https://github.com/apache/kafka/pull/24028#discussion_r4445947455
##########
storage/src/test/java/org/apache/kafka/server/log/remote/storage/RemoteLogManagerTest.java:
##########
@@ -1672,15 +1672,109 @@ void testFindOffsetByTimestampWithSegmentNotReady()
throws IOException, RemoteSt
leaderEpochFileCache.assign(targetLeaderEpoch, startOffset);
leaderEpochFileCache.assign(12, 500L);
- doTestFindOffsetByTimestamp(ts, startOffset, targetLeaderEpoch,
validSegmentEpochs, RemoteLogSegmentState.COPY_SEGMENT_STARTED);
+ setUpRemoteSegmentForFindOffsetByTimestamp(ts, startOffset,
targetLeaderEpoch, validSegmentEpochs,
RemoteLogSegmentState.COPY_SEGMENT_STARTED);
Optional<FileRecords.TimestampAndOffset> maybeTimestampAndOffset =
remoteLogManager.findOffsetByTimestamp(tp, ts, startOffset,
leaderEpochFileCache);
assertEquals(Optional.empty(), maybeTimestampAndOffset);
}
- private void doTestFindOffsetByTimestamp(long ts, long startOffset, int
targetLeaderEpoch,
- TreeMap<Integer, Long>
validSegmentEpochs,
- RemoteLogSegmentState state)
throws IOException, RemoteStorageException {
+ @ParameterizedTest
+ @CsvSource({"false, false", "true, false", "true, true"})
+ void testFindOffsetByTimestampContinuesAfterEmptyRemoteSegment(boolean
nextEpoch, boolean spansEpochs) throws Exception {
Review Comment:
You're right, the segments aren't empty. I've renamed both tests to describe
the match being below the log start offset, in line with `UnifiedLogTest`.
##########
storage/src/main/java/org/apache/kafka/server/log/remote/storage/RemoteLogManager.java:
##########
@@ -752,14 +757,15 @@ && isRemoteSegmentWithinLeaderEpochs(rlsMetadata,
unifiedLog.logEndOffset(), epo
List<LogSegment> segmentsCopy = unifiedLog.logSegments();
if (segmentsCopy.isEmpty() || rlsMetadata.startOffset() <
segmentsCopy.get(0).baseOffset()) {
// search in remote-log
- return lookupTimestamp(rlsMetadata, timestamp,
startingOffset, maxRecordBodySize);
- } else {
- // search in local-log
- for (LogSegment segment : segmentsCopy) {
- if (segment.largestTimestamp() >= timestamp) {
- return
segment.findOffsetByTimestamp(timestamp, startingOffset, maxRecordBodySize);
- }
+ Optional<FileRecords.TimestampAndOffset>
timestampAndOffset =
+ lookupTimestamp(rlsMetadata, timestamp,
startingOffset, maxRecordBodySize);
+ if (timestampAndOffset.isPresent()) {
+ return timestampAndOffset;
}
+ searchedThrough = rlsMetadata.endOffset();
+ } else {
+ // The local log covers this segment and all later
offsets, so its result is final.
+ return unifiedLog.searchOffsetInLocalLog(timestamp,
startingOffset);
Review Comment:
You're right. The second snapshot can drop the segment we just used to
choose local storage. I added the overload and now pass `segmentsCopy` through.
The new regression changes the local segment list between calls: before the fix
it returned either a later offset or nothing; both cases now return the
original match.
##########
storage/src/test/java/org/apache/kafka/storage/internals/log/UnifiedLogTest.java:
##########
@@ -3542,6 +3542,29 @@ public void
testFetchOffsetByTimestampIncludesLeaderEpoch() throws IOException {
log.fetchOffsetByTimestamp(ListOffsetsRequest.LATEST_TIMESTAMP,
Optional.empty()));
}
+ @Test
+ public void
testFetchOffsetByTimestampSearchesNextSegmentWhenMatchIsBelowLogStartOffset()
throws IOException {
+ long targetTimestamp = mockTime.milliseconds();
+ LogConfig logConfig = new LogTestUtils.LogConfigBuilder()
+ .indexIntervalBytes(1)
+ .build();
+ log = createLog(logDir, logConfig);
+
+ // The expired record makes the first segment eligible, but its only
retained record has a lower timestamp.
Review Comment:
Agreed. Changed the record label to `deleted-max` and made the comment
explicit that `DeleteRecords` moves the start offset into the segment.
##########
storage/src/main/java/org/apache/kafka/server/log/remote/storage/RemoteLogManager.java:
##########
@@ -737,13 +737,18 @@ public Optional<FileRecords.TimestampAndOffset>
findOffsetByTimestamp(TopicParti
OptionalInt maybeEpoch =
leaderEpochCache.epochForOffset(startingOffset);
TopicIdPartition topicIdPartition = new TopicIdPartition(topicId, tp);
NavigableMap<Integer, Long> epochWithOffsets =
buildFilteredLeaderEpochMap(leaderEpochCache.epochWithOffsets());
+ long searchedThrough = startingOffset - 1;
while (maybeEpoch.isPresent()) {
int epoch = maybeEpoch.getAsInt();
// KAFKA-15802: Add a new API for RLMM to choose how to implement
the predicate.
// currently, all segments are returned and then iterated, and
filtered
Iterator<RemoteLogSegmentMetadata> iterator =
remoteLogMetadataManagerPlugin.get().listRemoteLogSegments(topicIdPartition,
epoch);
while (iterator.hasNext()) {
RemoteLogSegmentMetadata rlsMetadata = iterator.next();
+ // A segment spanning multiple leader epochs is listed once
for each epoch.
Review Comment:
Updated the comment to start with "Skip segments already searched" and kept
the explanation about segments spanning multiple epochs.
--
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]
