Skip to content

KAFKA-15241: Compute tiered copied offset by keeping the respective epochs in scope #14787

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 4 commits into from
Nov 27, 2023
Merged

KAFKA-15241: Compute tiered copied offset by keeping the respective epochs in scope #14787

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
c248de0
KAFKA-15241: Compute tiered copied offset by keeping the respective e…
kamalcph Nov 17, 2023
a487519
Use OffsetAndEpoch class.
kamalcph Nov 20, 2023
90a69e5
improve the logger statement
kamalcph Nov 20, 2023
c103f19
Addressed the review comments
kamalcph Nov 22, 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
62 changes: 42 additions & 20 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 @@ -41,6 +41,7 @@
import org.apache.kafka.common.utils.LogContext;
import org.apache.kafka.common.utils.Time;
import org.apache.kafka.common.utils.Utils;
import org.apache.kafka.server.common.OffsetAndEpoch;
import org.apache.kafka.server.log.remote.metadata.storage.ClassLoaderAwareRemoteLogMetadataManager;
import org.apache.kafka.server.log.remote.storage.ClassLoaderAwareRemoteStorageManager;
import org.apache.kafka.server.log.remote.storage.LogSegmentData;
Expand Down Expand Up @@ -590,7 +591,7 @@ boolean isLeader() {

// The copied and log-start offset is empty initially for a new leader RLMTask, and needs to be fetched inside
// the task's run() method.
private volatile OptionalLong copiedOffsetOption = OptionalLong.empty();
private volatile Optional<OffsetAndEpoch> copiedOffsetOption = Optional.empty();
private volatile boolean isLogStartOffsetUpdatedOnBecomingLeader = false;

public void convertToLeader(int leaderEpochVal) {
Expand All @@ -601,7 +602,7 @@ public void convertToLeader(int leaderEpochVal) {
leaderEpoch = leaderEpochVal;
}
// Reset copied and log-start offset, so that it is set in next run of RLMTask
copiedOffsetOption = OptionalLong.empty();
copiedOffsetOption = Optional.empty();
isLogStartOffsetUpdatedOnBecomingLeader = false;
}

Expand All @@ -625,9 +626,10 @@ private void maybeUpdateCopiedOffset(UnifiedLog log) throws RemoteStorageExcepti
// of a segment with that epoch copied into remote storage. If it can not find an entry then it checks for the
// previous leader epoch till it finds an entry, If there are no entries till the earliest leader epoch in leader
// epoch cache then it starts copying the segments from the earliest epoch entry's offset.
copiedOffsetOption = OptionalLong.of(findHighestRemoteOffset(topicIdPartition, log));
copiedOffsetOption = Optional.of(findHighestRemoteOffset(topicIdPartition, log));
logger.info("Found the highest copiedRemoteOffset: {} for partition: {} after becoming leader, " +
"leaderEpoch: {}", copiedOffsetOption, topicIdPartition, leaderEpoch);
copiedOffsetOption.ifPresent(offsetAndEpoch -> log.updateHighestOffsetInRemoteStorage(offsetAndEpoch.offset()));
}
}

