You are viewing a plain text version of this content. The canonical link for it is here.
Posted to jira@kafka.apache.org by "shantanukawale-stripe (via GitHub)" <gi...@apache.org> on 2026/10/30 11:22:50 UTC
[PR] KAFKA-21317: Continue tiered timestamp search after empty segment lookups [kafka]
shantanukawale-stripe opened a new pull request, #24028:
URL: https://github.com/apache/kafka/pull/24028
`RemoteLogManager.findOffsetByTimestamp` returns immediately when an eligible remote segment or local segment yields no match. If that segment's maximum timestamp belongs to a record below the requested starting offset, this skips later eligible segments and can cause the caller to return a later local offset or no result.
Continue searching after an empty lookup in both paths. Add regressions for subsequent remote segments in the same and next leader epoch, and for subsequent local segments.
Follow-up to #23880 under [KAFKA-21317](https://issues.apache.org/jira/browse/KAFKA-21317).
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-21317: Continue timestamp search across local and remote log segments [kafka]
Posted by "shantanukawale-stripe (via GitHub) via jira" <ji...@kafka.apache.org>.
shantanukawale-stripe commented on PR #24028:
URL: https://github.com/apache/kafka/pull/24028#issuecomment-6598934333
@junrao Thanks, I've made those three changes and replied inline. All 312 tests in `UnifiedLogTest` and `RemoteLogManagerTest` pass on Java 17, along with the storage checks. Could you take another look?
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-21317: Continue timestamp search across local and remote log segments [kafka]
Posted by "shantanukawale-stripe (via GitHub) via jira" <ji...@kafka.apache.org>.
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-21317: Continue tiered timestamp search after empty segment lookups [kafka]
Posted by "shantanukawale-stripe (via GitHub) via jira" <ji...@kafka.apache.org>.
shantanukawale-stripe commented on PR #24028:
URL: https://github.com/apache/kafka/pull/24028#issuecomment-6537722134
@junrao I've pushed the fixes and replied to each comment. Both full test classes pass on Java 17 (310 tests), along with storage Checkstyle, Spotless, and SpotBugs. The two new regression cases also reproduce the repeated scans before the fix.
This branch now includes #23880 so it can reuse the corrected local lookup. Could you take another look when you have a chance?
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-21317: Continue tiered timestamp search after empty segment lookups [kafka]
Posted by "junrao (via GitHub) via jira" <ji...@kafka.apache.org>.
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-21317: Continue tiered timestamp search after empty segment lookups [kafka]
Posted by "junrao (via GitHub) via jira" <ji...@kafka.apache.org>.
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-21317: Continue timestamp search across local and remote log segments [kafka]
Posted by "shantanukawale-stripe (via GitHub) via jira" <ji...@kafka.apache.org>.
shantanukawale-stripe commented on code in PR #24028:
URL: https://github.com/apache/kafka/pull/24028#discussion_r4452381176
##########
storage/src/main/java/org/apache/kafka/storage/internals/log/UnifiedLog.java:
##########
@@ -1812,12 +1812,32 @@ public boolean isEmpty() {
return logStartOffset == logEndOffset();
}
- private Optional<FileRecords.TimestampAndOffset> searchOffsetInLocalLog(long targetTimestamp, long startOffset) throws IOException {
+ /**
+ * Find the first local record at or after startOffset whose timestamp is at least targetTimestamp.
+ */
+ public Optional<FileRecords.TimestampAndOffset> searchOffsetInLocalLog(long targetTimestamp, long startOffset) throws IOException {
Review Comment:
Done, made the two-argument helper private again.
##########
storage/src/main/java/org/apache/kafka/storage/internals/log/UnifiedLog.java:
##########
@@ -1812,12 +1812,32 @@ public boolean isEmpty() {
return logStartOffset == logEndOffset();
}
- private Optional<FileRecords.TimestampAndOffset> searchOffsetInLocalLog(long targetTimestamp, long startOffset) throws IOException {
+ /**
+ * Find the first local record at or after startOffset whose timestamp is at least targetTimestamp.
+ */
+ public Optional<FileRecords.TimestampAndOffset> searchOffsetInLocalLog(long targetTimestamp, long startOffset) throws IOException {
// Cache to avoid race conditions.
- List<LogSegment> segments = logSegments();
- for (LogSegment segment : segments) {
+ return searchOffsetInLocalLog(logSegments(), targetTimestamp, startOffset);
+ }
+
+ /**
+ * Find the first record at or after startOffset in the given snapshot of local segments whose timestamp is at
+ * least targetTimestamp. The starting offset may be below the first segment.
+ */
+ public Optional<FileRecords.TimestampAndOffset> searchOffsetInLocalLog(List<LogSegment> segments, long targetTimestamp,
Review Comment:
Added the requirement that the segments be sorted by base offset in ascending order.
##########
storage/src/test/java/org/apache/kafka/server/log/remote/storage/RemoteLogManagerTest.java:
##########
@@ -1672,15 +1673,144 @@ 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 testFindOffsetByTimestampSearchesNextRemoteSegmentWhenMatchIsBelowLogStartOffset(boolean nextEpoch, boolean spansEpochs) 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 + (spansEpochs ? 2 : 1) : leaderEpoch;
+ TreeMap<Integer, Long> firstSegmentEpochs = new TreeMap<>(Map.of(leaderEpoch, baseOffset));
+ LeaderEpochFileCache cache = new LeaderEpochFileCache(leaderTopicIdPartition.topicPartition(), checkpoint, scheduler);
+ cache.assign(leaderEpoch, baseOffset);
+ if (spansEpochs) {
+ firstSegmentEpochs.put(leaderEpoch + 1, baseOffset + 2);
+ cache.assign(leaderEpoch + 1, baseOffset + 2);
+ }
+ if (nextEpoch) {
+ cache.assign(laterEpoch, baseOffset + 3);
+ }
+
+ RemoteLogSegmentMetadata firstSegment = setUpRemoteSegmentForFindOffsetByTimestamp(timestamp, baseOffset, leaderEpoch,
+ firstSegmentEpochs, RemoteLogSegmentState.COPY_SEGMENT_FINISHED);
+ RemoteLogSegmentMetadata laterSegment = new RemoteLogSegmentMetadata(
+ new RemoteLogSegmentId(leaderTopicIdPartition, Uuid.randomUuid()), baseOffset + 3, baseOffset + 5,
+ timestamp + 2, brokerId, timestamp, 1024, Optional.empty(), RemoteLogSegmentState.COPY_SEGMENT_FINISHED,
+ new TreeMap<>(Map.of(laterEpoch, baseOffset + 3)));
+ when(remoteLogMetadataManager.listRemoteLogSegments(eq(leaderTopicIdPartition), anyInt()))
+ .thenAnswer(invocation -> {
+ int epoch = invocation.getArgument(1);
+ return (nextEpoch
+ ? (epoch < laterEpoch ? List.of(firstSegment) : List.of(laterSegment))
+ : List.of(firstSegment, laterSegment)).iterator();
+ });
+ // The first segment's maximum belongs to a record below the requested starting offset.
+ MemoryRecords firstBatch = MemoryRecords.withRecords(baseOffset, Compression.NONE, leaderEpoch,
+ new SimpleRecord(timestamp + 2, new byte[0]), new SimpleRecord(timestamp - 1, new byte[0]));
+ MemoryRecords secondBatch = MemoryRecords.withRecords(baseOffset + 2, Compression.NONE,
+ spansEpochs ? leaderEpoch + 1 : leaderEpoch, new SimpleRecord(timestamp - 1, new byte[0]));
+ ByteBuffer firstSegmentRecords = ByteBuffer.allocate(firstBatch.sizeInBytes() + secondBatch.sizeInBytes());
+ firstSegmentRecords.put(firstBatch.buffer()).put(secondBatch.buffer());
+ when(remoteStorageManager.fetchLogSegment(firstSegment, 0)).thenAnswer(invocation ->
+ new ByteArrayInputStream(firstSegmentRecords.array()));
+ when(remoteStorageManager.fetchLogSegment(laterSegment, 0)).thenAnswer(invocation ->
+ new ByteArrayInputStream(records(timestamp, baseOffset + 3, laterEpoch).buffer().array()));
+
+ Optional<FileRecords.TimestampAndOffset> result = remoteLogManager.findOffsetByTimestamp(
+ leaderTopicIdPartition.topicPartition(), timestamp, baseOffset + 1, cache);
+ verify(remoteStorageManager).fetchLogSegment(firstSegment, 0);
+ verify(remoteStorageManager).fetchLogSegment(laterSegment, 0);
+ assertEquals(Optional.of(new FileRecords.TimestampAndOffset(timestamp + 1, baseOffset + 4, Optional.of(laterEpoch))), result);
+ }
+
+ @ParameterizedTest
+ @ValueSource(booleans = {false, true})
+ void testFindOffsetByTimestampSearchesNextLocalSegmentWhenMatchIsBelowLogStartOffset(boolean laterMatch) throws Exception {
+ long timestamp = time.milliseconds();
+ int leaderEpoch = 10;
+ LeaderEpochFileCache cache = new LeaderEpochFileCache(leaderTopicIdPartition.topicPartition(), checkpoint, scheduler);
+ cache.assign(leaderEpoch, 0L);
+ RemoteLogSegmentMetadata metadata = new RemoteLogSegmentMetadata(
+ new RemoteLogSegmentId(leaderTopicIdPartition, Uuid.randomUuid()), 0L, 1L,
+ timestamp + 1, brokerId, timestamp, 1024, Optional.empty(), RemoteLogSegmentState.COPY_SEGMENT_FINISHED,
+ new TreeMap<>(Map.of(leaderEpoch, 0L)));
+ RemoteLogSegmentMetadata laterMetadata = new RemoteLogSegmentMetadata(
+ new RemoteLogSegmentId(leaderTopicIdPartition, Uuid.randomUuid()), 2L, 2L,
+ timestamp, brokerId, timestamp, 1024, Optional.empty(), RemoteLogSegmentState.COPY_SEGMENT_FINISHED,
+ new TreeMap<>(Map.of(leaderEpoch, 2L)));
+ when(remoteLogMetadataManager.listRemoteLogSegments(leaderTopicIdPartition, leaderEpoch))
+ .thenAnswer(invocation -> List.of(metadata, laterMetadata).iterator());
+ when(mockLog.logEndOffset()).thenReturn(3L);
+ remoteLogManager.onLeadershipChange(Set.of(), Set.of(mockPartition(leaderTopicIdPartition)), topicIds);
+
+ try (LogSegment firstSegment = spy(LogSegment.open(new File(logDir), 0L, mockLog.config(), time, 0, false));
+ LogSegment laterSegment = LogSegment.open(new File(logDir), 2L, mockLog.config(), time, 0, false)) {
+ firstSegment.append(1L, MemoryRecords.withRecords(0L, Compression.NONE, leaderEpoch,
+ new SimpleRecord(timestamp + 1, new byte[0]), new SimpleRecord(timestamp - 1, new byte[0])));
+ laterSegment.append(2L, MemoryRecords.withRecords(2L, Compression.NONE, leaderEpoch,
+ new SimpleRecord(laterMatch ? timestamp : timestamp - 1, new byte[0])));
+ when(mockLog.logSegments()).thenReturn(List.of(firstSegment, laterSegment));
+ when(mockLog.searchOffsetInLocalLog(anyList(), anyLong(), anyLong())).thenCallRealMethod();
+
+ Optional<FileRecords.TimestampAndOffset> expected = laterMatch
+ ? Optional.of(new FileRecords.TimestampAndOffset(timestamp, 2L, Optional.of(leaderEpoch)))
+ : Optional.empty();
+ assertEquals(expected, remoteLogManager.findOffsetByTimestamp(
+ leaderTopicIdPartition.topicPartition(), timestamp, 1L, cache));
+ verify(firstSegment).findOffsetByTimestamp(timestamp, 1L, mockLog.config().maxDecompressedMessageBytes());
+ verify(mockLog).searchOffsetInLocalLog(List.of(firstSegment, laterSegment), timestamp, 1L);
+ verify(remoteStorageManager, never()).fetchLogSegment(any(), anyInt());
+ }
+ }
+
+ @ParameterizedTest
+ @ValueSource(booleans = {false, true})
+ void testFindOffsetByTimestampUsesLocalSegmentSnapshot(boolean laterMatch) throws Exception {
+ remoteLogManager = spy(remoteLogManager);
+ doNothing().when(remoteLogManager).doHandleFollowerPartition(leaderTopicIdPartition);
Review Comment:
Added a comment explaining why: the test needs to register the partition without starting the follower's background remote-offset updates against its mocks.
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-21317: Continue tiered timestamp search after empty segment lookups [kafka]
Posted by "shantanukawale-stripe (via GitHub)" <gi...@apache.org>.
shantanukawale-stripe commented on PR #24028:
URL: https://github.com/apache/kafka/pull/24028#issuecomment-6521457174
Validation on Java 17:
- Full `RemoteLogManagerTest`: 101 tests passed.
- `:storage:checkstyleMain`, `:storage:checkstyleTest`, `:storage:spotlessCheck`, and `:storage:spotbugsMain` passed.
- Regression coverage uses actual remote record streams and local log segments. It covers a later remote match within the same leader epoch and across epochs, a later local match, and no matching local record.
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-21317: Continue tiered timestamp search after empty segment lookups [kafka]
Posted by "shantanukawale-stripe (via GitHub) via jira" <ji...@kafka.apache.org>.
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-21317: Continue timestamp search across local and remote log segments [kafka]
Posted by "shantanukawale-stripe (via GitHub) via jira" <ji...@kafka.apache.org>.
shantanukawale-stripe commented on PR #24028:
URL: https://github.com/apache/kafka/pull/24028#issuecomment-6584573433
@junrao Pushed the snapshot fix and the test-name and comment updates. The new regression reproduces both failure modes before the fix and passes now. All 312 tests in `RemoteLogManagerTest` and `UnifiedLogTest` pass on Java 17, along with storage Checkstyle, Spotless, and SpotBugs.
Could you take another look when you have a chance?
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-21317: Continue timestamp search across local and remote log segments [kafka]
Posted by "junrao (via GitHub) via jira" <ji...@kafka.apache.org>.
junrao commented on code in PR #24028:
URL: https://github.com/apache/kafka/pull/24028#discussion_r4448750143
##########
storage/src/main/java/org/apache/kafka/storage/internals/log/UnifiedLog.java:
##########
@@ -1812,12 +1812,32 @@ public boolean isEmpty() {
return logStartOffset == logEndOffset();
}
- private Optional<FileRecords.TimestampAndOffset> searchOffsetInLocalLog(long targetTimestamp, long startOffset) throws IOException {
+ /**
+ * Find the first local record at or after startOffset whose timestamp is at least targetTimestamp.
+ */
+ public Optional<FileRecords.TimestampAndOffset> searchOffsetInLocalLog(long targetTimestamp, long startOffset) throws IOException {
Review Comment:
Could we keep this one as private?
##########
storage/src/main/java/org/apache/kafka/storage/internals/log/UnifiedLog.java:
##########
@@ -1812,12 +1812,32 @@ public boolean isEmpty() {
return logStartOffset == logEndOffset();
}
- private Optional<FileRecords.TimestampAndOffset> searchOffsetInLocalLog(long targetTimestamp, long startOffset) throws IOException {
+ /**
+ * Find the first local record at or after startOffset whose timestamp is at least targetTimestamp.
+ */
+ public Optional<FileRecords.TimestampAndOffset> searchOffsetInLocalLog(long targetTimestamp, long startOffset) throws IOException {
// Cache to avoid race conditions.
- List<LogSegment> segments = logSegments();
- for (LogSegment segment : segments) {
+ return searchOffsetInLocalLog(logSegments(), targetTimestamp, startOffset);
+ }
+
+ /**
+ * Find the first record at or after startOffset in the given snapshot of local segments whose timestamp is at
+ * least targetTimestamp. The starting offset may be below the first segment.
+ */
+ public Optional<FileRecords.TimestampAndOffset> searchOffsetInLocalLog(List<LogSegment> segments, long targetTimestamp,
Review Comment:
Could we add that `segments` need to be sorted in base offset order?
##########
storage/src/test/java/org/apache/kafka/server/log/remote/storage/RemoteLogManagerTest.java:
##########
@@ -1672,15 +1673,144 @@ 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 testFindOffsetByTimestampSearchesNextRemoteSegmentWhenMatchIsBelowLogStartOffset(boolean nextEpoch, boolean spansEpochs) 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 + (spansEpochs ? 2 : 1) : leaderEpoch;
+ TreeMap<Integer, Long> firstSegmentEpochs = new TreeMap<>(Map.of(leaderEpoch, baseOffset));
+ LeaderEpochFileCache cache = new LeaderEpochFileCache(leaderTopicIdPartition.topicPartition(), checkpoint, scheduler);
+ cache.assign(leaderEpoch, baseOffset);
+ if (spansEpochs) {
+ firstSegmentEpochs.put(leaderEpoch + 1, baseOffset + 2);
+ cache.assign(leaderEpoch + 1, baseOffset + 2);
+ }
+ if (nextEpoch) {
+ cache.assign(laterEpoch, baseOffset + 3);
+ }
+
+ RemoteLogSegmentMetadata firstSegment = setUpRemoteSegmentForFindOffsetByTimestamp(timestamp, baseOffset, leaderEpoch,
+ firstSegmentEpochs, RemoteLogSegmentState.COPY_SEGMENT_FINISHED);
+ RemoteLogSegmentMetadata laterSegment = new RemoteLogSegmentMetadata(
+ new RemoteLogSegmentId(leaderTopicIdPartition, Uuid.randomUuid()), baseOffset + 3, baseOffset + 5,
+ timestamp + 2, brokerId, timestamp, 1024, Optional.empty(), RemoteLogSegmentState.COPY_SEGMENT_FINISHED,
+ new TreeMap<>(Map.of(laterEpoch, baseOffset + 3)));
+ when(remoteLogMetadataManager.listRemoteLogSegments(eq(leaderTopicIdPartition), anyInt()))
+ .thenAnswer(invocation -> {
+ int epoch = invocation.getArgument(1);
+ return (nextEpoch
+ ? (epoch < laterEpoch ? List.of(firstSegment) : List.of(laterSegment))
+ : List.of(firstSegment, laterSegment)).iterator();
+ });
+ // The first segment's maximum belongs to a record below the requested starting offset.
+ MemoryRecords firstBatch = MemoryRecords.withRecords(baseOffset, Compression.NONE, leaderEpoch,
+ new SimpleRecord(timestamp + 2, new byte[0]), new SimpleRecord(timestamp - 1, new byte[0]));
+ MemoryRecords secondBatch = MemoryRecords.withRecords(baseOffset + 2, Compression.NONE,
+ spansEpochs ? leaderEpoch + 1 : leaderEpoch, new SimpleRecord(timestamp - 1, new byte[0]));
+ ByteBuffer firstSegmentRecords = ByteBuffer.allocate(firstBatch.sizeInBytes() + secondBatch.sizeInBytes());
+ firstSegmentRecords.put(firstBatch.buffer()).put(secondBatch.buffer());
+ when(remoteStorageManager.fetchLogSegment(firstSegment, 0)).thenAnswer(invocation ->
+ new ByteArrayInputStream(firstSegmentRecords.array()));
+ when(remoteStorageManager.fetchLogSegment(laterSegment, 0)).thenAnswer(invocation ->
+ new ByteArrayInputStream(records(timestamp, baseOffset + 3, laterEpoch).buffer().array()));
+
+ Optional<FileRecords.TimestampAndOffset> result = remoteLogManager.findOffsetByTimestamp(
+ leaderTopicIdPartition.topicPartition(), timestamp, baseOffset + 1, cache);
+ verify(remoteStorageManager).fetchLogSegment(firstSegment, 0);
+ verify(remoteStorageManager).fetchLogSegment(laterSegment, 0);
+ assertEquals(Optional.of(new FileRecords.TimestampAndOffset(timestamp + 1, baseOffset + 4, Optional.of(laterEpoch))), result);
+ }
+
+ @ParameterizedTest
+ @ValueSource(booleans = {false, true})
+ void testFindOffsetByTimestampSearchesNextLocalSegmentWhenMatchIsBelowLogStartOffset(boolean laterMatch) throws Exception {
+ long timestamp = time.milliseconds();
+ int leaderEpoch = 10;
+ LeaderEpochFileCache cache = new LeaderEpochFileCache(leaderTopicIdPartition.topicPartition(), checkpoint, scheduler);
+ cache.assign(leaderEpoch, 0L);
+ RemoteLogSegmentMetadata metadata = new RemoteLogSegmentMetadata(
+ new RemoteLogSegmentId(leaderTopicIdPartition, Uuid.randomUuid()), 0L, 1L,
+ timestamp + 1, brokerId, timestamp, 1024, Optional.empty(), RemoteLogSegmentState.COPY_SEGMENT_FINISHED,
+ new TreeMap<>(Map.of(leaderEpoch, 0L)));
+ RemoteLogSegmentMetadata laterMetadata = new RemoteLogSegmentMetadata(
+ new RemoteLogSegmentId(leaderTopicIdPartition, Uuid.randomUuid()), 2L, 2L,
+ timestamp, brokerId, timestamp, 1024, Optional.empty(), RemoteLogSegmentState.COPY_SEGMENT_FINISHED,
+ new TreeMap<>(Map.of(leaderEpoch, 2L)));
+ when(remoteLogMetadataManager.listRemoteLogSegments(leaderTopicIdPartition, leaderEpoch))
+ .thenAnswer(invocation -> List.of(metadata, laterMetadata).iterator());
+ when(mockLog.logEndOffset()).thenReturn(3L);
+ remoteLogManager.onLeadershipChange(Set.of(), Set.of(mockPartition(leaderTopicIdPartition)), topicIds);
+
+ try (LogSegment firstSegment = spy(LogSegment.open(new File(logDir), 0L, mockLog.config(), time, 0, false));
+ LogSegment laterSegment = LogSegment.open(new File(logDir), 2L, mockLog.config(), time, 0, false)) {
+ firstSegment.append(1L, MemoryRecords.withRecords(0L, Compression.NONE, leaderEpoch,
+ new SimpleRecord(timestamp + 1, new byte[0]), new SimpleRecord(timestamp - 1, new byte[0])));
+ laterSegment.append(2L, MemoryRecords.withRecords(2L, Compression.NONE, leaderEpoch,
+ new SimpleRecord(laterMatch ? timestamp : timestamp - 1, new byte[0])));
+ when(mockLog.logSegments()).thenReturn(List.of(firstSegment, laterSegment));
+ when(mockLog.searchOffsetInLocalLog(anyList(), anyLong(), anyLong())).thenCallRealMethod();
+
+ Optional<FileRecords.TimestampAndOffset> expected = laterMatch
+ ? Optional.of(new FileRecords.TimestampAndOffset(timestamp, 2L, Optional.of(leaderEpoch)))
+ : Optional.empty();
+ assertEquals(expected, remoteLogManager.findOffsetByTimestamp(
+ leaderTopicIdPartition.topicPartition(), timestamp, 1L, cache));
+ verify(firstSegment).findOffsetByTimestamp(timestamp, 1L, mockLog.config().maxDecompressedMessageBytes());
+ verify(mockLog).searchOffsetInLocalLog(List.of(firstSegment, laterSegment), timestamp, 1L);
+ verify(remoteStorageManager, never()).fetchLogSegment(any(), anyInt());
+ }
+ }
+
+ @ParameterizedTest
+ @ValueSource(booleans = {false, true})
+ void testFindOffsetByTimestampUsesLocalSegmentSnapshot(boolean laterMatch) throws Exception {
+ remoteLogManager = spy(remoteLogManager);
+ doNothing().when(remoteLogManager).doHandleFollowerPartition(leaderTopicIdPartition);
Review Comment:
Could we add a comment why this stub is needed?
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-21317: Continue timestamp search across local and remote log segments [kafka]
Posted by "junrao (via GitHub) via jira" <ji...@kafka.apache.org>.
junrao merged PR #24028:
URL: https://github.com/apache/kafka/pull/24028
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org