Avishek Das created KAFKA-20977:
-----------------------------------
Summary: 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: 4.3.1, 4.2.1, 4.1.2, 4.0.2, 3.9.2
Reporter: Avishek Das
Assignee: Avishek Das
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 #XXXXX: [PR
link|https://github.com/apache/kafka/blob/trunk/storage/src/main/java/org/apache/kafka/storage/internals/log/UnifiedLog.java#L2109]*
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)