Expand Down Expand Up @@ -664,7 +666,7 @@ public void copyLogSegmentsToRemote(UnifiedLog log) throws InterruptedException
try {
maybeUpdateLogStartOffsetOnBecomingLeader(log);
maybeUpdateCopiedOffset(log);
long copiedOffset = copiedOffsetOption.getAsLong();
long copiedOffset = copiedOffsetOption.get().offset();

// LSO indicates the offset below are ready to be consumed (high-watermark or committed)
long lso = log.lastStableOffset();
Expand Down Expand Up @@ -767,7 +769,11 @@ private void copyLogSegment(UnifiedLog log, LogSegment segment, long nextSegment
brokerTopicStats.topicStats(log.topicPartition().topic())
.remoteCopyBytesRate().mark(copySegmentStartedRlsm.segmentSizeInBytes());
brokerTopicStats.allTopicsStats().remoteCopyBytesRate().mark(copySegmentStartedRlsm.segmentSizeInBytes());
copiedOffsetOption = OptionalLong.of(endOffset);

// `epochEntries` cannot be empty, there is a pre-condition validation in RemoteLogSegmentMetadata
// constructor
int lastEpochInSegment = epochEntries.get(epochEntries.size() - 1).epoch;
copiedOffsetOption = Optional.of(new OffsetAndEpoch(endOffset, lastEpochInSegment));
// Update the highest offset in remote storage for this partition's log so that the local log segments
// are not deleted before they are copied to remote storage.
log.updateHighestOffsetInRemoteStorage(endOffset);
Expand Down Expand Up @@ -796,10 +802,10 @@ public void run() {
// Cleanup/delete expired remote log segments
cleanupExpiredRemoteLogSegments();
} else {
long offset = findHighestRemoteOffset(topicIdPartition, log);
OffsetAndEpoch offsetAndEpoch = findHighestRemoteOffset(topicIdPartition, log);
// Update the highest offset in remote storage for this partition's log so that the local log segments
// are not deleted before they are copied to remote storage.
log.updateHighestOffsetInRemoteStorage(offset);
log.updateHighestOffsetInRemoteStorage(offsetAndEpoch.offset());
}
} catch (InterruptedException ex) {
if (!isCancelled()) {
Expand Down Expand Up @@ -1422,20 +1428,37 @@ RecordBatch findFirstBatch(RemoteLogInputStream remoteLogInputStream, long offse
return nextBatch;
}

long findHighestRemoteOffset(TopicIdPartition topicIdPartition, UnifiedLog log) throws RemoteStorageException {
Optional<Long> offset = Optional.empty();

Option<LeaderEpochFileCache> maybeLeaderEpochFileCache = log.leaderEpochCache();
if (maybeLeaderEpochFileCache.isDefined()) {
LeaderEpochFileCache cache = maybeLeaderEpochFileCache.get();
OptionalInt epoch = cache.latestEpoch();
while (!offset.isPresent() && epoch.isPresent()) {
offset = remoteLogMetadataManager.highestOffsetForEpoch(topicIdPartition, epoch.getAsInt());
epoch = cache.previousEpoch(epoch.getAsInt());
OffsetAndEpoch findHighestRemoteOffset(TopicIdPartition topicIdPartition, UnifiedLog log) throws RemoteStorageException {
OffsetAndEpoch offsetAndEpoch = null;
Option<LeaderEpochFileCache> leaderEpochCacheOpt = log.leaderEpochCache();
if (leaderEpochCacheOpt.isDefined()) {
LeaderEpochFileCache cache = leaderEpochCacheOpt.get();
Optional<EpochEntry> maybeEpochEntry = cache.latestEntry();
while (offsetAndEpoch == null && maybeEpochEntry.isPresent()) {
int epoch = maybeEpochEntry.get().epoch;
Optional<Long> highestRemoteOffsetOpt =
remoteLogMetadataManager.highestOffsetForEpoch(topicIdPartition, epoch);
if (highestRemoteOffsetOpt.isPresent()) {
Map.Entry<Integer, Long> entry = cache.endOffsetFor(epoch, log.logEndOffset());
int requestedEpoch = entry.getKey();
long endOffset = entry.getValue();
long highestRemoteOffset = highestRemoteOffsetOpt.get();
if (endOffset <= highestRemoteOffset) {
LOGGER.info("The end-offset for epoch {}: ({}, {}) is less than or equal to the " +
"highest-remote-offset: {} for partition: {}", epoch, requestedEpoch, endOffset,
highestRemoteOffset, topicIdPartition);
offsetAndEpoch = new OffsetAndEpoch(endOffset - 1, requestedEpoch);
} else {
offsetAndEpoch = new OffsetAndEpoch(highestRemoteOffset, epoch);
}
}
maybeEpochEntry = cache.previousEntry(epoch);
}
}

return offset.orElse(-1L);
if (offsetAndEpoch == null) {
offsetAndEpoch = new OffsetAndEpoch(-1L, RecordBatch.NO_PARTITION_LEADER_EPOCH);
}
return offsetAndEpoch;
}

long findLogStartOffset(TopicIdPartition topicIdPartition, UnifiedLog log) throws RemoteStorageException {
Expand Down Expand Up @@ -1667,5 +1690,4 @@ public String toString() {
'}';
}
}

}
62 changes: 54 additions & 8 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 @@ -40,6 +40,7 @@
import org.apache.kafka.common.security.auth.SecurityProtocol;
import org.apache.kafka.common.utils.MockTime;
import org.apache.kafka.common.utils.Time;
import org.apache.kafka.server.common.OffsetAndEpoch;
import org.apache.kafka.server.log.remote.storage.ClassLoaderAwareRemoteStorageManager;
import org.apache.kafka.server.log.remote.storage.LogSegmentData;
import org.apache.kafka.server.log.remote.storage.NoOpRemoteLogMetadataManager;
Expand Down Expand Up @@ -237,18 +238,63 @@ void testGetLeaderEpochCheckpoint() {
assertEquals(epochEntry1, epochEntries.get(0));
}

@Test
void testFindHighestRemoteOffsetOnEmptyRemoteStorage() throws RemoteStorageException {
List<EpochEntry> totalEpochEntries = Arrays.asList(
new EpochEntry(0, 0),
new EpochEntry(1, 500)
);
checkpoint.write(totalEpochEntries);
LeaderEpochFileCache cache = new LeaderEpochFileCache(tp, checkpoint);
when(mockLog.leaderEpochCache()).thenReturn(Option.apply(cache));
TopicIdPartition tpId = new TopicIdPartition(Uuid.randomUuid(), tp);
OffsetAndEpoch offsetAndEpoch = remoteLogManager.findHighestRemoteOffset(tpId, mockLog);
assertEquals(new OffsetAndEpoch(-1L, -1), offsetAndEpoch);
}

@Test
void testFindHighestRemoteOffset() throws RemoteStorageException {
List<EpochEntry> totalEpochEntries = Arrays.asList(
new EpochEntry(0, 0),
new EpochEntry(1, 500)
);
checkpoint.write(totalEpochEntries);
LeaderEpochFileCache cache = new LeaderEpochFileCache(tp, checkpoint);
when(mockLog.leaderEpochCache()).thenReturn(Option.apply(cache));
TopicIdPartition tpId = new TopicIdPartition(Uuid.randomUuid(), tp);
long offset = remoteLogManager.findHighestRemoteOffset(tpId, mockLog);
assertEquals(-1, offset);
when(remoteLogMetadataManager.highestOffsetForEpoch(eq(tpId), anyInt())).thenAnswer(ans -> {
Integer epoch = ans.getArgument(1, Integer.class);
if (epoch == 0) {
return Optional.of(200L);
} else {
return Optional.empty();
}
});
OffsetAndEpoch offsetAndEpoch = remoteLogManager.findHighestRemoteOffset(tpId, mockLog);
assertEquals(new OffsetAndEpoch(200L, 0), offsetAndEpoch);
}

when(remoteLogMetadataManager.highestOffsetForEpoch(tpId, 2)).thenReturn(Optional.of(200L));
long offset2 = remoteLogManager.findHighestRemoteOffset(tpId, mockLog);
assertEquals(200, offset2);
@Test
void testFindHighestRemoteOffsetWithUncleanLeaderElection() throws RemoteStorageException {
List<EpochEntry> totalEpochEntries = Arrays.asList(
new EpochEntry(0, 0),
new EpochEntry(1, 150),
new EpochEntry(2, 300)
);
checkpoint.write(totalEpochEntries);
LeaderEpochFileCache cache = new LeaderEpochFileCache(tp, checkpoint);
when(mockLog.leaderEpochCache()).thenReturn(Option.apply(cache));
TopicIdPartition tpId = new TopicIdPartition(Uuid.randomUuid(), tp);
when(remoteLogMetadataManager.highestOffsetForEpoch(eq(tpId), anyInt())).thenAnswer(ans -> {
Integer epoch = ans.getArgument(1, Integer.class);
if (epoch == 0) {
return Optional.of(200L);
} else {
return Optional.empty();
}
});
OffsetAndEpoch offsetAndEpoch = remoteLogManager.findHighestRemoteOffset(tpId, mockLog);
assertEquals(new OffsetAndEpoch(149L, 0), offsetAndEpoch);
}

@Test
Expand Down Expand Up @@ -484,7 +530,7 @@ private void assertCopyExpectedLogSegmentsToRemote(long oldSegmentStartOffset,

// verify the highest remote offset is updated to the expected value
ArgumentCaptor<Long> argument = ArgumentCaptor.forClass(Long.class);
verify(mockLog, times(1)).updateHighestOffsetInRemoteStorage(argument.capture());
verify(mockLog, times(2)).updateHighestOffsetInRemoteStorage(argument.capture());
assertEquals(oldSegmentEndOffset, argument.getValue());

// Verify the metric for remote writes is updated correctly
Expand Down Expand Up @@ -735,7 +781,7 @@ void testMetricsUpdateOnCopyLogSegmentsFailure() throws Exception {
verify(remoteStorageManager, times(1)).copyLogSegmentData(any(RemoteLogSegmentMetadata.class), any(LogSegmentData.class));

// Verify we should not have updated the highest offset because of write failure
verify(mockLog, times(0)).updateHighestOffsetInRemoteStorage(anyLong());
verify(mockLog).updateHighestOffsetInRemoteStorage(anyLong());
// Verify the metric for remote write requests/failures was updated.
assertEquals(1, brokerTopicStats.topicStats(leaderTopicIdPartition.topic()).remoteCopyRequestRate().count());
assertEquals(1, brokerTopicStats.topicStats(leaderTopicIdPartition.topic()).failedRemoteCopyRequestRate().count());
Expand Down Expand Up @@ -775,7 +821,7 @@ void testCopyLogSegmentsToRemoteShouldNotCopySegmentForFollower() throws Excepti
verify(remoteLogMetadataManager, never()).addRemoteLogSegmentMetadata(any(RemoteLogSegmentMetadata.class));
verify(remoteStorageManager, never()).copyLogSegmentData(any(RemoteLogSegmentMetadata.class), any(LogSegmentData.class));
verify(remoteLogMetadataManager, never()).updateRemoteLogSegmentMetadata(any(RemoteLogSegmentMetadataUpdate.class));
verify(mockLog, never()).updateHighestOffsetInRemoteStorage(anyLong());
verify(mockLog).updateHighestOffsetInRemoteStorage(anyLong());
}

@Test
Expand Down
19 changes: 18 additions & 1 deletion core/src/test/scala/unit/kafka/server/epoch/LeaderEpochFileCacheTest.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -27,7 +27,7 @@ import org.junit.jupiter.api.Assertions._
import org.junit.jupiter.api.Test

import java.io.File
import java.util.{Collections, OptionalInt}
import java.util.{Collections, OptionalInt, Optional}
import scala.collection.Seq
import scala.jdk.CollectionConverters._

Expand Down Expand Up @@ -604,6 +604,23 @@ class LeaderEpochFileCacheTest {
assertEquals(OptionalInt.of(2), cache.previousEpoch(cache.latestEpoch.getAsInt))
}

@Test
def testFindPreviousEntry(): Unit = {
assertEquals(Optional.empty(), cache.previousEntry(2))

cache.assign(2, 10)
assertEquals(Optional.empty(), cache.previousEntry(2))

cache.assign(4, 15)
assertEquals(Optional.of(new EpochEntry(2, 10)), cache.previousEntry(4))

cache.assign(10, 20)
assertEquals(Optional.of(new EpochEntry(4, 15)), cache.previousEntry(10))

cache.truncateFromEnd(18)
assertEquals(Optional.of(new EpochEntry(2, 10)), cache.previousEntry(cache.latestEpoch.getAsInt))
}

@Test
def testFindNextEpoch(): Unit = {
cache.assign(0, 0)
Expand Down
7 changes: 6 additions & 1 deletion ...g/apache/kafka/server/log/remote/metadata/storage/TopicBasedRemoteLogMetadataManager.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -455,7 +455,12 @@ boolean doesTopicExist(Admin adminClient, String topic) {
.topicNameValues()
.get(topic)
.get();
log.info("Topic {} exists. Description: {}", topic, description);
if (description != null) {
log.info("Topic {} exists. TopicId: {}, numPartitions: {}, ", topic,
description.topicId(), description.partitions().size());
} else {
log.info("Topic {} does not exist.", topic);
}
return description != null;
} catch (ExecutionException | InterruptedException ex) {
log.info("Topic {} does not exist. Error: {}", topic, ex.getCause().getMessage());
Expand Down
9 changes: 9 additions & 0 deletions storage/src/main/java/org/apache/kafka/storage/internals/epoch/LeaderEpochFileCache.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -215,6 +215,15 @@ public OptionalInt previousEpoch(int epoch) {
}
}

public Optional<EpochEntry> previousEntry(int epoch) {
lock.readLock().lock();
try {
return Optional.ofNullable(epochs.lowerEntry(epoch)).map(Map.Entry::getValue);
} finally {
lock.readLock().unlock();
}
}

public OptionalInt nextEpoch(int epoch) {
lock.readLock().lock();
try {
Expand Down