shantanukawale-stripe commented on code in PR #24028:
URL: https://github.com/apache/kafka/pull/24028#discussion_r4429959487
##########
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:
Good catch. I added `searchedThrough` and only advance it after an empty
remote lookup. The regression now includes a segment spanning two epochs and
verifies that it is fetched once before the search finds the later match.
##########
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:
Agreed. Once the search reaches the local log, it now returns that result
directly, including an empty result. The regression includes a second eligible
remote metadata entry and verifies that the local search only runs once.
##########
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:
Done. `searchOffsetInLocalLog` is now public, and this path calls it instead
of keeping a second loop. I've brought in #23880 as a prerequisite so it uses
the corrected lookup and skips segments before the starting offset.
##########
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:
Renamed it to `setUpRemoteSegmentForFindOffsetByTimestamp`.
--
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]
