Skip to content

KAFKA-15272: Fix the logic which finds candidate log segments to upload it to tiered storage #14128

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
satishd merged 11 commits into from
Aug 2, 2023
Merged

KAFKA-15272: Fix the logic which finds candidate log segments to upload it to tiered storage #14128

Show file tree
Hide file tree
Changes from 5 commits
Commits
Show all changes
11 commits
Select commit Hold shift + click to select a range
df79e31
KAFKA-15272: Fix the logic which finds candidate log segments to uplo…
kamalcph Jul 29, 2023
c2cd2de
Avoid calling expensive LogSegment#readNextOffset() operation.
kamalcph Jul 31, 2023
3dab4ce
Addressed the review comments.
kamalcph Jul 31, 2023
882f772
Addressed the review-2 comments.
kamalcph Jul 31, 2023
154c525
Added toString
kamalcph Jul 31, 2023
272443c
Renamed EnrichedLogSegment to CandidateLogSegment and addressed the r…
kamalcph Jul 31, 2023
c800c6c
Update the logger.
kamalcph Jul 31, 2023
63dd63f
Update the test parameter.
kamalcph Jul 31, 2023
fe6ccd3
Rename CandidateLogSegment to EnrichedLogSegment
kamalcph Aug 1, 2023
8fc4e2e
Update the test
kamalcph Aug 1, 2023
c8f06a7
Update the test
kamalcph Aug 1, 2023
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
95 changes: 63 additions & 32 deletions core/src/main/java/kafka/log/remote/RemoteLogManager.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -84,13 +84,12 @@
import java.security.PrivilegedAction;
import java.util.ArrayList;
import java.util.Collections;
import java.util.Comparator;
import java.util.HashMap;
import java.util.HashSet;
import java.util.Iterator;
import java.util.List;
import java.util.ListIterator;
import java.util.Map;
import java.util.Objects;
import java.util.Optional;
import java.util.OptionalInt;
import java.util.OptionalLong;
Expand Down Expand Up @@ -525,6 +524,20 @@ private void maybeUpdateReadOffset(UnifiedLog log) throws RemoteStorageException
}
}

List<EnrichedLogSegment> enrichedLogSegments(UnifiedLog log, Long fromOffset) {
List<EnrichedLogSegment> enrichedLogSegments = new ArrayList<>();
List<LogSegment> segments = JavaConverters.seqAsJavaList(log.logSegments(fromOffset, Long.MAX_VALUE).toSeq());
if (!segments.isEmpty()) {
for (int idx = 1; idx < segments.size(); idx++) {
LogSegment previousSeg = segments.get(idx - 1);
LogSegment currentSeg = segments.get(idx);
enrichedLogSegments.add(new EnrichedLogSegment(previousSeg, currentSeg.baseOffset()));
}
// Discard the last active segment
}
return enrichedLogSegments;
}

