[
https://issues.apache.org/jira/browse/KAFKA-20977?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Ken Huang updated KAFKA-20977:
------------------------------
Fix Version/s: 4.3.2
> RemoteCopyLagSegments/RemoteCopyLagBytes report a phantom lag of 1 on
> low-throughput partitions (single-record segments)
> -------------------------------------------------------------------------------------------------------------------------
>
> Key: KAFKA-20977
> URL: https://issues.apache.org/jira/browse/KAFKA-20977
> Project: Kafka
> Issue Type: Bug
> Components: Tiered-Storage
> Affects Versions: 3.9.2, 4.0.2, 4.1.2, 4.2.1, 4.3.1
> Reporter: Avishek Das
> Assignee: Avishek Das
> Priority: Minor
> Labels: tiered-storage
> Fix For: 4.3.2, 4.5.0
>
> Attachments: Screenshot 2026-08-23 at 5.23.26 PM.png
>
>
> h2. Summary
> On tiered-storage partitions whose throughput is low enough that a rolled
> segment contains a single record ({{{}baseOffset == endOffset{}}}),
> {{RemoteCopyLagSegments}} gets stuck reporting {{1}} (and
> {{RemoteCopyLagBytes}} reports that segment's size) even though every closed
> segment has already been copied to remote and only the active segment remains
> local. There is no real backlog and no data loss — the copy path is healthy.
> The metric is computed wrong at the segment boundary and then never
> recomputed while the partition is idle.
> This is a {*}residual off-by-one that survives KAFKA-16895{*}.
> KAFKA-16895 removed the _active_ segment from the count (the {{{}- 1{}}});
> this issue is about the _already-copied boundary_ se gment being re-counted
> by a {{>=}} comparison.
> h2. Affects
> 3.9.0 through trunk (reproduced on 4.2). Present wherever
> {{UnifiedLog.onlyLocalLogSegmentsCount()}} uses {*}{{>=}}{*}.
> h2. Root cause
> *(1) [Boundary off-by-one in
> UnifiedLog|https://github.com/apache/kafka/blob/trunk/storage/src/main/java/org/apache/kafka/storage/internals/log/UnifiedLog.java#L2109]*
> {code:java}
> public long onlyLocalLogSegmentsCount() {
> return logSegments().stream()
> .filter(s -> s.baseOffset() >= highestOffsetInRemoteStorage()) //
> should be '>'
> .count();
> }
> {code}
> {{highestOffsetInRemoteStorage()}} holds the *end* offset of the last segment
> already copied to remote (set via
> {{{}updateHighestOffsetInRemoteStorage(endOffset){}}}). A segment is "only
> local / not yet in remote" iff {{{}baseOffset >
> highestOffsetInRemoteStorage{}}}. The {{>=}} wrongly includes the just-copied
> segment *when that segment is single-record* ({{{}baseOffset == endOffset ==
> highestOffsetInRemoteStorage{}}}). {{onlyLocalLogSegmentsSize()}} has the
> same bug.
> {{RemoteLogManager.RLMCopyTask.recordLagStats}} then computes:
> {code:java}
> long segmentsLag = log.onlyLocalLogSegmentsCount() - 1; // '-1' removes
> active seg (KAFKA-16895)
> {code}
> With the boundary segment wrongly counted: count = (copied boundary seg) +
> (active seg) = 2, so segmentsLag = 2 − 1 = 1 instead of 0.
> *(2) The gauge is only refreshed on a copy*
> {{recordLagStats}} runs only after a successful copy or on a copy exception;
> {{resetLagStats()}} (→ 0) runs only on leadership loss. The "no candidate
> segments" branch does not update the gauge. So once a low-throughput
> partition stops rolling new segments, the last (wrong) value stays latched
> indefinitely.
> h2. Worked example
> {code:java}
> # topic: tiered-test (partitions=1, RF=1)
> remote.storage.enable=true
> segment.bytes=<~1 record> # force frequent, eventually single-record rolls
> local.retention.ms=1000 # evict local quickly once copied
> retention.ms=3600000
> # broker: remote.log.manager.task.interval.ms=5000
> {code}
> *Watch:*
> {_}{{kafka.server:type=BrokerTopicMetrics,name=RemoteCopyLagSegments,topic=tiered-test}}{_}.
> *Phase A — multi-record segments (metric CORRECT)*
> ||Segment||base||end||highestRemote after copy||onlyLocalCount||lag
> (count-1)||Result||
> |s0|0|22|22|{active s1 (base 23)}= 1|1-1 = 0|CORRECT|
> |s1|23|24|24|{active s2 (base 25)}= 1|1-1 = 0|CORRECT|
> _*Note:* baseOffset != highestRemote for the copied segment, so ">=" and ">"
> agree._
>
> *Phase B — throughput drops to ONE record per segment (metric BREAKS)*
> ||Segment||base||end||highestRemote after copy||'>=' (buggy) count ->
> lag||'>' (fixed) count -> lag||Correct?||
> |s2|25|25|25|{s2(25>=25, already copied), s3(base 26)}= 2 -> lag 1|{s3(base
> 26)}= 1 -> lag 0|fix = 0|
> |s3|26|26|26|{s3(26>=26), s4(base 27)}= 2 -> lag 1|{s4(base 27)}= 1 -> lag
> 0|fix = 0|
> |producer stops| | | |no roll -> no copy -> recordLagStats never runs ->
> FROZEN at 1|(would be 0)|—|
>
> The instant a segment becomes single-record ({{{}base == end{}}}), its own
> baseOffset equals the new {{{}highestOffsetInRemoteStorage{}}}, so {{>=}}
> keeps counting a segment that is already in remote. After the last write
> there is no roll/copy to recompute it, so it is frozen at 1 even though on
> disk only the empty active segment remains and remote is fully caught up.
> Because the gauge is summed across partitions, every quiet partition
> contributes a phantom 1 and the cluster aggregate steps up monotonically (1
> -> 3 -> ... -> N) and stays there — looking exactly like a growing copy
> backlog when nothing is lagging.
> h2. Steps to reproduce
> # Create a tiered topic sized so rolled segments eventually hold one record.
> # Produce briefly at a rate giving multi-record segments -> confirm
> {{{}RemoteCopyLagSegments == 0{}}}.
> # Drop to <=1 record per rolled segment; let them copy and be locally
> evicted.
> # Stop producing. Observe {{RemoteCopyLagSegments}} stuck at 1 indefinitely
> while there is no uncopied closed segment on disk.
> h2. Expected vs actual
> * *Expected:* 0 once all closed segments are copied.
> * *Actual:* 1 per low-throughput partition, permanently.
> h2. Proposed fix
> Compare strictly — a segment is only-local iff its base offset is beyond the
> highest remote offset:
> {code:java}
> - .filter(s -> s.baseOffset() >= highestOffsetInRemoteStorage())
> + .filter(s -> s.baseOffset() > highestOffsetInRemoteStorage())
> {code}
> in both {{onlyLocalLogSegmentsCount()}} and
> {{{}onlyLocalLogSegmentsSize(){}}}.
> This also removes a latent double-count in the size-based retention total
> ({{{}onlyLocalLogSegmentsSize + remoteLogSizeBytes{}}}), where the boundary
> segment was counted in both terms.
> Optional hardening: also invoke {{recordLagStats(log)}} on the "no candidate
> segments" branch so idle partitions converge to the correct value without
> waiting for the next copy.
> *A fix has been proposed in PR #23246: [PR
> link|https://github.com/apache/kafka/pull/23246]*
> h2. Related issues
> * KAFKA-16895 (fixed 3.9.0) — removed the _active_ segment from the count
> ({{{}- 1{}}}). This issue is the residual boundary off-by-one on top of that
> fix.
> * KAFKA-16948 — reset lag metrics on becoming follower (only reset path).
> * KAFKA-19995 — record lag metrics during copy failures.
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
