Skip to content

KAFKA-15260: RLM Task should handle uninitialized RLMM for the associated topic-parititon #14113

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
Sep 4, 2023
+58 −1
Merged

KAFKA-15260: RLM Task should handle uninitialized RLMM for the associated topic-parititon #14113

Show file tree
Hide file tree
Changes from 1 commit
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
d92f883
KAFKA-15260: RLM Task should handle uninitialized RLMM
abhijeetk88 Aug 26, 2023
7f83f05
Review comments
abhijeetk88 Sep 1, 2023
e16fc26
address review comments
abhijeetk88 Sep 1, 2023
30d7328
Revert spy changes
abhijeetk88 Sep 3, 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
7 changes: 7 additions & 0 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 @@ -28,6 +28,7 @@
import org.apache.kafka.common.TopicPartition;
import org.apache.kafka.common.Uuid;
import org.apache.kafka.common.errors.OffsetOutOfRangeException;
import org.apache.kafka.common.errors.RetriableException;
import org.apache.kafka.common.message.FetchResponseData;
import org.apache.kafka.common.record.FileRecords;
import org.apache.kafka.common.record.MemoryRecords;
Expand Down Expand Up @@ -673,6 +674,9 @@ public void copyLogSegmentsToRemote(UnifiedLog log) throws InterruptedException
this.cancel();
} catch (InterruptedException ex) {
throw ex;
} catch (RetriableException ex) {
logger.debug("Encountered a retryable error while copying log segments of partition: {}", topicIdPartition, ex);
throw ex;
} catch (Exception ex) {
if (!isCancelled()) {
brokerTopicStats.topicStats(log.topicPartition().topic()).failedRemoteCopyRequestRate().mark();
Expand Down Expand Up @@ -776,6 +780,9 @@ public void run() {
if (!isCancelled()) {
logger.warn("Current thread for topic-partition-id {} is interrupted. Reason: {}", topicIdPartition, ex.getMessage());
}
} catch (RetriableException ex) {
logger.debug("Encountered a retryable error while executing current task for topic-partition {}. " +

@kamalcph kamalcph Sep 1, 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.

Why are we logging it twice in L678 and L784?

@abhijeetk88 abhijeetk88 Sep 1, 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.

removing from the line 678

"Reason: {}", topicIdPartition, ex.getMessage());

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

Let us print the complete exception, which prints stack trace too. That will be helpful while debugging.

} catch (Exception ex) {
if (!isCancelled()) {
logger.warn("Current task for topic-partition {} received error but it will be scheduled. " +
Expand Down
56 changes: 56 additions & 0 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 @@ -28,6 +28,7 @@
import org.apache.kafka.common.TopicPartition;
import org.apache.kafka.common.Uuid;
import org.apache.kafka.common.config.AbstractConfig;
import org.apache.kafka.common.errors.ReplicaNotAvailableException;
import org.apache.kafka.common.network.ListenerName;
import org.apache.kafka.common.record.CompressionType;
import org.apache.kafka.common.record.FileRecords;
Expand Down Expand Up @@ -754,6 +755,61 @@ void testCopyLogSegmentsToRemoteShouldNotCopySegmentForFollower() throws Excepti
verify(mockLog, never()).updateHighestOffsetInRemoteStorage(anyLong());
}

@Test
void testRLMTaskDoesNotUploadSegmentsWhenRemoteLogMetadataManagerIsNotInitialized() throws Exception {
long oldSegmentStartOffset = 0L;
long nextSegmentStartOffset = 150L;

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));

// Throw a retryable exception so indicate that the remote log metadata manager is not initialized yet
when(remoteLogMetadataManager.highestOffsetForEpoch(any(TopicIdPartition.class), anyInt()))
.thenThrow(new ReplicaNotAvailableException("Remote log metadata cache is not initialized for partition: " + leaderTopicIdPartition));

@kamalcph kamalcph Sep 1, 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.

can we have another invocation where it returns the value instead of exception? (eg) once the partition is initialized, the copy segments should work as expected:

 when(remoteLogMetadataManager.highestOffsetForEpoch(any(TopicIdPartition.class), anyInt()))
            .thenThrow(new ReplicaNotAvailableException("Remote log metadata cache is not initialized for partition: " + leaderTopicIdPartition))
           .thenReturn(-1);

@abhijeetk88 abhijeetk88 Sep 1, 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.

There is already a test for this behaviour testCopyLogSegmentsToRemoteShouldCopyExpectedLogSegment


// create 2 log segments, with 0 and 150 as log start offset
LogSegment oldSegment = mock(LogSegment.class);
LogSegment activeSegment = mock(LogSegment.class);

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

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

// Ensure the metrics for remote write requests/failures is zero before attempt to copy log segment
assertEquals(0, brokerTopicStats.topicStats(leaderTopicIdPartition.topic()).remoteCopyRequestRate().count());
assertEquals(0, brokerTopicStats.topicStats(leaderTopicIdPartition.topic()).failedRemoteCopyRequestRate().count());
// Ensure aggregate metrics
assertEquals(0, brokerTopicStats.allTopicsStats().remoteCopyRequestRate().count());
assertEquals(0, brokerTopicStats.allTopicsStats().failedRemoteCopyRequestRate().count());

RemoteLogManager.RLMTask task = remoteLogManager.new RLMTask(leaderTopicIdPartition, 128);
task.convertToLeader(0);

RemoteLogManager.RLMTask spyTask = spy(task);

@kamalcph kamalcph Sep 1, 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.

Why the RLMTask is wrapped with spy? The test doesn't use the spy to verify/mock any behaviour.

@abhijeetk88 abhijeetk88 Sep 1, 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.

Yes, we don't need this anymore. Previously I was verifying a behaviour on the spy but that is removed now.

@abhijeetk88 abhijeetk88 Sep 1, 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.

Added it back to verify a behaviour.

spyTask.run();

// verify the remoteLogMetadataManager never add any metadata and remoteStorageManager never copy log segments
verify(remoteLogMetadataManager, never()).addRemoteLogSegmentMetadata(any(RemoteLogSegmentMetadata.class));
verify(remoteStorageManager, never()).copyLogSegmentData(any(RemoteLogSegmentMetadata.class), any(LogSegmentData.class));
Comment on lines +795 to +799

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

We can't assure if the RLMTask thread is already run once or not when verifying these things under current testing. Could we just invoke task.copyLogSegmentsToRemote and verify it throws exception we expected. And then, we verify the following things.

@abhijeetk88 abhijeetk88 Sep 1, 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.

I added a verification so that copyLogSegmentsToRemote is indeed called.

verify(remoteLogMetadataManager, never()).updateRemoteLogSegmentMetadata(any(RemoteLogSegmentMetadataUpdate.class));
verify(mockLog, never()).updateHighestOffsetInRemoteStorage(anyLong());

// Verify the metric for remote write requests/failures was not updated.
assertEquals(0, brokerTopicStats.topicStats(leaderTopicIdPartition.topic()).remoteCopyRequestRate().count());
assertEquals(0, brokerTopicStats.topicStats(leaderTopicIdPartition.topic()).failedRemoteCopyRequestRate().count());
// Verify aggregate metrics
assertEquals(0, brokerTopicStats.allTopicsStats().remoteCopyRequestRate().count());
assertEquals(0, brokerTopicStats.allTopicsStats().failedRemoteCopyRequestRate().count());
}

private void verifyRemoteLogSegmentMetadata(RemoteLogSegmentMetadata remoteLogSegmentMetadata,
long oldSegmentStartOffset,
long oldSegmentEndOffset,
Expand Down