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 2 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
90 changes: 59 additions & 31 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,26 @@ private void maybeUpdateReadOffset(UnifiedLog log) throws RemoteStorageException
}
}

List<EnrichedLogSegment> enrichedLogSegments(UnifiedLog log, Long fromOffset, Long lastStableOffset) {
List<EnrichedLogSegment> enrichedLogSegments = new ArrayList<>();
List<LogSegment> segments = JavaConverters.seqAsJavaList(log.nonActiveLogSegmentsFrom(fromOffset).toSeq());
if (!segments.isEmpty()) {
int idx = 1;
for (; idx < segments.size(); idx++) {
LogSegment previous = segments.get(idx - 1);
LogSegment current = segments.get(idx);
enrichedLogSegments.add(new EnrichedLogSegment(previous, current.baseOffset()));
}
// LogSegment#readNextOffset() is an expensive call, so we only call it when necessary.
int lastIdx = idx - 1;
if (segments.get(lastIdx).baseOffset() < lastStableOffset) {
LogSegment last = segments.get(lastIdx);
enrichedLogSegments.add(new EnrichedLogSegment(last, last.readNextOffset()));
}

@showuon showuon Jul 31, 2023 •
edited
Loading

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.

Why can't we use the next segment's baseOffset here like the above did?
Also, I'm not very familiar with this, what's the difference between readNextOffset and baseOffset of next segment?

@kamalcph kamalcph Jul 31, 2023 •
edited
Loading

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.

For a given LogSegment, we know about start-offset (base-offset) but not the end offset. readNextOffset denotes end-offset-of-that-segment + 1.

To exclude the active segment, we are using log.nonActiveLogSegmentsFrom. With this, we cannot use the active-segment-base-offset as the operations are not done atomically. (the active segment might gets rotated in the mean time).

If we want to avoid LogSegment#nextReadOffset operation altogether, then we can list all the segments including the active and discard/filter the final entry.

@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 we want to avoid LogSegment#nextReadOffset operation altogether, then we can list all the segments including the active and discard/filter the final entry.

Yes, that's my thought, given the LogSegment#nextReadOffset is expensive. Any drawbacks for this solution?

@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.

Updated

}
return enrichedLogSegments;
}

public void copyLogSegmentsToRemote(UnifiedLog log) throws InterruptedException {
if (isCancelled())
return;
Expand All @@ -538,35 +557,32 @@ 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) {
// 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, lso)
.stream()
.filter(enriched -> enriched.segment.readNextOffset() <= lso)
.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, log.activeSegment().baseOffset());
} 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.segment, enrichedLogSegment.readNextOffset);
}
}
} else {
Expand All @@ -583,18 +599,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 +1006,28 @@ public void close() {
}
}

static class EnrichedLogSegment {
private final LogSegment segment;
private final long readNextOffset;

public EnrichedLogSegment(LogSegment segment,
long readNextOffset) {
this.segment = segment;
this.readNextOffset = readNextOffset;
}

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

@Override
public int hashCode() {
return Objects.hash(segment, readNextOffset);
}
}

}
90 changes: 83 additions & 7 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,38 @@ 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 when there are no new messages:

@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.

nit: 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 +310,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 @@ -297,12 +322,13 @@ void testCopyLogSegmentsToRemoteShouldCopyExpectedLogSegment() throws Exception

when(mockLog.activeSegment()).thenReturn(activeSegment);
when(mockLog.logStartOffset()).thenReturn(oldSegmentStartOffset);
when(mockLog.logSegments(anyLong(), anyLong())).thenReturn(JavaConverters.collectionAsScalaIterable(Arrays.asList(oldSegment, activeSegment)));
when(mockLog.nonActiveLogSegmentsFrom(anyLong()))
.thenReturn(JavaConverters.collectionAsScalaIterable(Arrays.asList(oldSegment)));

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 @@ -349,7 +375,7 @@ void testCopyLogSegmentsToRemoteShouldCopyExpectedLogSegment() throws Exception
assertEquals(remoteLogSegmentMetadataArg.getValue(), remoteLogSegmentMetadataArg2.getValue());
// The old segment should only contain leader epoch [0->0, 1->100] since its offset range is [0, 149]
verifyLogSegmentData(logSegmentDataArg.getValue(), idx, timeIdx, txnIndex, tempFile, mockProducerSnapshotIndex,
Arrays.asList(epochEntry0, epochEntry1));
Arrays.asList(epochEntry0, epochEntry1));

