You are viewing a plain text version of this content. The canonical link for it is here.
Posted to jira@kafka.apache.org by "Avishek Das (Jira)" <ji...@apache.org> on 2026/08/23 15:43:00 UTC
[jira] [Updated] (KAFKA-20977) RemoteCopyLagSegments/RemoteCopyLagBytes report a phantom lag of 1 on low-throughput partitions (single-record segments)
[ https://issues.apache.org/jira/browse/KAFKA-20977?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel ]
Avishek Das updated KAFKA-20977:
--------------------------------
Description:
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.
was:
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/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.
> 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
> 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)