junrao commented on code in PR #24028:
URL: https://github.com/apache/kafka/pull/24028#discussion_r4445610716
##########
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:
EmptyRemoteSegment is not accurate. Could we change the test name to match
what's in UnifiedLogTest? Ditto for
testFindOffsetByTimestampContinuesAfterEmptyLocalSegment.
##########
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:
`expired` is not very accurate. Retention always removes the full segment.
Only deleteRecords can move the start offset into the middle of a segment.
##########
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:
Claude found the following.
```
1. Use the same segment snapshot for the check and the search
(RemoteLogManager.java:757-768). segmentsCopy decides whether the remote
segment is covered by the local log, but searchOffsetInLocalLog takes a new
snapshot with logSegments().
If retention deletes the first local segments between the two calls, the
local log can start after rlsMetadata.startOffset(). The offsets in between are
then only in remote storage. The local search misses them and returns a later
offset, or nothing, as the final answer.
The window is tiny, but the old inline loop searched segmentsCopy, which
avoided it. An overload that takes the snapshot fixes it:
public Optional<FileRecords.TimestampAndOffset> searchOffsetInLocalLog(long
targetTimestamp, long startOffset) throws IOException {
// Cache to avoid race conditions.
return searchOffsetInLocalLog(logSegments(), targetTimestamp,
startOffset);
}
/**
* Find the first record at or after startOffset, in the given snapshot of
the local segments, whose timestamp is at
* least targetTimestamp. startOffset may be below the first segment.
*/
public Optional<FileRecords.TimestampAndOffset>
searchOffsetInLocalLog(List<LogSegment> segments, long targetTimestamp,
long
startOffset) throws IOException {
// ... the current body, without taking the snapshot
}
RemoteLogManager would then call
unifiedLog.searchOffsetInLocalLog(segmentsCopy, timestamp, startingOffset). The
two thenCallRealMethod() stubs in RemoteLogManagerTest would need to target the
three-argument overload.
```
##########
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:
Claude suggested the following.
`The skip comment in RemoteLogManager ("A segment spanning multiple leader
epochs is listed once for each epoch") explains why the skip is needed, but not
what it does. Something like "Skip segments already searched: a segment
spanning multiple leader epochs is listed once for each epoch" says both.`
--
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]