// verify remoteLogMetadataManager did add the expected RemoteLogSegmentMetadataUpdate
ArgumentCaptor<RemoteLogSegmentMetadataUpdate> remoteLogSegmentMetadataUpdateArg = ArgumentCaptor.forClass(RemoteLogSegmentMetadataUpdate.class);
Expand Down Expand Up @@ -480,7 +506,7 @@ void testMetricsUpdateOnCopyLogSegmentsFailure() throws Exception {

when(mockLog.activeSegment()).thenReturn(activeSegment);
when(mockLog.logStartOffset()).thenReturn(oldSegmentStartOffset);
when(mockLog.logSegments(anyLong(), anyLong())).thenReturn(JavaConverters.collectionAsScalaIterable(Arrays.asList(oldSegment, activeSegment)));
when(mockLog.nonActiveLogSegmentsFrom(anyLong())).thenReturn(JavaConverters.collectionAsScalaIterable(Arrays.asList(oldSegment)));

ProducerStateManager mockStateManager = mock(ProducerStateManager.class);
when(mockLog.producerStateManager()).thenReturn(mockStateManager);
Expand Down Expand Up @@ -544,7 +570,7 @@ void testCopyLogSegmentsToRemoteShouldNotCopySegmentForFollower() throws Excepti

when(mockLog.activeSegment()).thenReturn(activeSegment);
when(mockLog.logStartOffset()).thenReturn(oldSegmentStartOffset);
when(mockLog.logSegments(anyLong(), anyLong())).thenReturn(JavaConverters.collectionAsScalaIterable(Arrays.asList(oldSegment, activeSegment)));
when(mockLog.nonActiveLogSegmentsFrom(anyLong())).thenReturn(JavaConverters.collectionAsScalaIterable(Arrays.asList(oldSegment)));
when(mockLog.lastStableOffset()).thenReturn(250L);

RemoteLogManager.RLMTask task = remoteLogManager.new RLMTask(leaderTopicIdPartition);
Expand Down Expand Up @@ -823,6 +849,56 @@ public RemoteLogMetadataManager createRemoteLogMetadataManager() {
}
}

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

when(segment1.baseOffset()).thenReturn(5L);
when(segment2.baseOffset()).thenReturn(10L);
when(segment3.baseOffset()).thenReturn(15L);
when(segment3.readNextOffset()).thenReturn(22L);

when(log.nonActiveLogSegmentsFrom(5L))
.thenReturn(JavaConverters.collectionAsScalaIterable(Arrays.asList(segment1, segment2, segment3)));

RemoteLogManager.RLMTask task = remoteLogManager.new RLMTask(leaderTopicIdPartition);
List<RemoteLogManager.EnrichedLogSegment> expected =
Arrays.asList(
new RemoteLogManager.EnrichedLogSegment(segment1, 10L),
new RemoteLogManager.EnrichedLogSegment(segment2, 15L),
new RemoteLogManager.EnrichedLogSegment(segment3, 22L)
);

assertEquals(expected, task.enrichedLogSegments(log, 5L, 20L));
}

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

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

when(log.nonActiveLogSegmentsFrom(5L))
.thenReturn(JavaConverters.collectionAsScalaIterable(Arrays.asList(segment1, segment2, segment3)));

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

assertEquals(expected, task.enrichedLogSegments(log, 5L, 15L));
}

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