junrao commented on code in PR #24028:
URL: https://github.com/apache/kafka/pull/24028#discussion_r4429324625
##########
storage/src/main/java/org/apache/kafka/server/log/remote/storage/RemoteLogManager.java:
##########
@@ -752,12 +752,20 @@ && 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);
+ Optional<FileRecords.TimestampAndOffset>
timestampAndOffset =
+ lookupTimestamp(rlsMetadata, timestamp,
startingOffset, maxRecordBodySize);
+ if (timestampAndOffset.isPresent()) {
+ return timestampAndOffset;
+ }
} else {
// search in local-log
for (LogSegment segment : segmentsCopy) {
if (segment.largestTimestamp() >= timestamp) {
- return
segment.findOffsetByTimestamp(timestamp, startingOffset, maxRecordBodySize);
+ Optional<FileRecords.TimestampAndOffset>
timestampAndOffset =
+
segment.findOffsetByTimestamp(timestamp, startingOffset, maxRecordBodySize);
Review Comment:
Claude found the following.
```
2. The local log is scanned again for every later remote segment.
What happens: once a remote segment starts at or after the first local
segment, the local branch searches the whole local log. If that finds nothing,
the PR moves on to the next remote segment. Later segments start at the same or
higher offsets, so each one takes the local branch too and searches the whole
local log again.
Why that's unnecessary: the local log holds that segment and everything
after it, so the local result is final, empty or not. Return it.
} else {
// The local log holds this segment and all later ones, so its result is
final
for (LogSegment segment : segmentsCopy) {
...
if (timestampAndOffset.isPresent())
return timestampAndOffset;
}
return Optional.empty();
}
```
##########
storage/src/main/java/org/apache/kafka/server/log/remote/storage/RemoteLogManager.java:
##########
@@ -752,12 +752,20 @@ && 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);
+ Optional<FileRecords.TimestampAndOffset>
timestampAndOffset =
+ lookupTimestamp(rlsMetadata, timestamp,
startingOffset, maxRecordBodySize);
Review Comment:
Claude found the following issue.
```
1. A remote segment spanning several leader epochs is searched once per
epoch.
Why: listRemoteLogSegments(tip, epoch) returns every segment that contains
that epoch, per its Javadoc ("which contains the given leader epoch"). So a
segment spanning epochs 10 and 11 is listed under both.
Why it's new: before this PR the method returned at the first segment that
could match, so seeing a segment twice didn't matter. Now an empty lookup
continues, and the same segment comes back in the next epoch's list.
Cost: each lookup calls lookupTimestamp again, which reads the remote
segment from the indexed position all the way to its end when nothing matches.
Confirmed: my test has the first segment span epochs 10 and 11, with its
only match below the starting offset. fetchLogSegment(firstSegment, 0) was
called twice. The result was still right (offset 124 from the later segment). A
segment spanning many epochs, after frequent leader changes, would be fetched
once per epoch.
Fix: remember how far the search has gone.
long searchedThrough = startingOffset - 1;
...
RemoteLogSegmentMetadata rlsMetadata = iterator.next();
// A segment with several leader epochs is listed once per epoch
if (rlsMetadata.endOffset() <= searchedThrough)
continue;
...
Optional<FileRecords.TimestampAndOffset> timestampAndOffset =
lookupTimestamp(rlsMetadata, timestamp, startingOffset,
maxRecordBodySize);
if (timestampAndOffset.isPresent())
return timestampAndOffset;
searchedThrough = rlsMetadata.endOffset();
```
##########
storage/src/test/java/org/apache/kafka/server/log/remote/storage/RemoteLogManagerTest.java:
##########
@@ -1678,9 +1678,86 @@ void testFindOffsetByTimestampWithSegmentNotReady()
throws IOException, RemoteSt
assertEquals(Optional.empty(), maybeTimestampAndOffset);
}
- private void doTestFindOffsetByTimestamp(long ts, long startOffset, int
targetLeaderEpoch,
- TreeMap<Integer, Long>
validSegmentEpochs,
- RemoteLogSegmentState state)
throws IOException, RemoteStorageException {
+ @ParameterizedTest
+ @ValueSource(booleans = {false, true})
+ void testFindOffsetByTimestampContinuesAfterEmptyRemoteSegment(boolean
nextEpoch) throws Exception {
+ // Register the partition without starting background tasks while the
lookup mocks are configured.
+ remoteLogManager = spy(remoteLogManager);
+
doNothing().when(remoteLogManager).doHandleLeaderPartition(eq(leaderTopicIdPartition),
eq(false));
+ long timestamp = time.milliseconds();
+ long baseOffset = 120L;
+ int leaderEpoch = 10;
+ int laterEpoch = nextEpoch ? leaderEpoch + 1 : leaderEpoch;
+ LeaderEpochFileCache cache = new
LeaderEpochFileCache(leaderTopicIdPartition.topicPartition(), checkpoint,
scheduler);
+ cache.assign(leaderEpoch, baseOffset);
+ if (nextEpoch) {
+ cache.assign(laterEpoch, baseOffset + 3);
+ }
+
+ RemoteLogSegmentMetadata firstSegment =
doTestFindOffsetByTimestamp(timestamp, baseOffset, leaderEpoch,
Review Comment:
Claude found the following existing issue.
`Nit: doTestFindOffsetByTimestamp now returns the segment metadata, so its
name no longer fits. Something like setUpRemoteSegmentForFindOffsetByTimestamp
would.`
##########
storage/src/main/java/org/apache/kafka/server/log/remote/storage/RemoteLogManager.java:
##########
@@ -752,12 +752,20 @@ && 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);
+ Optional<FileRecords.TimestampAndOffset>
timestampAndOffset =
+ lookupTimestamp(rlsMetadata, timestamp,
startingOffset, maxRecordBodySize);
+ if (timestampAndOffset.isPresent()) {
+ return timestampAndOffset;
+ }
} else {
// search in local-log
for (LogSegment segment : segmentsCopy) {
if (segment.largestTimestamp() >= timestamp) {
- return
segment.findOffsetByTimestamp(timestamp, startingOffset, maxRecordBodySize);
+ Optional<FileRecords.TimestampAndOffset>
timestampAndOffset =
+
segment.findOffsetByTimestamp(timestamp, startingOffset, maxRecordBodySize);
+ if (timestampAndOffset.isPresent()) {
+ return timestampAndOffset;
Review Comment:
Claude found the following.
`Reuse UnifiedLog.searchOffsetInLocalLog. The local loop here copies it.
#23880 fixes it the same way and also skips segments that end before the
starting offset. Making it public and calling
unifiedLog.searchOffsetInLocalLog(timestamp, startingOffset)`
--
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]
