KAFKA-19530 RemoteLogManager should record lag stats when remote storage is offline (#20218)

When remote storage is offline, then the segmentLag and bytesLag metrics
are not recorded.   These metrics are useful to know the pending data to
upload when remote storage is down.

Reviewers: TaiJuWu <tjwu1217@gmail.com>, Kamal Chandraprakash
 <kamal.chandraprakash@gmail.com>

Author: m1a2st

storage/src/main/java/org/apache/kafka/server/log/remote/storage/RemoteLogManager.java

@@ -982,6 +982,9 @@ public void copyLogSegmentsToRemote(UnifiedLog log) throws InterruptedException
                             segmentIdsBeingCopied.add(segmentId);
                             try {
                                 copyLogSegment(log, candidateLogSegment.logSegment, segmentId, candidateLogSegment.nextSegmentOffset);
+                            } catch (Exception e) {
+                                recordLagStats(log);
+                                throw e;
                             } finally {
                                 segmentIdsBeingCopied.remove(segmentId);
                             }
@@ -1088,6 +1091,10 @@ private void copyLogSegment(UnifiedLog log, LogSegment segment, RemoteLogSegment
             logger.info("Copied {} to remote storage with segment-id: {}",
                     logFileName, copySegmentFinishedRlsm.remoteLogSegmentId());
 
+            recordLagStats(log);
+        }
+
+        private void recordLagStats(UnifiedLog log) {
             long bytesLag = log.onlyLocalLogSegmentsSize() - log.activeSegment().size();
             long segmentsLag = log.onlyLocalLogSegmentsCount() - 1;
             recordLagStats(bytesLag, segmentsLag);

storage/src/test/java/org/apache/kafka/server/log/remote/storage/RemoteLogManagerTest.java

@@ -691,6 +691,8 @@ void testFailedCopyShouldDeleteTheDanglingSegment() throws Exception {
         long lastStableOffset = 150L;
         long logEndOffset = 150L;
 
+        when(mockLog.onlyLocalLogSegmentsSize()).thenReturn(12L);
+        when(mockLog.onlyLocalLogSegmentsCount()).thenReturn(2L);
         when(mockLog.topicPartition()).thenReturn(leaderTopicIdPartition.topicPartition());
 
         // leader epoch preparation
@@ -708,6 +710,7 @@ void testFailedCopyShouldDeleteTheDanglingSegment() throws Exception {
 
         when(oldSegment.baseOffset()).thenReturn(oldSegmentStartOffset);
         when(activeSegment.baseOffset()).thenReturn(nextSegmentStartOffset);
+        when(activeSegment.size()).thenReturn(2);
         verify(oldSegment, times(0)).readNextOffset();
         verify(activeSegment, times(0)).readNextOffset();
 
@@ -764,6 +767,8 @@ void testFailedCopyShouldDeleteTheDanglingSegment() throws Exception {
         assertEquals(1, brokerTopicStats.allTopicsStats().remoteCopyRequestRate().count());
         assertEquals(0, brokerTopicStats.allTopicsStats().remoteCopyBytesRate().count());
         assertEquals(1, brokerTopicStats.allTopicsStats().failedRemoteCopyRequestRate().count());
+        assertEquals(10, brokerTopicStats.allTopicsStats().remoteCopyLagBytesAggrMetric().value());
+        assertEquals(1, brokerTopicStats.allTopicsStats().remoteCopyLagSegmentsAggrMetric().value());
     }
 
     @Test