(kafka) branch 4.4 updated: KAFKA-20977: Fix RemoteCopyLagSegments off-by-one on single-record segments (#23246)

[ch...@apache.org]

This is an automated email from the ASF dual-hosted git repository.

chia7712 pushed a commit to branch 4.4
in repository https://gitbox.apache.org/repos/asf/kafka.git


The following commit(s) were added to refs/heads/4.4 by this push:
     new a46fdf2b79a KAFKA-20977: Fix RemoteCopyLagSegments off-by-one on single-record segments (#23246)
a46fdf2b79a is described below

commit a46fdf2b79a4d20887bc8e7d248a28f51f328f92
Author: Avishek Das <av...@gmail.com>
AuthorDate: Fri Aug 28 09:15:00 2026 +0530

    KAFKA-20977: Fix RemoteCopyLagSegments off-by-one on single-record segments (#23246)
    
    ### JIRA
    https://issues.apache.org/jira/browse/KAFKA-20977
    
    ### Problem
    `UnifiedLog.onlyLocalLogSegmentsCount()` / `onlyLocalLogSegmentsSize()`
    filtered segments with
    `baseOffset() >= highestOffsetInRemoteStorage()`.
    `highestOffsetInRemoteStorage` holds the **end**
    offset of the last segment already copied to remote, so a segment whose
    base-offset equals that
    value has itself already been copied. This happens for **single-record
    segments**
    (`baseOffset == endOffset == highestOffsetInRemoteStorage`), which are
    common on low-throughput
    tiered-storage partitions. The `>=` comparison wrongly counted that
    already-copied boundary
    segment as "only local".
    
    `RemoteLogManager.RLMCopyTask.recordLagStats` derives
    `RemoteCopyLagSegments` as
    `onlyLocalLogSegmentsCount() - 1` (the `- 1` removes the active segment,
    per KAFKA-16895) and only
    refreshes the gauge after a copy. So on a partition that has gone idle
    the metric stays latched at a
    **phantom lag of 1** even though every closed segment is already in
    remote storage and only the
    active segment remains local. `RemoteCopyLagBytes` and the size-based
    retention total
    (`onlyLocalLogSegmentsSize + remoteLogSizeBytes`) are inflated the same
    way. Summed across
    partitions this looks like a growing copy backlog when nothing is
    actually lagging.
    
    This is a residual off-by-one on top of KAFKA-16895: that fix removed
    the *active* segment from the count; this one removes the
    *already-copied boundary* segment.
    
    ### Change
    Use a strict `>` in both `onlyLocalLogSegmentsCount()` and
    `onlyLocalLogSegmentsSize()` — a segment
    is "only local" iff its base-offset is strictly greater than
    `highestOffsetInRemoteStorage()`.
    Javadoc added explaining the boundary condition.
    
    ### Tests
    Added two `UnifiedLogTest` cases:
    - `onlyLocalLogSegmentsExcludeCopiedSingleRecordBoundarySegment` —
    single-record segments; asserts
      the count/lag is 0 once all closed segments are copied. **Fails on the
    previous `>=`**
      (`expected: <2> but was: <3>`), passes with the fix.
    - `onlyLocalLogSegmentsWithMultiRecordSegmentsIsUnaffectedByFix` —
    multi-record segments; guards the
      common case where `>=` and `>` agree so the fix does not change
    correct behavior.
    
    Verified locally:
    ```
    ./gradlew :storage:test --tests "*UnifiedLogTest.onlyLocalLogSegments*"
    ```
    Both pass; `checkstyleMain`, `checkstyleTest`, and `spotbugsMain` are
    clean.
    
    Reviewers: Kamal Chandraprakash <ka...@gmail.com>
    
    Co-authored-by: imavishek <im...@users.noreply.github.com>
---
 .../kafka/storage/internals/log/UnifiedLog.java    | 15 ++++++-
 .../storage/internals/log/UnifiedLogTest.java      | 50 ++++++++++++++++++++++
 2 files changed, 63 insertions(+), 2 deletions(-)

diff --git a/storage/src/main/java/org/apache/kafka/storage/internals/log/UnifiedLog.java b/storage/src/main/java/org/apache/kafka/storage/internals/log/UnifiedLog.java
index affa31076ec..0d4a2d65625 100644
--- a/storage/src/main/java/org/apache/kafka/storage/internals/log/UnifiedLog.java
+++ b/storage/src/main/java/org/apache/kafka/storage/internals/log/UnifiedLog.java
@@ -2104,16 +2104,27 @@ public class UnifiedLog implements AutoCloseable {
 
     /**
      * The log size in bytes for all segments that are only in local log but not yet in remote log.
+     *
+     * <p>A segment is considered "only local" (not yet in remote) when its base-offset is strictly
+     * greater than {@link #highestOffsetInRemoteStorage()}. The strict {@code >} (rather than
+     * {@code >=}) matters: {@code highestOffsetInRemoteStorage} holds the end-offset of the last
+     * segment already copied to remote, so the segment whose base-offset equals that value has
+     * itself been copied. This arises for single-record segments (base-offset == end-offset), which
+     * are common on low-throughput partitions; counting such a segment as local would double-count a
+     * segment that is already in remote storage.
      */
     public long onlyLocalLogSegmentsSize() {
-        return LogSegments.sizeInBytes(logSegments().stream().filter(s -> s.baseOffset() >= highestOffsetInRemoteStorage()).collect(Collectors.toList()));
+        return LogSegments.sizeInBytes(logSegments().stream().filter(s -> s.baseOffset() > highestOffsetInRemoteStorage()).collect(Collectors.toList()));
     }
 
     /**
      * The number of segments that are only in local log but not yet in remote log.
+     *
+     * <p>See {@link #onlyLocalLogSegmentsSize()} for why the base-offset comparison is a strict
+     * {@code >} against {@link #highestOffsetInRemoteStorage()} rather than {@code >=}.
      */
     public long onlyLocalLogSegmentsCount() {
-        return logSegments().stream().filter(s -> s.baseOffset() >= highestOffsetInRemoteStorage()).count();
+        return logSegments().stream().filter(s -> s.baseOffset() > highestOffsetInRemoteStorage()).count();
     }
 
     /**
diff --git a/storage/src/test/java/org/apache/kafka/storage/internals/log/UnifiedLogTest.java b/storage/src/test/java/org/apache/kafka/storage/internals/log/UnifiedLogTest.java
index 48953d133a5..4fa95365843 100644
--- a/storage/src/test/java/org/apache/kafka/storage/internals/log/UnifiedLogTest.java
+++ b/storage/src/test/java/org/apache/kafka/storage/internals/log/UnifiedLogTest.java
@@ -467,6 +467,56 @@ public class UnifiedLogTest {
         assertTrue(deletedSegments > 0, "At least one segment should be deleted");
     }
 
+    @Test
+    public void onlyLocalLogSegmentsExcludeCopiedBoundarySegment() throws IOException {
+        Supplier<MemoryRecords> records = () -> singletonRecords("test".getBytes(), "test".getBytes(), mockTime.milliseconds());
+        LogConfig config = new LogTestUtils.LogConfigBuilder()
+                .remoteLogStorageEnable(true)
+                .build();
+        log = createLog(logDir, config, true);
+
+        // Build single-record segments: closed segments at base-offsets 0, 1 and 2 (each with
+        // end-offset == base-offset), plus an empty active segment at base-offset 3.
+        for (int i = 0; i < 3; i++) {
+            log.appendAsLeader(records.get(), 0);
+            log.roll();
+        }
+        assertEquals(3L, log.logEndOffset());
+        assertEquals(4, log.numberOfSegments());
+
+        // Nothing copied yet (highestOffsetInRemoteStorage == -1): every segment is local-only.
+        assertEquals(4L, log.onlyLocalLogSegmentsCount());
+
+        // Segments up to offset 1 copied. The closed single-record segment at base-offset 2 is
+        // genuinely uncopied, so it is counted alongside the active segment => lag 1.
+        log.updateHighestOffsetInRemoteStorage(1L);
+        assertEquals(2L, log.onlyLocalLogSegmentsCount());
+
+        // Every closed segment is now copied; highestOffsetInRemoteStorage == base-offset of the last
+        // (single-record) segment. Only the empty active segment remains local-only => lag must be 0.
+        // With the previous ">=" comparison the copied boundary segment was still counted, giving 1.
+        log.updateHighestOffsetInRemoteStorage(2L);
+        assertEquals(1L, log.onlyLocalLogSegmentsCount());
+        assertEquals(log.activeSegment().size(), log.onlyLocalLogSegmentsSize());
+
+        // Two closed segments of two records each plus an empty active segment
+        for (int seg = 0; seg < 2; seg++) {
+            log.appendAsLeader(records.get(), 0);
+            log.appendAsLeader(records.get(), 0);
+            log.roll();
+        }
+        assertEquals(7L, log.logEndOffset());
+        // segments 0 - 2 contain 1 record each and seg 3 - 4 contain 2 records each, and the active segment is empty.
+        assertEquals(6, log.numberOfSegments());
+
+        log.updateHighestOffsetInRemoteStorage(4L);
+        assertEquals(2L, log.onlyLocalLogSegmentsCount());
+
+        // Both closed segments copied: only the active segment is local-only => lag 0.
+        log.updateHighestOffsetInRemoteStorage(6L);
+        assertEquals(1L, log.onlyLocalLogSegmentsCount());
+    }
+
     @Test
     public void shouldDeleteLocalLogSegmentsWhenPolicyIsEmptyWithMsRetention() throws IOException {
         long oldTimestamp = mockTime.milliseconds() - 20000;