You are viewing a plain text version of this content. The canonical link for it is here.
Posted to jira@kafka.apache.org by "abhijeetk88 (via GitHub)" <gi...@apache.org> on 2023/07/27 16:05:55 UTC
[GitHub] [kafka] abhijeetk88 opened a new pull request, #14113: KAFKA-15260: RLM Task should wait for RLMM to initialize
abhijeetk88 opened a new pull request, #14113:
URL: https://github.com/apache/kafka/pull/14113
*More detailed description of your change,
if necessary. The PR title and PR message become
the squashed commit message, so use a separate
comment to ping reviewers.*
*Summary of testing strategy (including rationale)
for the feature or bug fix. Unit and/or integration
tests are expected for any behaviour change and
system tests should be considered for larger changes.*
### Committer Checklist (excluded from commit message)
- [ ] Verify design and implementation
- [ ] Verify test coverage and CI build status
- [ ] Verify documentation (including upgrade notes)
--
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
[GitHub] [kafka] abhijeetk88 commented on a diff in pull request #14113: KAFKA-15260: RLM Task should handle uninitialized RLMM for the associated topic-parititon
Posted by "abhijeetk88 (via GitHub)" <gi...@apache.org>.
abhijeetk88 commented on code in PR #14113:
URL: https://github.com/apache/kafka/pull/14113#discussion_r1312962330
##########
core/src/test/java/kafka/log/remote/RemoteLogManagerTest.java:
##########
@@ -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));
+
+ // 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);
Review Comment:
Added it back to verify a behaviour.
--
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
[GitHub] [kafka] satishd commented on pull request #14113: KAFKA-15260: RLM Task should handle uninitialized RLMM for the associated topic-parititon
Posted by "satishd (via GitHub)" <gi...@apache.org>.
satishd commented on PR #14113:
URL: https://github.com/apache/kafka/pull/14113#issuecomment-1704567178
Merging to trunk and 3.6 as the test failures are not related to this change.
--
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
[GitHub] [kafka] abhijeetk88 commented on a diff in pull request #14113: KAFKA-15260: RLM Task should handle uninitialized RLMM for the associated topic-parititon
Posted by "abhijeetk88 (via GitHub)" <gi...@apache.org>.
abhijeetk88 commented on code in PR #14113:
URL: https://github.com/apache/kafka/pull/14113#discussion_r1312963078
##########
core/src/test/java/kafka/log/remote/RemoteLogManagerTest.java:
##########
@@ -754,6 +755,59 @@ 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));
+
+ // 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);
+ task.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));
Review Comment:
I added a verification so that `copyLogSegmentsToRemote` is indeed called.
--
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
[GitHub] [kafka] satishd commented on a diff in pull request #14113: KAFKA-15260: RLM Task should handle uninitialized RLMM for the associated topic-parititon
Posted by "satishd (via GitHub)" <gi...@apache.org>.
satishd commented on code in PR #14113:
URL: https://github.com/apache/kafka/pull/14113#discussion_r1312784758
##########
core/src/main/java/kafka/log/remote/RemoteLogManager.java:
##########
@@ -776,6 +777,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 {}. " +
+ "Reason: {}", topicIdPartition, ex.getMessage());
Review Comment:
Let us print the complete exception, which prints stack trace too. That will be helpful while debugging.
--
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
[GitHub] [kafka] abhijeetk88 commented on a diff in pull request #14113: KAFKA-15260: RLM Task should handle uninitialized RLMM for the associated topic-parititon
Posted by "abhijeetk88 (via GitHub)" <gi...@apache.org>.
abhijeetk88 commented on code in PR #14113:
URL: https://github.com/apache/kafka/pull/14113#discussion_r1312760241
##########
core/src/test/java/kafka/log/remote/RemoteLogManagerTest.java:
##########
@@ -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));
Review Comment:
There is already a test for this behaviour `testCopyLogSegmentsToRemoteShouldCopyExpectedLogSegment`
--
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
[GitHub] [kafka] abhijeetk88 commented on a diff in pull request #14113: KAFKA-15260: RLM Task should handle uninitialized RLMM for the associated topic-parititon
Posted by "abhijeetk88 (via GitHub)" <gi...@apache.org>.
abhijeetk88 commented on code in PR #14113:
URL: https://github.com/apache/kafka/pull/14113#discussion_r1314290595
##########
core/src/test/java/kafka/log/remote/RemoteLogManagerTest.java:
##########
@@ -754,6 +755,62 @@ 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));
+
+ // 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);
+ spyTask.run();
+
+ verify(spyTask, times(1)).copyLogSegmentsToRemote(mockLog);
Review Comment:
I have reverted the changes, though I think keeping it unchanged would have also been fine.
Thanks for reviewing the changes.
--
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
[GitHub] [kafka] kamalcph commented on a diff in pull request #14113: KAFKA-15260: RLM Task should handle uninitialized RLMM for the associated topic-parititon
Posted by "kamalcph (via GitHub)" <gi...@apache.org>.
kamalcph commented on code in PR #14113:
URL: https://github.com/apache/kafka/pull/14113#discussion_r1314112162
##########
core/src/main/java/kafka/log/remote/RemoteLogManager.java:
##########
@@ -776,6 +777,8 @@ 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 {}", topicIdPartition, ex);
Review Comment:
Does `debug` takes throwable as parameter? What does current task mean to the user?
--
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
[GitHub] [kafka] abhijeetk88 commented on a diff in pull request #14113: KAFKA-15260: RLM Task should handle uninitialized RLMM for the associated topic-parititon
Posted by "abhijeetk88 (via GitHub)" <gi...@apache.org>.
abhijeetk88 commented on code in PR #14113:
URL: https://github.com/apache/kafka/pull/14113#discussion_r1314155819
##########
core/src/main/java/kafka/log/remote/RemoteLogManager.java:
##########
@@ -776,6 +777,8 @@ 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 {}", topicIdPartition, ex);
Review Comment:
It does. Sample logs
```
[2023-09-03 12:59:26,937] DEBUG [RemoteLogManager=0 partition=p6WDWrGtTZyCbzgZG_9XBg:Leader-0] Encountered a retryable error while executing current task for topic-partition p6WDWrGtTZyCbzgZG_9XBg:Leader-0 (kafka.log.remote.RemoteLogManager$RLMTask:781)
org.apache.kafka.common.errors.ReplicaNotAvailableException: Remote log metadata cache is not initialized for partition: p6WDWrGtTZyCbzgZG_9XBg:Leader-0
```
I kept the debug message in line with the warning message at L784.
--
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
[GitHub] [kafka] kamalcph commented on a diff in pull request #14113: KAFKA-15260: RLM Task should handle uninitialized RLMM for the associated topic-parititon
Posted by "kamalcph (via GitHub)" <gi...@apache.org>.
kamalcph commented on code in PR #14113:
URL: https://github.com/apache/kafka/pull/14113#discussion_r1312514688
##########
core/src/test/java/kafka/log/remote/RemoteLogManagerTest.java:
##########
@@ -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));
+
+ // 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);
Review Comment:
Why the RLMTask is wrapped with `spy`? The test doesn't use the `spy` to verify/mock any behaviour.
##########
core/src/test/java/kafka/log/remote/RemoteLogManagerTest.java:
##########
@@ -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));
Review Comment:
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:
```java
when(remoteLogMetadataManager.highestOffsetForEpoch(any(TopicIdPartition.class), anyInt()))
.thenThrow(new ReplicaNotAvailableException("Remote log metadata cache is not initialized for partition: " + leaderTopicIdPartition))
.thenReturn(-1);
```
##########
core/src/main/java/kafka/log/remote/RemoteLogManager.java:
##########
@@ -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 {}. " +
Review Comment:
Why are we logging it twice in L678 and L784?
--
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
[GitHub] [kafka] abhijeetk88 commented on a diff in pull request #14113: KAFKA-15260: RLM Task should handle uninitialized RLMM for the associated topic-parititon
Posted by "abhijeetk88 (via GitHub)" <gi...@apache.org>.
abhijeetk88 commented on code in PR #14113:
URL: https://github.com/apache/kafka/pull/14113#discussion_r1312761621
##########
core/src/main/java/kafka/log/remote/RemoteLogManager.java:
##########
@@ -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 {}. " +
Review Comment:
removing from the line 678
--
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
[GitHub] [kafka] showuon commented on a diff in pull request #14113: KAFKA-15260: RLM Task should handle uninitialized RLMM for the associated topic-parititon
Posted by "showuon (via GitHub)" <gi...@apache.org>.
showuon commented on code in PR #14113:
URL: https://github.com/apache/kafka/pull/14113#discussion_r1314161952
##########
core/src/test/java/kafka/log/remote/RemoteLogManagerTest.java:
##########
@@ -754,6 +755,62 @@ 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));
+
+ // 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);
+ spyTask.run();
+
+ verify(spyTask, times(1)).copyLogSegmentsToRemote(mockLog);
Review Comment:
Ah, I was wrong. You're not starting the task with `start`. You call `run` directly. So, everything is fine! Your original version is good. That means, we don't need to `spy(task);`, and no need to `verify(spyTask, times(1)).copyLogSegmentsToRemote(mockLog);`, either.
Please revert it and I don't have any other comment. Sorry and thanks.
--
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
[GitHub] [kafka] satishd commented on pull request #14113: KAFKA-15260: RLM Task should handle uninitialized RLMM for the associated topic-parititon
Posted by "satishd (via GitHub)" <gi...@apache.org>.
satishd commented on PR #14113:
URL: https://github.com/apache/kafka/pull/14113#issuecomment-1704007735
@kamalcph @showuon Please look into the replies from @abhijeetk88 on your respective comments.
--
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
[GitHub] [kafka] abhijeetk88 commented on a diff in pull request #14113: KAFKA-15260: RLM Task should handle uninitialized RLMM for the associated topic-parititon
Posted by "abhijeetk88 (via GitHub)" <gi...@apache.org>.
abhijeetk88 commented on code in PR #14113:
URL: https://github.com/apache/kafka/pull/14113#discussion_r1312761175
##########
core/src/test/java/kafka/log/remote/RemoteLogManagerTest.java:
##########
@@ -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));
+
+ // 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);
Review Comment:
Yes, we don't need this anymore. Previously I was verifying a behaviour on the spy but that is removed now.
--
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
[GitHub] [kafka] showuon commented on a diff in pull request #14113: KAFKA-15260: RLM Task should handle uninitialized RLMM for the associated topic-parititon
Posted by "showuon (via GitHub)" <gi...@apache.org>.
showuon commented on code in PR #14113:
URL: https://github.com/apache/kafka/pull/14113#discussion_r1312968348
##########
core/src/test/java/kafka/log/remote/RemoteLogManagerTest.java:
##########
@@ -754,6 +755,62 @@ 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));
+
+ // 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);
+ spyTask.run();
+
+ verify(spyTask, times(1)).copyLogSegmentsToRemote(mockLog);
Review Comment:
@abhijeetk88 , although we added this line to verify it is called, it will make this test unstable since sometimes it'll fail here because the RLMTask haven't invoked `copyLogSegmentsToRemote`, right? You can either use `waitUntil`, or call `RLMTask#copyLogSegmentsToRemote` directly to fix that.
--
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
[GitHub] [kafka] showuon commented on a diff in pull request #14113: KAFKA-15260: RLM Task should handle uninitialized RLMM for the associated topic-parititon
Posted by "showuon (via GitHub)" <gi...@apache.org>.
showuon commented on code in PR #14113:
URL: https://github.com/apache/kafka/pull/14113#discussion_r1312804527
##########
core/src/test/java/kafka/log/remote/RemoteLogManagerTest.java:
##########
@@ -754,6 +755,59 @@ 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));
+
+ // 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);
+ task.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));
Review Comment:
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.
--
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
[GitHub] [kafka] abhijeetk88 commented on a diff in pull request #14113: KAFKA-15260: RLM Task should handle uninitialized RLMM for the associated topic-parititon
Posted by "abhijeetk88 (via GitHub)" <gi...@apache.org>.
abhijeetk88 commented on code in PR #14113:
URL: https://github.com/apache/kafka/pull/14113#discussion_r1313011389
##########
core/src/test/java/kafka/log/remote/RemoteLogManagerTest.java:
##########
@@ -754,6 +755,62 @@ 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));
+
+ // 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);
+ spyTask.run();
+
+ verify(spyTask, times(1)).copyLogSegmentsToRemote(mockLog);
Review Comment:
Curious to understand why copyLogSegmentsToRemote may not get invoked?
We are directly calling the run method here.
--
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
[GitHub] [kafka] abhijeetk88 commented on a diff in pull request #14113: KAFKA-15260: RLM Task should handle uninitialized RLMM for the associated topic-parititon
Posted by "abhijeetk88 (via GitHub)" <gi...@apache.org>.
abhijeetk88 commented on code in PR #14113:
URL: https://github.com/apache/kafka/pull/14113#discussion_r1313011389
##########
core/src/test/java/kafka/log/remote/RemoteLogManagerTest.java:
##########
@@ -754,6 +755,62 @@ 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));
+
+ // 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);
+ spyTask.run();
+
+ verify(spyTask, times(1)).copyLogSegmentsToRemote(mockLog);
Review Comment:
Curious to understand why copyLogSegmentsToRemote may not get invoked?
--
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
[GitHub] [kafka] kamalcph commented on a diff in pull request #14113: KAFKA-15260: RLM Task should handle uninitialized RLMM for the associated topic-parititon
Posted by "kamalcph (via GitHub)" <gi...@apache.org>.
kamalcph commented on code in PR #14113:
URL: https://github.com/apache/kafka/pull/14113#discussion_r1314112162
##########
core/src/main/java/kafka/log/remote/RemoteLogManager.java:
##########
@@ -776,6 +777,8 @@ 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 {}", topicIdPartition, ex);
Review Comment:
Does `debug` takes throwable as parameter?
--
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
[GitHub] [kafka] showuon commented on pull request #14113: KAFKA-15260: RLM Task should handle uninitialized RLMM for the associated topic-parititon
Posted by "showuon (via GitHub)" <gi...@apache.org>.
showuon commented on PR #14113:
URL: https://github.com/apache/kafka/pull/14113#issuecomment-1704496025
See if @satishd has any other comments.
--
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
[GitHub] [kafka] satishd merged pull request #14113: KAFKA-15260: RLM Task should handle uninitialized RLMM for the associated topic-parititon
Posted by "satishd (via GitHub)" <gi...@apache.org>.
satishd merged PR #14113:
URL: https://github.com/apache/kafka/pull/14113
--
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