public void copyLogSegmentsToRemote(UnifiedLog log) throws InterruptedException {
if (isCancelled())
return;
Expand All @@ -538,35 +551,33 @@ public void copyLogSegmentsToRemote(UnifiedLog log) throws InterruptedException
if (lso < 0) {
logger.warn("lastStableOffset for partition {} is {}, which should not be negative.", topicIdPartition, lso);
} else if (lso > 0 && copiedOffset < lso) {
// Copy segments only till the last-stable-offset as remote storage should contain only committed/acked
// messages
long toOffset = lso;
logger.debug("Checking for segments to copy, copiedOffset: {} and toOffset: {}", copiedOffset, toOffset);
long activeSegBaseOffset = log.activeSegment().baseOffset();
// log-start-offset can be ahead of the read-offset, when:
// 1) log-start-offset gets incremented via delete-records API (or)
// 2) enabling the remote log for the first time
long fromOffset = Math.max(copiedOffset + 1, log.logStartOffset());
ArrayList<LogSegment> sortedSegments = new ArrayList<>(JavaConverters.asJavaCollection(log.logSegments(fromOffset, toOffset)));
sortedSegments.sort(Comparator.comparingLong(LogSegment::baseOffset));
List<Long> sortedBaseOffsets = sortedSegments.stream().map(LogSegment::baseOffset).collect(Collectors.toList());
int activeSegIndex = Collections.binarySearch(sortedBaseOffsets, activeSegBaseOffset);

// sortedSegments becomes empty list when fromOffset and toOffset are same, and activeSegIndex becomes -1
if (activeSegIndex < 0) {
long activeSegmentBaseOffset = log.activeSegment().baseOffset();

// Segments which match the following criteria are eligible for copying to remote storage:
// 1) Segment is not the active segment and
// 2) Segment end-offset is less than the last-stable-offset as remote storage should contain only
// committed/acked messages
List<EnrichedLogSegment> candidateSegments = enrichedLogSegments(log, fromOffset)
.stream()
.filter(enrichedSegment -> enrichedSegment.logSegment.baseOffset() != activeSegmentBaseOffset && enrichedSegment.nextSegmentOffset <= lso)

@showuon showuon Jul 31, 2023

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

If a segment contains 0 records and log.roll.ms timeout passed, then what will be the baseOffset of next active segment? Do you have any idea? I asked because I'm not sure if enrichedSegment.logSegment.baseOffset() != activeSegmentBaseOffset is necessary. If we already filtered out the active segment in the enrichedLogSegments method, should we also need this check?

@kamalcph kamalcph Jul 31, 2023

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

If a segment contains 0 records and log.roll.ms timeout passed, then what will be the baseOffset of next active segment?

If the active segment is empty, then it won't be rotated.

The in-memory check enrichedSegment.logSegment.baseOffset() != activeSegmentBaseOffset is kind of redundant, we can remove it later when required.

@clolov clolov Jul 31, 2023

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Nit: Instead of doing this filtering after we have created the enriched segments list and iterating again can we not push the condition as part of creating the enriched segments list?

.collect(Collectors.toList());
logger.debug("Checking for segments to copy, copiedOffset: {} and lso: {}, candidateSegments: {}",
copiedOffset, lso, candidateSegments);
if (candidateSegments.isEmpty()) {
logger.debug("No segments found to be copied for partition {} with copiedOffset: {} and active segment's base-offset: {}",
topicIdPartition, copiedOffset, activeSegBaseOffset);
topicIdPartition, copiedOffset, activeSegmentBaseOffset);
} else {
ListIterator<LogSegment> logSegmentsIter = sortedSegments.subList(0, activeSegIndex).listIterator();
while (logSegmentsIter.hasNext()) {
LogSegment segment = logSegmentsIter.next();
for (EnrichedLogSegment enrichedLogSegment : candidateSegments) {
if (isCancelled() || !isLeader()) {
logger.info("Skipping copying log segments as the current task state is changed, cancelled: {} leader:{}",
isCancelled(), isLeader());
return;
}

copyLogSegment(log, segment, getNextSegmentBaseOffset(activeSegBaseOffset, logSegmentsIter));
copyLogSegment(log, enrichedLogSegment.logSegment, enrichedLogSegment.nextSegmentOffset);
}
}
} else {
Expand All @@ -583,18 +594,6 @@ public void copyLogSegmentsToRemote(UnifiedLog log) throws InterruptedException
}
}

private long getNextSegmentBaseOffset(long activeSegBaseOffset, ListIterator<LogSegment> logSegmentsIter) {
long nextSegmentBaseOffset;
if (logSegmentsIter.hasNext()) {
nextSegmentBaseOffset = logSegmentsIter.next().baseOffset();
logSegmentsIter.previous();
} else {
nextSegmentBaseOffset = activeSegBaseOffset;
}

return nextSegmentBaseOffset;
}

private void copyLogSegment(UnifiedLog log, LogSegment segment, long nextSegmentBaseOffset) throws InterruptedException, ExecutionException, RemoteStorageException, IOException {
File logFile = segment.log().file();
String logFileName = logFile.getName();
Expand Down Expand Up @@ -1002,4 +1001,36 @@ public void close() {
}
}

static class EnrichedLogSegment {
private final LogSegment logSegment;
private final long nextSegmentOffset;

public EnrichedLogSegment(LogSegment logSegment,
long nextSegmentOffset) {
this.logSegment = logSegment;
this.nextSegmentOffset = nextSegmentOffset;
}

@Override
public boolean equals(Object o) {
if (this == o) return true;
if (o == null || getClass() != o.getClass()) return false;
EnrichedLogSegment that = (EnrichedLogSegment) o;
return nextSegmentOffset == that.nextSegmentOffset && Objects.equals(logSegment, that.logSegment);
}

@Override
public int hashCode() {
return Objects.hash(logSegment, nextSegmentOffset);
}

@Override
public String toString() {
return "EnrichedLogSegment{" +
"logSegment=" + logSegment +
", nextSegmentOffset=" + nextSegmentOffset +
'}';
}
}

}
54 changes: 51 additions & 3 deletions core/src/test/java/kafka/log/remote/RemoteLogManagerTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -269,15 +269,37 @@ void testStartup() {
void testCopyLogSegmentsToRemoteShouldCopyExpectedLogSegment() throws Exception {
long oldSegmentStartOffset = 0L;
long nextSegmentStartOffset = 150L;
long oldSegmentEndOffset = nextSegmentStartOffset - 1;
long lso = 250L;
long leo = 300L;
assertCopyExpectedLogSegmentsToRemote(oldSegmentStartOffset, nextSegmentStartOffset, lso, leo);
}

/**
* The following values will be equal when the active segment gets rotated to passive and there are no new messages:
* last-stable-offset = high-water-mark = log-end-offset = base-offset-of-active-segment.
* This test asserts that the active log segment that was rotated after log.roll.ms are copied to remote storage.
*/
@Test
void testCopyLogSegmentToRemoteForStaleTopic() throws Exception {

@satishd satishd Aug 1, 2023

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Good test for the edge case of a stale topic with no activity for longer durations.

long oldSegmentStartOffset = 0L;
long nextSegmentStartOffset = 150L;
long lso = 150L;
long leo = 150L;
assertCopyExpectedLogSegmentsToRemote(oldSegmentStartOffset, nextSegmentStartOffset, lso, leo);
}

private void assertCopyExpectedLogSegmentsToRemote(long oldSegmentStartOffset,
long nextSegmentStartOffset,
long lastStableOffset,
long logEndOffset) throws Exception {
long oldSegmentEndOffset = nextSegmentStartOffset - 1;
when(mockLog.topicPartition()).thenReturn(leaderTopicIdPartition.topicPartition());

// leader epoch preparation
checkpoint.write(totalEpochEntries);
LeaderEpochFileCache cache = new LeaderEpochFileCache(leaderTopicIdPartition.topicPartition(), checkpoint);
when(mockLog.leaderEpochCache()).thenReturn(Option.apply(cache));
when(remoteLogMetadataManager.highestOffsetForEpoch(any(TopicIdPartition.class), anyInt())).thenReturn(Optional.of(0L));
when(remoteLogMetadataManager.highestOffsetForEpoch(any(TopicIdPartition.class), anyInt())).thenReturn(Optional.of(-1L));

File tempFile = TestUtils.tempFile();
File mockProducerSnapshotIndex = TestUtils.tempFile();
Expand All @@ -287,7 +309,9 @@ void testCopyLogSegmentsToRemoteShouldCopyExpectedLogSegment() throws Exception
LogSegment activeSegment = mock(LogSegment.class);

when(oldSegment.baseOffset()).thenReturn(oldSegmentStartOffset);
when(oldSegment.readNextOffset()).thenReturn(nextSegmentStartOffset);
when(activeSegment.baseOffset()).thenReturn(nextSegmentStartOffset);
when(activeSegment.readNextOffset()).thenReturn(logEndOffset);

FileRecords fileRecords = mock(FileRecords.class);
when(oldSegment.log()).thenReturn(fileRecords);
Expand All @@ -302,7 +326,7 @@ void testCopyLogSegmentsToRemoteShouldCopyExpectedLogSegment() throws Exception
ProducerStateManager mockStateManager = mock(ProducerStateManager.class);
when(mockLog.producerStateManager()).thenReturn(mockStateManager);
when(mockStateManager.fetchSnapshot(anyLong())).thenReturn(Optional.of(mockProducerSnapshotIndex));
when(mockLog.lastStableOffset()).thenReturn(250L);
when(mockLog.lastStableOffset()).thenReturn(lastStableOffset);

LazyIndex idx = LazyIndex.forOffset(UnifiedLog.offsetIndexFile(tempDir, oldSegmentStartOffset, ""), oldSegmentStartOffset, 1000);
LazyIndex timeIdx = LazyIndex.forTime(UnifiedLog.timeIndexFile(tempDir, oldSegmentStartOffset, ""), oldSegmentStartOffset, 1500);
Expand Down Expand Up @@ -823,6 +847,30 @@ public RemoteLogMetadataManager createRemoteLogMetadataManager() {
}
}

@Test
public void testEnrichedLogSegments() {
UnifiedLog log = mock(UnifiedLog.class);
LogSegment segment1 = mock(LogSegment.class);
LogSegment segment2 = mock(LogSegment.class);
LogSegment activeSegment = mock(LogSegment.class);

when(segment1.baseOffset()).thenReturn(5L);
when(segment2.baseOffset()).thenReturn(10L);
when(activeSegment.baseOffset()).thenReturn(15L);

when(log.logSegments(5L, Long.MAX_VALUE))
.thenReturn(JavaConverters.collectionAsScalaIterable(Arrays.asList(segment1, segment2, activeSegment)));

RemoteLogManager.RLMTask task = remoteLogManager.new RLMTask(leaderTopicIdPartition);
List<RemoteLogManager.EnrichedLogSegment> expected =
Arrays.asList(
new RemoteLogManager.EnrichedLogSegment(segment1, 10L),
new RemoteLogManager.EnrichedLogSegment(segment2, 15L)
);
List<RemoteLogManager.EnrichedLogSegment> actual = task.enrichedLogSegments(log, 5L);
assertEquals(expected, actual);
}

private Partition mockPartition(TopicIdPartition topicIdPartition) {
TopicPartition tp = topicIdPartition.topicPartition();
Partition partition = mock(Partition.class);
Expand Down